diff --git a/CLAUDE.md b/CLAUDE.md index d0104b46..ff0dd661 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -72,8 +72,12 @@ separate **cortex-viz** MCP (reads this same store read-only). ## Code Style -- 300 lines max per file; 40 lines max per method. Enforced by the - craftsmanship-checker pre-commit hook. +- 300 lines max per file; 40 lines max per method — a local tightening of + coding-standards.md §4.1/§4.2 (≤500/≤50; CONTRIBUTING.md § Code Style + cites the same 300/40 numbers). Enforced by code review today; no + automated pre-commit hook checks this yet (issue #276 corrected the + prior claim of a "craftsmanship-checker" hook — none exists in + `.git/hooks/` or a `.pre-commit-config.yaml`). - Import rule: `core/` imports only `shared/` + stdlib; `infrastructure/` never imports core/handlers. Verify: `grep -rn "from mcp_server.infrastructure" mcp_server/core/` should return nothing for new code — 3 pre-existing violations in diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 34790baf..4a2e4348 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -152,7 +152,9 @@ project-specific rules: the 568-diagnostic ratchet backlog was burned to zero in issue #197 (2026-07-28) and the ratchet retired. History: [`docs/provenance/pyright-remediation-plan.md`](docs/provenance/pyright-remediation-plan.md). -- **§4.1 File ≤500 lines, §4.2 function ≤50 lines.** +- **File ≤300 lines, function ≤40 lines** — this repo's local tightening of + coding-standards.md §4.1/§4.2 (≤500/≤50); see CLAUDE.md § Code Style for + the authoritative numbers (issue #276). The full standard lives in [zetetic coding standards](https://github.com/cdeust/zetetic-team-subagents/blob/main/rules/coding-standards.md). diff --git a/mcp_server/handlers/consolidation/anchor_authoring.py b/mcp_server/handlers/consolidation/anchor_authoring.py new file mode 100644 index 00000000..1a6231a3 --- /dev/null +++ b/mcp_server/handlers/consolidation/anchor_authoring.py @@ -0,0 +1,164 @@ +"""Anchor-page authoring for the headless authoring worker. + +A project missing its architecture / services / api / ci-cd / mcp / +ai-usage / prd / decisions anchor page has no gap marker to drain — the +page simply doesn't exist. This module detects missing anchors via the +coverage audit, feeds Claude a project-level overview (file tree, +README, key config files, source file counts), and asks it to author +the anchor from scratch. Split out of ``drain_operations`` (Fowler: +Move Function, issue #276) to keep that module under the size limit; +the public import surface stays ``headless_authoring``, which +re-exports ``drain_missing_anchors``. + +Import-cycle note (issue #237 family): a module-top ``from . import +headless_authoring as _root`` would deadlock a fresh interpreter +importing this module before ``headless_authoring`` finishes (it +imports ``drain_missing_anchors`` back at load time). ``_root`` is +resolved lazily at call time instead — every +``monkeypatch.setattr(headless_authoring, ...)`` stays observed. +""" + +from __future__ import annotations + +import time +from collections.abc import Awaitable, Callable +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from .drain_operations import _drain_result, _elapsed_ms +from .page_io import _scope_anchor_prompt, _write_anchor_page +from mcp_server.core.wiki_coverage import _project_source_root, audit_domain +from mcp_server.shared.domain_mapping import _build_registry + + +def _filled_count(results: list[Any]) -> int: + """Count how many DrainResults in ``results`` have status='filled'.""" + return len([r for r in results if r.status == "filled"]) + + +async def _invoke_anchor_fill( + domain: str, + sc: Any, + src_root: str, + invoke: Callable[..., Awaitable[Any]], + _root: Any, +) -> tuple[Any, int]: + """Build the anchor prompt, invoke claude, return (ir, elapsed_ms).""" + t0 = time.monotonic() + prompt = _scope_anchor_prompt( + domain=domain, + scope_name=sc.scope.name, + scope_title=sc.scope.title, + scope_description=sc.scope.description, + source_root=src_root, + delegate_hint=_root._delegation_hint_for(sc.scope.suggested_kind), + ) + ir = await invoke(prompt, cwd=src_root, source_root=src_root) + return ir, _elapsed_ms(t0) + + +async def _author_one_anchor( + domain: str, + sc: Any, + src_root: str, + wiki_root: Path, + today: str, + invoke: Callable[..., Awaitable[Any]], + _root: Any, +) -> Any: + """Invoke claude for one missing anchor scope and write the result page.""" + gap = f"anchor:{sc.scope.name}" + ir, ms = await _invoke_anchor_fill(domain, sc, src_root, invoke, _root) + response = ir.text + if not response or response.strip() == "": + return _drain_result( + sc.suggested_path, gap, "failed", ms, "claude returned empty", _root + ) + written = await _write_anchor_page( + wiki_root=wiki_root, + domain=domain, + scope_name=sc.scope.name, + suggested_kind=sc.scope.suggested_kind, + suggested_path=sc.suggested_path, + body_markdown=response.strip(), + today=today, + ) + return _drain_result( + str(written) if written else sc.suggested_path, + gap, + "filled" if written else "failed", + ms, + "" if written else "page write failed", + _root, + ) + + +async def _drain_domain_anchors( + domain: str, + wiki_root: Path, + max_drains: int, + today: str, + invoke: Callable[..., Awaitable[Any]], + results: list[Any], + _root: Any, +) -> None: + """Author missing anchors for one domain, appending to ``results`` in place.""" + src_root = _project_source_root(domain) + if not src_root: + return + cov = audit_domain(str(wiki_root), domain) + for sc in cov.scopes: + if sc.covered: + continue + # Groundable-only filter: content that can't be derived from the + # source tree alone (prd, decisions, changelog, roadmap, ...) is + # skipped entirely rather than fabricated (zetetic-forbidden). + if not sc.scope.groundable: + continue + if _filled_count(results) >= max_drains: + break + results.append( + await _author_one_anchor( + domain, sc, src_root, wiki_root, today, invoke, _root + ) + ) + + +async def drain_missing_anchors( + wiki_root: Path, + *, + max_drains: int = 30, + today: str | None = None, + invoke: Callable[..., Awaitable[Any]] | None = None, +) -> list[Any]: + """Author missing canonical anchor pages for every project. + + For each domain x scope combination with no covered anchor, calls + ``claude -p`` with a project-level context block and writes the + response as the new anchor page. Up to ``max_drains`` authored per + invocation so a single cycle stays time-bounded. Ungroundable + scopes are omitted (not skipped-with-result) — they remain visible + as coverage gaps for human authors. + + Pre: ``wiki_root`` is a valid directory; ``invoke`` matches + ``_claude_invoke``'s signature. + Post: up to ``max_drains`` new anchor pages written to disk. + """ + # Deferred import (issue #237): see module docstring's import-cycle note. + from . import headless_authoring as _root # noqa: PLC0415 — import cycle (partner: headless_authoring, #237) + + if invoke is None: + invoke = _root._claude_invoke + + today = today or datetime.now(timezone.utc).date().isoformat() + domains = sorted({r.canonical for r in _build_registry().repos}) + + results: list[Any] = [] + for domain in domains: + if _filled_count(results) >= max_drains: + break + await _drain_domain_anchors( + domain, wiki_root, max_drains, today, invoke, results, _root + ) + return results diff --git a/mcp_server/handlers/consolidation/candidate_scan.py b/mcp_server/handlers/consolidation/candidate_scan.py index 5a79e5ab..88d6a89e 100644 --- a/mcp_server/handlers/consolidation/candidate_scan.py +++ b/mcp_server/handlers/consolidation/candidate_scan.py @@ -17,67 +17,119 @@ from pathlib import Path from typing import Any -from . import headless_authoring as _root from .page_io import _parse_frontmatter from mcp_server.observability import silent_failure from mcp_server.core.wiki_coverage import _project_source_root, audit_domain from mcp_server.shared.domain_mapping import _build_registry -def _scan_pages_with_gaps(wiki_root: Path) -> list[tuple[Path, dict[str, Any], str]]: - """Walk the wiki and return ``(path, meta, body)`` for pages with gaps. +def _load_missing_sections_probe() -> Any: + """Lazy-import the optional live-audit probe. - A page is "with gaps" when EITHER the frontmatter declares - ``curation_gaps`` non-empty OR a live audit of the body shows - missing canonical sections. The second axis catches pages that - were complete under the old section catalogue but are incomplete - after the catalogue gained new sections (e.g. sequence-diagram, - parameters, request-example, response-example added 2026-05-18). - - Only pages classified as kind=reference / explanation file-docs - are audited live; ADRs / specs / guides have their own section - sets and shouldn't be force-fed file-doc sections. + Returns None (a handled degraded mode) when the optional module is + unavailable, so callers skip the live-audit axis entirely. """ - if not wiki_root.is_dir(): - return [] - # Lazy import to keep this module self-contained. try: from mcp_server.core.wiki_curation_gaps import missing_sections # noqa: PLC0415 — optional-feature probe: ImportError here is a handled degraded mode + + return missing_sections except ImportError: - missing_sections = None # type: ignore[assignment] + return None + + +def _gap_entry_for_page( + md: Path, wiki_root: Path, missing_sections: Any +) -> tuple[Path, dict[str, Any], str] | None: + """Return ``(path, meta, body)`` for ``md`` if it has curation gaps, else None. + + A page is "with gaps" when EITHER the frontmatter declares + ``curation_gaps`` non-empty OR a live audit of the body shows + missing canonical sections (only for kind=reference file-docs — + ADRs / specs / guides have their own section sets). + """ + rel = md.relative_to(wiki_root) + if any(part.startswith((".", "_")) for part in rel.parts): + return None + try: + text = md.read_text(encoding="utf-8", errors="ignore") + except OSError: + return None + meta, body, _ = _parse_frontmatter(text) + gaps = meta.get("curation_gaps") + if isinstance(gaps, list) and gaps: + return (md, meta, body) + # No frozen gaps — but a file-doc might still be missing sections + # that were added to the catalogue after generation. + if ( + missing_sections is not None + and meta.get("kind") == "reference" + and meta.get("source_file_path") + ): + try: + live = missing_sections(body) + except Exception as exc: # noqa: BLE001 — mechanism boundary; failure is observable via silent_failure + silent_failure.note("candidate_scan.live_audit", exc) + live = [] + if live: + return (md, meta, body) + return None + +def _scan_pages_with_gaps(wiki_root: Path) -> list[tuple[Path, dict[str, Any], str]]: + """Walk the wiki and return ``(path, meta, body)`` for pages with gaps. + + See ``_gap_entry_for_page`` for the per-page criteria (frozen + frontmatter gaps OR a live section audit for file-docs). + """ + if not wiki_root.is_dir(): + return [] + missing_sections = _load_missing_sections_probe() out: list[tuple[Path, dict[str, Any], str]] = [] for md in wiki_root.rglob("*.md"): - rel = md.relative_to(wiki_root) - if any(part.startswith((".", "_")) for part in rel.parts): - continue - try: - text = md.read_text(encoding="utf-8", errors="ignore") - except OSError: - continue - meta, body, _ = _parse_frontmatter(text) - gaps = meta.get("curation_gaps") - if isinstance(gaps, list) and gaps: - out.append((md, meta, body)) - continue - # No frozen gaps — but a file-doc might still be missing - # sections that were added to the catalogue after generation. - # Only force-audit file-docs (kind=reference + has source_file_path). - if ( - missing_sections is not None - and meta.get("kind") == "reference" - and meta.get("source_file_path") - ): - try: - live = missing_sections(body) - except Exception as exc: # noqa: BLE001 — mechanism boundary; failure is observable via silent_failure - silent_failure.note("candidate_scan.live_audit", exc) - live = [] - if live: - out.append((md, meta, body)) + entry = _gap_entry_for_page(md, wiki_root, missing_sections) + if entry is not None: + out.append(entry) return out +def _extend_anchor_candidates_for_domain( + domain: str, + wiki_root: Path, + max_drains: int, + candidates: list[Any], + _root: Any, +) -> None: + """Append ``domain``'s missing groundable anchor candidates in place. + + Stops appending once ``candidates`` reaches ``max_drains`` (checked + after every append, matching the original inline loop's early exit). + """ + src_root = _project_source_root(domain) + if not src_root: + return + try: + cov = audit_domain(str(wiki_root), domain) + except Exception as exc: # noqa: BLE001 — mechanism boundary; failure is observable via silent_failure + silent_failure.note("candidate_scan.audit_domain", exc) + return + for sc in cov.scopes: + if sc.covered or not sc.scope.groundable: + continue + candidates.append( + _root._AnchorCandidate( + domain=domain, + scope_name=sc.scope.name, + scope_title=sc.scope.title, + scope_description=sc.scope.description, + source_root=src_root, + suggested_path=sc.suggested_path, + suggested_kind=sc.scope.suggested_kind, + ) + ) + if len(candidates) >= max_drains: + return + + def _collect_anchor_candidates( wiki_root: Path, max_drains: int, @@ -89,6 +141,15 @@ def _collect_anchor_candidates( representing a missing scope that passes the groundable filter and has a resolvable source root. """ + # Deferred import (issue #237): headless_authoring imports this function + # back at load time, so a module-top-level `from . import + # headless_authoring` here would deadlock a fresh interpreter that + # imports candidate_scan before headless_authoring finishes initializing. + # Resolved at call time instead — the constructed type still matches the + # re-exported ``headless_authoring._AnchorCandidate`` exactly (same + # module object, no copy). + from . import headless_authoring as _root # noqa: PLC0415 — import cycle (partner: headless_authoring, #237) + try: domains = sorted({r.canonical for r in _build_registry().repos}) except Exception as exc: # noqa: BLE001 — mechanism boundary; failure is observable via silent_failure @@ -97,30 +158,9 @@ def _collect_anchor_candidates( candidates: list[Any] = [] for domain in domains: - src_root = _project_source_root(domain) - if not src_root: - continue - try: - cov = audit_domain(str(wiki_root), domain) - except Exception as exc: # noqa: BLE001 — mechanism boundary; failure is observable via silent_failure - silent_failure.note("candidate_scan.audit_domain", exc) - continue - for sc in cov.scopes: - if sc.covered: - continue - if not sc.scope.groundable: - continue - candidates.append( - _root._AnchorCandidate( - domain=domain, - scope_name=sc.scope.name, - scope_title=sc.scope.title, - scope_description=sc.scope.description, - source_root=src_root, - suggested_path=sc.suggested_path, - suggested_kind=sc.scope.suggested_kind, - ) - ) - if len(candidates) >= max_drains: - return candidates + if len(candidates) >= max_drains: + break + _extend_anchor_candidates_for_domain( + domain, wiki_root, max_drains, candidates, _root + ) return candidates diff --git a/mcp_server/handlers/consolidation/claude_cli.py b/mcp_server/handlers/consolidation/claude_cli.py index d4afee8a..d17f2ef1 100644 --- a/mcp_server/handlers/consolidation/claude_cli.py +++ b/mcp_server/handlers/consolidation/claude_cli.py @@ -84,8 +84,6 @@ import os -from . import headless_authoring as _root - # Anthropic credential keys stripped from the child env in subscription mode. _API_CREDENTIAL_KEYS = ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN") @@ -133,6 +131,16 @@ def _build_argv(source_root: str | None) -> list[str]: Post-condition: returns the option-only argv; the tool surface is read-only in both modes (no Write/Edit/Bash). """ + # Deferred import (issue #237): headless_authoring re-exports this + # function and imports back from this module at load time, so a + # module-top-level `from . import headless_authoring` would deadlock + # any fresh interpreter that imports claude_cli before headless_authoring + # has finished initializing (partial-module ImportError). Resolving the + # back-reference here, at call time, breaks the load-time cycle while + # keeping `monkeypatch.setattr(headless_authoring, "CORTEX_HEADLESS_AGENTS", ...)` + # observed — the attribute is read off the live module object, not a copy. + from . import headless_authoring as _root # noqa: PLC0415 — import cycle (partner: headless_authoring, #237) + argv = [_root._CLAUDE_BIN, "--print", "--no-session-persistence"] if _root.CORTEX_HEADLESS_AGENTS: # Agents mode: roster + Task delegation under a hard write/exec ceiling. diff --git a/mcp_server/handlers/consolidation/claude_invoke.py b/mcp_server/handlers/consolidation/claude_invoke.py new file mode 100644 index 00000000..1c8094c0 --- /dev/null +++ b/mcp_server/handlers/consolidation/claude_invoke.py @@ -0,0 +1,160 @@ +"""``claude -p`` subprocess invocation for the headless authoring worker. + +Split out of ``headless_authoring`` (Fowler: Move Function, issue #276) +to keep that module under the size limit. The public import surface +stays ``headless_authoring``, which re-exports ``_claude_invoke``. + +Import-cycle note (issue #237 family): ``headless_authoring`` defines +``InvokeResult``/``CLAUDE_CALL_TIMEOUT_SEC`` and imports +``_claude_invoke`` back at its own module top, so a module-top ``from . +import headless_authoring as _root`` here would deadlock a fresh +interpreter importing ``claude_invoke`` first. ``_root`` is resolved +lazily inside ``_claude_invoke`` instead, exactly like the other four +siblings in this package. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from typing import Any + +from .claude_cli import _build_argv, _subprocess_env + +logger = logging.getLogger(__name__) + + +async def _spawn_claude_process( + argv: list[str], cwd: str | None, child_env: dict[str, str] +) -> Any: + """Start the claude subprocess; return None (logged) on spawn failure.""" + try: + return await asyncio.create_subprocess_exec( + *argv, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + env=child_env, + ) + except FileNotFoundError: + logger.warning("headless-authoring: claude binary not found on PATH") + return None + except Exception as exc: # noqa: BLE001 — last-resort boundary — failure is logged; degraded mode continues + logger.warning("headless-authoring: failed to start claude subprocess: %s", exc) + return None + + +async def _communicate_with_timeout( + proc: Any, prompt: str, call_timeout: float +) -> tuple[bytes, bytes] | None: + """Send the prompt and await the response, killing the process on + timeout/failure/cancellation. Returns None (logged) on no response. + """ + try: + return await asyncio.wait_for( + proc.communicate(input=prompt.encode("utf-8")), timeout=call_timeout + ) + except asyncio.TimeoutError: + logger.warning( + "headless-authoring: claude -p timed out after %.0fs", call_timeout + ) + return None + except Exception as exc: # noqa: BLE001 — last-resort boundary — failure is logged; degraded mode continues + logger.warning("headless-authoring: claude -p communicate failed: %s", exc) + return None + finally: + # CancelledError is a BaseException — it escapes the except clauses + # above. Without this finally, a cancelled drain leaves a zombie + # subprocess. Covers the timeout path too (returncode is None after + # wait_for cancels communicate), so the kill lives in one place. + if proc.returncode is None: + try: + proc.kill() + except ProcessLookupError: + pass + await proc.wait() + + +def _parse_invoke_response( + returncode: int, stdout_bytes: bytes, stderr_bytes: bytes, _root: Any +) -> Any: + """Decode the subprocess output and parse ``--output-format json``. + + Documented fields (source: code.claude.com/docs/en/headless): + ``result`` (str, assistant text), ``total_cost_usd`` (float, client-side + cost estimate). ``usage``/``is_error`` are NOT guaranteed — returncode only. + """ + stdout = stdout_bytes.decode("utf-8", errors="replace") if stdout_bytes else "" + stderr = stderr_bytes.decode("utf-8", errors="replace") if stderr_bytes else "" + + if returncode != 0: + logger.warning( + "headless-authoring: claude -p exit %d stderr=%r", returncode, stderr[:300] + ) + return _root.InvokeResult(text=None, cost_usd=0.0) + + stdout = stdout.strip() + if not stdout: + return _root.InvokeResult(text=None, cost_usd=0.0) + + try: + data = json.loads(stdout) + text: str | None = data.get("result") or None + cost_usd = float(data.get("total_cost_usd") or 0.0) + except (json.JSONDecodeError, ValueError): + # Defensive: returncode==0 but JSON parse failed. Can happen if + # --output-format json isn't supported by an older claude CLI build. + # Treat raw stdout as the text to degrade gracefully rather than + # losing a successful response. + logger.debug( + "headless-authoring: JSON parse failed (returncode=0); " + "treating raw stdout as text (cost unknown)" + ) + text = stdout or None + cost_usd = 0.0 + + return _root.InvokeResult(text=text, cost_usd=cost_usd) + + +async def _claude_invoke( + prompt: str, + *, + cwd: str | None = None, + source_root: str | None = None, + timeout: float | None = None, +) -> Any: + """Run ``claude -p`` asynchronously and return an InvokeResult. + + Non-blocking on the event loop; on timeout the subprocess is killed + and an empty InvokeResult is returned. The argv/child environment — + including the audit-B-1 security argument and auth mode — are built + by ``claude_cli``. The prompt is fed via STDIN, not a positional argv + element: the variadic ``--add-dir`` would otherwise swallow a + trailing prompt (see ``claude_cli._build_argv``). + """ + # Deferred import (issue #237 family): see module docstring's note. + from . import headless_authoring as _root # noqa: PLC0415 — import cycle (partner: headless_authoring, #237) + + argv = _build_argv(source_root) + # Stays inline: _root's concrete type here is what lets pyright resolve + # CLAUDE_CALL_TIMEOUT_SEC's real type (see cycle_orchestration.py's + # commit note on the same narrowing behavior). + call_timeout = ( + timeout if timeout is not None else float(_root.CLAUDE_CALL_TIMEOUT_SEC) + ) + # Subscription by default + hook-neutralising child flag; API key passes + # through only on CORTEX_HEADLESS_AUTH=api opt-in. See claude_cli. + child_env = _subprocess_env() + + proc = await _spawn_claude_process(argv, cwd, child_env) + if proc is None: + return _root.InvokeResult(text=None, cost_usd=0.0) + + comm = await _communicate_with_timeout(proc, prompt, call_timeout) + if comm is None: + return _root.InvokeResult(text=None, cost_usd=0.0) + stdout_bytes, stderr_bytes = comm + + return _parse_invoke_response(proc.returncode, stdout_bytes, stderr_bytes, _root) diff --git a/mcp_server/handlers/consolidation/cycle_orchestration.py b/mcp_server/handlers/consolidation/cycle_orchestration.py index 23ed060e..5402948b 100644 --- a/mcp_server/handlers/consolidation/cycle_orchestration.py +++ b/mcp_server/handlers/consolidation/cycle_orchestration.py @@ -10,22 +10,24 @@ exhausted candidates return status="skipped" immediately. 4. cost_usd from InvokeResult charges the budget after each call. -Split out of ``headless_authoring`` to keep that module under the size -limit (Fowler: Move Function). The public import surface remains -``headless_authoring``; ``run_headless_authoring_cycle`` is re-exported -there. - -Patchability contract (the reason this module reads ``_root.X`` instead -of importing the names): the throttle tests do -``monkeypatch.setattr(ha, "CORTEX_HEADLESS_CONCURRENCY", 2)``, -``monkeypatch.setattr(ha, "_collect_anchor_candidates", ...)``, and -``monkeypatch.setattr(ha, "_scan_pages_with_gaps", ...)`` then call -``run_headless_authoring_cycle``. For those patches to be observed, -this function MUST resolve those names at CALL TIME from the -``headless_authoring`` module namespace. The ``from . import -headless_authoring as _root`` below is a deliberate circular import -that works ONLY because we touch ``_root.X`` at call time, never at -import time. Do not change ``_root.X`` accesses to direct imports. +Split out of ``headless_authoring`` (Fowler: Move Function); the public +import surface remains ``headless_authoring``, which re-exports +``run_headless_authoring_cycle``. + +Patchability contract: the throttle tests do +``monkeypatch.setattr(ha, "CORTEX_HEADLESS_CONCURRENCY", 2)`` (and +similarly for ``_collect_anchor_candidates`` / ``_scan_pages_with_gaps``) +then call ``run_headless_authoring_cycle`` — every name below MUST +resolve off the live ``headless_authoring`` module at CALL time for +those patches to be observed. + +Import-cycle note (issue #237): a module-top ``from . import +headless_authoring as _root`` would deadlock a fresh interpreter (it +imports ``run_headless_authoring_cycle`` back at load time). ``_root`` +is bound once, in ``run_headless_authoring_cycle``, and passed +explicitly to every helper below (Extract Function, issue #276 — the +two coroutines that used to close over it as a local now take it as a +parameter) — every ``_root.X`` access still resolves at call time. """ from __future__ import annotations @@ -36,162 +38,169 @@ from pathlib import Path from typing import Any -from . import headless_authoring as _root -from .drain_operations import drain_all_gaps_on_page +from .drain_operations import _drain_result, _elapsed_ms, drain_all_gaps_on_page from .page_io import _scope_anchor_prompt, _write_anchor_page from datetime import datetime, timezone from mcp_server.infrastructure.config import WIKI_ROOT -async def run_headless_authoring_cycle( - wiki_root: Path | None = None, - *, - max_drains: int = _root.CORTEX_HEADLESS_MAX_FILE_DRAINS, - max_anchor_drains: int = _root.CORTEX_HEADLESS_MAX_ANCHOR_DRAINS, - invoke: Callable[..., Awaitable[Any]] = _root._claude_invoke, -) -> Any: - """One autonomous cycle: author missing anchor pages, then drain - file-doc curation gaps. Runs concurrently under a shared semaphore - and a per-cycle budget (wall-clock + USD). - - Anchor pages come first — a project missing its - architecture/services/api page is more visibly incomplete than a - single file-doc with a missing "Callers" section. - - Pre-condition: ``invoke`` is an async callable with the signature of - ``_claude_invoke``. ``wiki_root`` resolves to a valid - wiki directory. - Post-condition: CycleSummary reflects all outcomes including budget - telemetry (usd_spent, wall_clock_ms, skipped_budget). - Invariant: no more than CORTEX_HEADLESS_CONCURRENCY in-flight - subprocess calls at any point in the cycle. - """ - - cycle_start = time.monotonic() - if wiki_root is None: - wiki_root = Path(WIKI_ROOT) - - today = datetime.now(timezone.utc).date().isoformat() - - budget = _root.CycleBudget( - deadline=time.monotonic() + _root.CORTEX_HEADLESS_BUDGET_SEC, - usd_cap=_root.CORTEX_HEADLESS_USD_BUDGET, - ) - sem = asyncio.Semaphore(_root.CORTEX_HEADLESS_CONCURRENCY) +def _build_cycle_candidates( + wiki_root: Path, max_drains: int, max_anchor_drains: int, _root: Any +) -> tuple[list[Any], list[tuple[Path, dict[str, Any], str]], list[Any]]: + """Phase 1: build the full candidate list (no claude calls yet). - # ── Phase 1: build full candidate list (no claude calls) ────────── + Returns ``(anchor_cands, file_cands, file_cands_raw)`` — the last is + the unsliced scan result, kept for the cycle summary's pages_scanned. + """ anchor_cands = _root._collect_anchor_candidates(wiki_root, max_anchor_drains) file_cands_raw = _root._scan_pages_with_gaps(wiki_root) file_cands_raw.sort( key=lambda c: (-(len(c[1].get("curation_gaps") or [])), str(c[0])) ) file_cands = file_cands_raw[:max_drains] + return anchor_cands, file_cands, file_cands_raw - # ── Phase 2: coroutine factories ────────────────────────────────── - - async def drain_anchor_bounded(cand: Any) -> Any: - """Drain one anchor candidate under semaphore + budget control.""" - async with sem: - if budget.exhausted(): - return _root.DrainResult( - page_path=cand.suggested_path, - gap=f"anchor:{cand.scope_name}", - status="skipped", - duration_ms=0, - detail="budget exhausted", - ) - t0 = time.monotonic() - prompt = _scope_anchor_prompt( - domain=cand.domain, - scope_name=cand.scope_name, - scope_title=cand.scope_title, - scope_description=cand.scope_description, - source_root=cand.source_root, - ) - # Effective per-call timeout = remaining budget time, capped at - # CLAUDE_CALL_TIMEOUT_SEC and floored at 1 s. - eff_timeout = max( - 1.0, min(float(_root.CLAUDE_CALL_TIMEOUT_SEC), budget.time_left()) - ) - ir = await invoke( - prompt, - cwd=cand.source_root, - source_root=cand.source_root, - timeout=eff_timeout, - ) - budget.charge(ir.cost_usd) - ms = int((time.monotonic() - t0) * 1000) - if not ir.text or ir.text.strip() == "": - return _root.DrainResult( - page_path=cand.suggested_path, - gap=f"anchor:{cand.scope_name}", - status="failed", - duration_ms=ms, - detail="claude returned empty", - ) - written = await _write_anchor_page( - wiki_root=wiki_root, # type: ignore[arg-type] - domain=cand.domain, - scope_name=cand.scope_name, - suggested_kind=cand.suggested_kind, - suggested_path=cand.suggested_path, - body_markdown=ir.text.strip(), - today=today, + +async def _invoke_anchor_call( + cand: Any, invoke: Callable[..., Awaitable[Any]], budget: Any, _root: Any +) -> tuple[Any, int]: + """Build the anchor prompt, invoke claude under a budget-aware timeout, + charge the budget, and return (ir, elapsed_ms). + """ + t0 = time.monotonic() + prompt = _scope_anchor_prompt( + domain=cand.domain, + scope_name=cand.scope_name, + scope_title=cand.scope_title, + scope_description=cand.scope_description, + source_root=cand.source_root, + ) + # Effective per-call timeout = remaining budget time, capped at + # CLAUDE_CALL_TIMEOUT_SEC and floored at 1 s. + eff_timeout = max( + 1.0, min(float(_root.CLAUDE_CALL_TIMEOUT_SEC), budget.time_left()) + ) + ir = await invoke( + prompt, cwd=cand.source_root, source_root=cand.source_root, timeout=eff_timeout + ) + budget.charge(ir.cost_usd) + return ir, _elapsed_ms(t0) + + +async def _drain_anchor_bounded( + cand: Any, + wiki_root: Path, + today: str, + invoke: Callable[..., Awaitable[Any]], + sem: asyncio.Semaphore, + budget: Any, + _root: Any, +) -> Any: + """Drain one anchor candidate under semaphore + budget control.""" + gap = f"anchor:{cand.scope_name}" + async with sem: + if budget.exhausted(): + return _drain_result( + cand.suggested_path, gap, "skipped", 0, "budget exhausted", _root ) - return _root.DrainResult( - page_path=str(written) if written else cand.suggested_path, - gap=f"anchor:{cand.scope_name}", - status="filled" if written else "failed", - duration_ms=ms, - detail="" if written else "page write failed", + ir, ms = await _invoke_anchor_call(cand, invoke, budget, _root) + if not ir.text or ir.text.strip() == "": + return _drain_result( + cand.suggested_path, gap, "failed", ms, "claude returned empty", _root ) + written = await _write_anchor_page( + wiki_root=wiki_root, + domain=cand.domain, + scope_name=cand.scope_name, + suggested_kind=cand.suggested_kind, + suggested_path=cand.suggested_path, + body_markdown=ir.text.strip(), + today=today, + ) + return _drain_result( + str(written) if written else cand.suggested_path, + gap, + "filled" if written else "failed", + ms, + "" if written else "page write failed", + _root, + ) - async def drain_page_bounded( - page_path: Path, meta: dict[str, Any], body: str - ) -> list[Any]: - """Drain all gaps on one file page under semaphore + budget control.""" - async with sem: - if budget.exhausted(): - return [ - _root.DrainResult( - page_path=str(page_path), - gap="all", - status="skipped", - duration_ms=0, - detail="budget exhausted", - ) - ] - - # Wrap invoke to auto-charge the budget after each call and - # inject the effective timeout derived from remaining budget time. - async def charging_invoke(prompt: str, **kw: Any) -> Any: - kw.setdefault( - "timeout", - max( - 1.0, - min(float(_root.CLAUDE_CALL_TIMEOUT_SEC), budget.time_left()), - ), - ) - ir = await invoke(prompt, **kw) - budget.charge(ir.cost_usd) - return ir - - return await drain_all_gaps_on_page( - page_path, meta, body, wiki_root=wiki_root, invoke=charging_invoke - ) - # ── Phase 3: gather all candidates concurrently ─────────────────── - anchor_coros = [drain_anchor_bounded(c) for c in anchor_cands] - page_coros = [drain_page_bounded(p, m, b) for p, m, b in file_cands] +def _make_charging_invoke( + invoke: Callable[..., Awaitable[Any]], budget: Any, _root: Any +) -> Callable[..., Awaitable[Any]]: + """Wrap ``invoke`` to auto-charge the budget and inject the effective timeout.""" - anchor_results: list[Any] = ( - list(await asyncio.gather(*anchor_coros)) if anchor_coros else [] - ) - page_results_nested: list[list[Any]] = ( - list(await asyncio.gather(*page_coros)) if page_coros else [] - ) - file_results: list[Any] = [r for nested in page_results_nested for r in nested] + async def charging_invoke(prompt: str, **kw: Any) -> Any: + kw.setdefault( + "timeout", + max(1.0, min(float(_root.CLAUDE_CALL_TIMEOUT_SEC), budget.time_left())), + ) + ir = await invoke(prompt, **kw) + budget.charge(ir.cost_usd) + return ir + return charging_invoke + + +async def _drain_page_bounded( + page_path: Path, + meta: dict[str, Any], + body: str, + wiki_root: Path, + invoke: Callable[..., Awaitable[Any]], + sem: asyncio.Semaphore, + budget: Any, + _root: Any, +) -> list[Any]: + """Drain all gaps on one file page under semaphore + budget control.""" + async with sem: + if budget.exhausted(): + return [ + _drain_result(page_path, "all", "skipped", 0, "budget exhausted", _root) + ] + charging_invoke = _make_charging_invoke(invoke, budget, _root) + return await drain_all_gaps_on_page( + page_path, meta, body, wiki_root=wiki_root, invoke=charging_invoke + ) + + +async def _gather_cycle_results( + anchor_cands: list[Any], + file_cands: list[tuple[Path, dict[str, Any], str]], + wiki_root: Path, + today: str, + invoke: Callable[..., Awaitable[Any]], + sem: asyncio.Semaphore, + budget: Any, + _root: Any, +) -> tuple[list[Any], list[Any]]: + """Phase 3: scatter every candidate concurrently under sem + budget.""" + anchor_coros = [ + _drain_anchor_bounded(c, wiki_root, today, invoke, sem, budget, _root) + for c in anchor_cands + ] + page_coros = [ + _drain_page_bounded(p, m, b, wiki_root, invoke, sem, budget, _root) + for p, m, b in file_cands + ] + anchor_results = list(await asyncio.gather(*anchor_coros)) if anchor_coros else [] + page_results_nested = list(await asyncio.gather(*page_coros)) if page_coros else [] + file_results = [r for nested in page_results_nested for r in nested] + return anchor_results, file_results + + +def _summarize_cycle( + cycle_start: float, + file_cands_raw: list[Any], + file_cands: list[Any], + anchor_results: list[Any], + file_results: list[Any], + budget: Any, + _root: Any, +) -> Any: + """Phase 4: aggregate anchor + file results into a CycleSummary.""" all_results = anchor_results + file_results filled = sum(1 for r in all_results if r.status == "filled") failed = sum(1 for r in all_results if r.status == "failed") @@ -204,7 +213,7 @@ async def charging_invoke(prompt: str, **kw: Any) -> Any: # (budget exhausted before its turn) never called claude, so it does not # count as an attempt. filled + failed are the only attempted outcomes. drains_attempted = sum(1 for r in all_results if r.status != "skipped") - wall_ms = int((time.monotonic() - cycle_start) * 1000) + wall_ms = _elapsed_ms(cycle_start) return _root.CycleSummary( pages_scanned=len(file_cands_raw), @@ -218,3 +227,73 @@ async def charging_invoke(prompt: str, **kw: Any) -> Any: wall_clock_ms=wall_ms, skipped_budget=skipped_budget, ) + + +def _build_cycle_budget_and_sem(_root: Any) -> tuple[Any, asyncio.Semaphore]: + """Construct the per-cycle budget tracker and concurrency semaphore.""" + budget = _root.CycleBudget( + deadline=time.monotonic() + _root.CORTEX_HEADLESS_BUDGET_SEC, + usd_cap=_root.CORTEX_HEADLESS_USD_BUDGET, + ) + return budget, asyncio.Semaphore(_root.CORTEX_HEADLESS_CONCURRENCY) + + +async def _execute_cycle( + wiki_root: Path, + max_drains: int, + max_anchor_drains: int, + invoke: Callable[..., Awaitable[Any]], + cycle_start: float, + _root: Any, +) -> Any: + """Build candidates, run them concurrently, and summarize the cycle.""" + today = datetime.now(timezone.utc).date().isoformat() + budget, sem = _build_cycle_budget_and_sem(_root) + anchor_cands, file_cands, file_cands_raw = _build_cycle_candidates( + wiki_root, max_drains, max_anchor_drains, _root + ) + anchor_results, file_results = await _gather_cycle_results( + anchor_cands, file_cands, wiki_root, today, invoke, sem, budget, _root + ) + return _summarize_cycle( + cycle_start, + file_cands_raw, + file_cands, + anchor_results, + file_results, + budget, + _root, + ) + + +async def run_headless_authoring_cycle( + wiki_root: Path | None = None, + *, + max_drains: int | None = None, + max_anchor_drains: int | None = None, + invoke: Callable[..., Awaitable[Any]] | None = None, +) -> Any: + """One autonomous cycle: author missing anchor pages, then drain + file-doc gaps, concurrently under a shared semaphore + per-cycle + budget. Anchor pages first (more visibly incomplete than one + missing file-doc section). Invariant: <=CORTEX_HEADLESS_CONCURRENCY + in-flight calls. + """ + # Deferred import (issue #237): see module docstring's import-cycle note. + from . import headless_authoring as _root # noqa: PLC0415 — import cycle (partner: headless_authoring, #237) + + # Stays inline: _root's concrete type here is what lets pyright narrow + # away `| None` (an extracted helper taking `_root: Any` cannot). + if max_drains is None: + max_drains = _root.CORTEX_HEADLESS_MAX_FILE_DRAINS + if max_anchor_drains is None: + max_anchor_drains = _root.CORTEX_HEADLESS_MAX_ANCHOR_DRAINS + if invoke is None: + invoke = _root._claude_invoke + if wiki_root is None: + wiki_root = Path(WIKI_ROOT) + + cycle_start = time.monotonic() + return await _execute_cycle( + wiki_root, max_drains, max_anchor_drains, invoke, cycle_start, _root + ) diff --git a/mcp_server/handlers/consolidation/cycle_types.py b/mcp_server/handlers/consolidation/cycle_types.py new file mode 100644 index 00000000..4c583d42 --- /dev/null +++ b/mcp_server/handlers/consolidation/cycle_types.py @@ -0,0 +1,90 @@ +"""Cycle execution/telemetry data types for the headless authoring worker. + +Split out of ``headless_authoring`` (Fowler: Move Function, issue #276) +to keep that module under the size limit; ``InvokeResult`` and +``_AnchorCandidate`` stay there (worker identity/config, a distinct +concern from a cycle's execution telemetry). The public import surface +stays ``headless_authoring``, which re-exports all three names below — +every existing ``_root.CycleBudget`` / ``_root.DrainResult`` / +``_root.CycleSummary`` call site is unaffected: same class objects, +same module attribute, just defined in a different file. + +No import-cycle concern here: these are plain dataclasses with no +dependency on ``headless_authoring`` or any sibling. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field + + +@dataclass +class CycleBudget: + """Per-cycle wall-clock + USD budget tracker. + + Pre-condition: ``deadline`` is a ``time.monotonic()`` value in the + future; ``usd_cap`` is a float (<=0 means unlimited). + Invariant: ``usd_spent`` is monotonically non-decreasing. + + Concurrency note: with CORTEX_HEADLESS_CONCURRENCY > 1, multiple + coroutines can see ``exhausted() == False`` simultaneously before any + of them has charged the budget. The USD cap is therefore a SOFT + ceiling with overshoot of at most ``concurrency - 1`` calls beyond + the cap. This is acceptable for an operational safety rail. + """ + + deadline: float # time.monotonic() timestamp + usd_cap: float # <=0 means no USD cap + usd_spent: float = field(default=0.0) + + def time_left(self) -> float: + """Remaining seconds until deadline (negative when expired).""" + return self.deadline - time.monotonic() + + def exhausted(self) -> bool: + """True when wall-clock time is up OR the USD cap is reached.""" + if self.time_left() <= 0: + return True + return self.usd_cap > 0 and self.usd_spent >= self.usd_cap + + def charge(self, usd: float) -> None: + """Add ``usd`` to the running spend. + + Pre-condition: ``usd`` >= 0. + Post-condition: ``self.usd_spent`` incremented by ``usd``. + """ + self.usd_spent += usd + + +@dataclass +class DrainResult: + """One drain attempt's outcome.""" + + page_path: str + gap: str + status: str # "filled" | "failed" | "skipped" + duration_ms: int + detail: str = "" + + +@dataclass +class CycleSummary: + """Per-invocation roll-up with budget telemetry.""" + + pages_scanned: int + pages_with_gaps: int + drains_attempted: int + drains_filled: int + drains_failed: int + duration_ms: int + results: list[DrainResult] + # Budget telemetry fields (added in throttle refactor — callers that + # access only the original fields are not affected). + usd_spent: float = 0.0 + # wall_clock_ms is intentionally equal to duration_ms for the cycle: both + # measure the cycle's wall-clock span. Kept as a distinct, explicitly-named + # field because wiki_maintenance's telemetry dict emits both keys; the + # original duration_ms is retained for pre-throttle callers. + wall_clock_ms: int = 0 + skipped_budget: int = 0 diff --git a/mcp_server/handlers/consolidation/drain_operations.py b/mcp_server/handlers/consolidation/drain_operations.py index e58998d6..2f511d31 100644 --- a/mcp_server/handlers/consolidation/drain_operations.py +++ b/mcp_server/handlers/consolidation/drain_operations.py @@ -1,16 +1,17 @@ -"""Drain operations for the headless authoring worker. - -Per-page and per-anchor drain routines that issue ``claude -p`` calls -and rewrite wiki pages. Split out of ``headless_authoring`` to keep -that module under the size limit (Fowler: Move Function). The public -import surface remains ``headless_authoring``; these names are -re-exported there. - -Patchability contract: the default ``invoke`` and the dataclass types -are read from the root module so that the re-exported references stay -identical to the public ones. ``_claude_invoke`` is not monkeypatched -in tests (callers pass ``invoke`` explicitly), so binding the default -at import time preserves the original behaviour. +"""Per-page drain routines for the headless authoring worker. + +Issues ``claude -p`` calls and rewrites wiki pages. Split out of +``headless_authoring`` (Fowler: Move Function); anchor-page authoring +is a separate concern, split further into ``anchor_authoring`` (issue +#276). The public import surface stays ``headless_authoring``, which +these names re-export. + +Import-cycle note (issue #237): a module-top ``from . import +headless_authoring as _root`` would deadlock a fresh interpreter +importing this module before ``headless_authoring`` finishes (it +imports these functions back at load time). Each function resolves +``_root`` lazily at call time instead — every +``monkeypatch.setattr(headless_authoring, ...)`` stays observed. """ from __future__ import annotations @@ -20,7 +21,6 @@ from pathlib import Path from typing import Any -from . import headless_authoring as _root from .authoring_prompts import ( _GAP_DESCRIPTIONS, _build_page_prompt, @@ -32,23 +32,16 @@ ) from mcp_server.observability import silent_failure -from .page_io import ( - _project_source_for_page, - _rewrite_page, - _scope_anchor_prompt, - _write_anchor_page, -) -from mcp_server.core.wiki_coverage import _project_source_root, audit_domain -from datetime import datetime, timezone -from mcp_server.shared.domain_mapping import _build_registry +from .page_io import _project_source_for_page, _rewrite_page +from mcp_server.core.wiki_coverage import _project_source_root def _optional_source_root(meta: dict[str, Any]) -> str | None: """Resolve source_root for --add-dir scope extension (audit B-1). - NOTE: --add-dir extends readable scope; it does NOT confine reads. - Returns None when the domain is absent or resolution fails — the - drain proceeds without the extra scope either way. + --add-dir extends readable scope; it does NOT confine reads. Returns + None when the domain is absent or resolution fails (drain proceeds + without the extra scope either way). """ domain = meta.get("domain") if not domain or not isinstance(domain, str): @@ -60,55 +53,50 @@ def _optional_source_root(meta: dict[str, Any]) -> str | None: return None -async def drain_one( +def _elapsed_ms(start: float) -> int: + """Milliseconds elapsed since ``start`` (a ``time.monotonic()`` value).""" + return int((time.monotonic() - start) * 1000) + + +def _drain_result( + page_path: Path | str, + gap: str, + status: str, + duration_ms: int, + detail: str, + _root: Any, +) -> Any: + """Build one DrainResult — the shared return shape for every drain step.""" + return _root.DrainResult( + page_path=str(page_path), + gap=gap, + status=status, + duration_ms=duration_ms, + detail=detail, + ) + + +async def _finish_drain_one( + ir: Any, page_path: Path, - meta: dict[str, Any], body: str, - *, wiki_root: Path, - invoke: Callable[..., Awaitable[Any]] = _root._claude_invoke, + gaps: list[str], + gap_name: str, + gap_desc: str, + start: float, + _root: Any, ) -> Any: - """Drain the first curation gap on one page (legacy single-section path). + """Apply ``drain_one``'s claude response to the page body and persist it.""" - Pre-condition: ``page_path`` is a valid wiki page under ``wiki_root``; - ``meta`` is parsed - frontmatter; ``body`` is the page body. - Post-condition: the gap marker is replaced in the file on disk and the - result reflects the outcome (filled/failed/skipped). - """ - start = time.monotonic() - gaps = meta.get("curation_gaps") or [] - if not gaps: - return _root.DrainResult( - page_path=str(page_path), - gap="", - status="skipped", - duration_ms=0, - detail="no gaps", + def result(status: str, detail: str) -> Any: + return _drain_result( + page_path, gap_name, status, _elapsed_ms(start), detail, _root ) - gap_name = gaps[0] - gap_desc = _GAP_DESCRIPTIONS.get(gap_name) or gap_name - _, source_text = _project_source_for_page(meta) - src_root = _optional_source_root(meta) - prompt = _build_section_prompt( - page_path=str(page_path), - page_meta=meta, - gap_name=_gap_heading(gap_name), - gap_description=gap_desc, - source_text=source_text, - delegate_hint=_root._delegation_hint_for(meta.get("kind") or "file-doc"), - ) - ir = await invoke(prompt, source_root=src_root) response = ir.text if response is None or response.strip() == "": - return _root.DrainResult( - page_path=str(page_path), - gap=gap_name, - status="failed", - duration_ms=int((time.monotonic() - start) * 1000), - detail="claude invocation failed", - ) + return result("failed", "claude invocation failed") response_stripped = response.strip() if response_stripped.upper().startswith("NO INFORMATION AVAILABLE"): new_body, did = _replace_gap_marker( @@ -117,94 +105,85 @@ async def drain_one( else: new_body, did = _replace_gap_marker(body, gap_desc, response_stripped) if not did: - return _root.DrainResult( - page_path=str(page_path), - gap=gap_name, - status="failed", - duration_ms=int((time.monotonic() - start) * 1000), - detail="gap marker not found in body", - ) + return result("failed", "gap marker not found in body") new_gaps = [g for g in gaps if g != gap_name] ok = await _rewrite_page( page_path, wiki_root, new_body=new_body, new_curation_gaps=new_gaps ) - return _root.DrainResult( - page_path=str(page_path), - gap=gap_name, - status="filled" if ok else "failed", - duration_ms=int((time.monotonic() - start) * 1000), - detail="" if ok else "page rewrite failed", - ) + return result("filled" if ok else "failed", "" if ok else "page rewrite failed") -# ── Whole-page drain (drain_all_gaps_on_page) ────────────────────── -# -# The single-section drain (above) was the proof of concept. The -# bulk-drain below issues ONE ``claude -p`` call per page that fills -# EVERY missing section in one response — about 7-8× faster per -# page, gives the LLM the full picture so cross-references between -# sections stay coherent, and lets one autonomous cycle materially -# move the 14k-gap backlog instead of nibbling at it. +async def _invoke_section_fill( + page_path: Path, + meta: dict[str, Any], + gap_name: str, + invoke: Callable[..., Awaitable[Any]], + _root: Any, +) -> tuple[Any, str]: + """Build the single-section fill prompt, invoke claude, return (ir, gap_desc).""" + gap_desc = _GAP_DESCRIPTIONS.get(gap_name) or gap_name + _, source_text = _project_source_for_page(meta) + src_root = _optional_source_root(meta) + prompt = _build_section_prompt( + page_path=str(page_path), + page_meta=meta, + gap_name=_gap_heading(gap_name), + gap_description=gap_desc, + source_text=source_text, + delegate_hint=_root._delegation_hint_for(meta.get("kind") or "file-doc"), + ) + ir = await invoke(prompt, source_root=src_root) + return ir, gap_desc -async def drain_all_gaps_on_page( +async def drain_one( page_path: Path, meta: dict[str, Any], body: str, *, wiki_root: Path, - invoke: Callable[..., Awaitable[Any]] = _root._claude_invoke, -) -> list[Any]: - """Fill every curation gap on one page in a single ``claude -p`` call. + invoke: Callable[..., Awaitable[Any]] | None = None, +) -> Any: + """Drain the first curation gap on one page (legacy single-section path). - Returns one DrainResult per gap so the cycle summary still - accounts for each individually. A failure on one gap leaves the - others' fills intact — the parser tolerates missing delimiters, - so partial responses still make progress. - - The gap set is computed by LIVE AUDIT (not the frozen frontmatter - list) so newly-added canonical sections — sequence diagram, - parameters, request/response examples — get filled on pages that - already exist. - - Pre-condition: ``page_path`` is a wiki page with curation gaps under - ``wiki_root``; ``invoke`` is an async callable matching - ``_claude_invoke``'s signature. - Post-condition: all filled sections are written to disk through the - governed wiki-write path (write_class='mechanical' - pointer memory + citation sync — see - ``write_governed_page``); returned list has one - DrainResult per gap (filled/failed). + Pre: ``page_path`` is a valid wiki page under ``wiki_root``; ``meta`` + is parsed frontmatter; ``body`` is the page body. + Post: the gap marker is replaced on disk; result reflects the outcome + (filled/failed/skipped). """ + # Deferred import (issue #237): see module docstring's import-cycle note. + from . import headless_authoring as _root # noqa: PLC0415 — import cycle (partner: headless_authoring, #237) + + if invoke is None: + invoke = _root._claude_invoke + start = time.monotonic() - frozen = [g for g in (meta.get("curation_gaps") or []) if isinstance(g, str)] - gaps = _live_audit_gaps(body, frozen) + gaps = meta.get("curation_gaps") or [] if not gaps: - return [] + return _drain_result(page_path, "", "skipped", 0, "no gaps", _root) - src_root = _optional_source_root(meta) - _, source_text = _project_source_for_page(meta) - prompt = _build_page_prompt( - page_path=str(page_path), - page_meta=meta, - gaps=gaps, - source_text=source_text, - delegate_hint=_root._delegation_hint_for(meta.get("kind") or "file-doc"), + gap_name = gaps[0] + ir, gap_desc = await _invoke_section_fill(page_path, meta, gap_name, invoke, _root) + return await _finish_drain_one( + ir, page_path, body, wiki_root, gaps, gap_name, gap_desc, start, _root ) - ir = await invoke(prompt, source_root=src_root) - base_ms = int((time.monotonic() - start) * 1000) - response = ir.text - if not response: - return [ - _root.DrainResult( - page_path=str(page_path), - gap=g, - status="failed", - duration_ms=base_ms, - detail="claude invocation failed", - ) - for g in gaps - ] + + +def _fill_gaps_from_response( + response: str, + gaps: list[str], + body: str, + page_path: Path, + base_ms: int, + _root: Any, +) -> tuple[str, list[str], list[Any]]: + """Parse the sectioned response and apply each fill to the page body. + + Returns ``(new_body, filled_gaps, results)`` — one DrainResult per gap. + """ + + def result(gap: str, status: str, detail: str) -> Any: + return _drain_result(page_path, gap, status, base_ms, detail, _root) filled_map = _parse_sectioned_response(response, gaps) new_body = body @@ -214,157 +193,108 @@ async def drain_all_gaps_on_page( content = filled_map.get(g) gap_desc = _GAP_DESCRIPTIONS.get(g) or g if not content: - results.append( - _root.DrainResult( - page_path=str(page_path), - gap=g, - status="failed", - duration_ms=base_ms, - detail="not in response", - ) - ) + results.append(result(g, "failed", "not in response")) continue if content.upper().startswith("NO INFORMATION AVAILABLE"): content = "_(no information available for this section)_" new_body, did = _replace_gap_marker(new_body, gap_desc, content) if not did: - results.append( - _root.DrainResult( - page_path=str(page_path), - gap=g, - status="failed", - duration_ms=base_ms, - detail="marker not found", - ) - ) + results.append(result(g, "failed", "marker not found")) continue filled_gaps.append(g) - results.append( - _root.DrainResult( - page_path=str(page_path), - gap=g, - status="filled", - duration_ms=base_ms, - detail="", - ) - ) - if filled_gaps: - remaining = [g for g in gaps if g not in filled_gaps] - wrote = await _rewrite_page( - page_path, wiki_root, new_body=new_body, new_curation_gaps=remaining - ) - if not wrote: - # The governed write failed after content was successfully parsed - # out of the response — downgrade the optimistic "filled" results - # for this page to "failed" so the cycle summary (and any caller - # counting drains_filled) reflects reality, not the parse step. - for r in results: - if r.gap in filled_gaps: - r.status = "failed" - r.detail = "governed write failed" - return results + results.append(result(g, "filled", "")) + return new_body, filled_gaps, results -# ── Anchor-page authoring path ───────────────────────────────────── -# -# A project that's missing its architecture / services / api / ci-cd -# / mcp / ai-usage / prd / decisions anchor pages doesn't have any -# gap markers to drain — the pages simply don't exist. The fix is -# the symmetric move: detect missing anchors via the coverage audit, -# feed Claude a project-level overview (file tree, README, key -# config files, source file counts), and ask it to author the -# anchor from scratch. -# -# 285 anchors total (15 scopes × 19 projects); these drain in one -# autonomous run because each anchor is one ``claude -p`` call. +async def _persist_filled_gaps( + page_path: Path, + wiki_root: Path, + new_body: str, + gaps: list[str], + filled_gaps: list[str], + results: list[Any], +) -> None: + """Write remaining gaps; downgrade results to "failed" if the write fails.""" + remaining = [g for g in gaps if g not in filled_gaps] + wrote = await _rewrite_page( + page_path, wiki_root, new_body=new_body, new_curation_gaps=remaining + ) + if not wrote: + for r in results: + if r.gap in filled_gaps: + r.status = "failed" + r.detail = "governed write failed" -async def drain_missing_anchors( - wiki_root: Path, +async def _invoke_page_fill( + page_path: Path, + meta: dict[str, Any], + gaps: list[str], + invoke: Callable[..., Awaitable[Any]], + _root: Any, +) -> tuple[Any, int]: + """Build the whole-page fill prompt, invoke claude, return (ir, elapsed_ms).""" + start = time.monotonic() + src_root = _optional_source_root(meta) + _, source_text = _project_source_for_page(meta) + prompt = _build_page_prompt( + page_path=str(page_path), + page_meta=meta, + gaps=gaps, + source_text=source_text, + delegate_hint=_root._delegation_hint_for(meta.get("kind") or "file-doc"), + ) + ir = await invoke(prompt, source_root=src_root) + return ir, _elapsed_ms(start) + + +def _no_response_results( + page_path: Path, gaps: list[str], base_ms: int, _root: Any +) -> list[Any]: + """DrainResults for a page whose ``claude -p`` call returned no text.""" + return [ + _drain_result( + page_path, g, "failed", base_ms, "claude invocation failed", _root + ) + for g in gaps + ] + + +async def drain_all_gaps_on_page( + page_path: Path, + meta: dict[str, Any], + body: str, *, - max_drains: int = 30, - today: str | None = None, - invoke: Callable[..., Awaitable[Any]] = _root._claude_invoke, + wiki_root: Path, + invoke: Callable[..., Awaitable[Any]] | None = None, ) -> list[Any]: - """Author missing canonical anchor pages for every project. - - For each domain × scope combination with no covered anchor, calls - ``claude -p`` with a project-level context block and writes the - response as the new anchor page. Up to ``max_drains`` authored per - invocation so a single cycle stays time-bounded. - - Piece 3 — Groundable-only filter: scopes marked ``groundable=False`` - in ``wiki_coverage.SCOPES`` (e.g. prd, decisions, changelog, roadmap, - accessibility, localization) are skipped entirely — their content - cannot be derived by reading the source tree and autonomous authoring - would be fabrication (zetetic-forbidden). They remain visible as - coverage gaps for human authors. - - Pre-condition: ``wiki_root`` is a valid directory; ``invoke`` is an - async callable matching ``_claude_invoke``'s signature. - Post-condition: up to ``max_drains`` new anchor pages written to disk; - ungroundable scopes are omitted (not skipped-with-result). + """Fill every curation gap on one page in a single ``claude -p`` call. + + One request/page (vs. ``drain_one``'s one/gap) is ~7-8x faster and + keeps cross-references coherent; gap set is a LIVE AUDIT. One + DrainResult per gap; a failure on one gap leaves others intact. """ + # Deferred import (issue #237): see module docstring's import-cycle note. + from . import headless_authoring as _root # noqa: PLC0415 — import cycle (partner: headless_authoring, #237) - today = today or datetime.now(timezone.utc).date().isoformat() - domains = sorted({r.canonical for r in _build_registry().repos}) + if invoke is None: + invoke = _root._claude_invoke - results: list[Any] = [] - for domain in domains: - if len([r for r in results if r.status == "filled"]) >= max_drains: - break - src_root = _project_source_root(domain) - if not src_root: - continue - cov = audit_domain(str(wiki_root), domain) - for sc in cov.scopes: - if sc.covered: - continue - # Piece 3: skip ungroundable scopes — content cannot be - # derived from source tree alone; autonomous authoring would - # fabricate rather than document. - if not sc.scope.groundable: - continue - if len([r for r in results if r.status == "filled"]) >= max_drains: - break - t0 = time.monotonic() - prompt = _scope_anchor_prompt( - domain=domain, - scope_name=sc.scope.name, - scope_title=sc.scope.title, - scope_description=sc.scope.description, - source_root=src_root, - delegate_hint=_root._delegation_hint_for(sc.scope.suggested_kind), - ) - ir = await invoke(prompt, cwd=src_root, source_root=src_root) - response = ir.text - if not response or response.strip() == "": - results.append( - _root.DrainResult( - page_path=sc.suggested_path, - gap=f"anchor:{sc.scope.name}", - status="failed", - duration_ms=int((time.monotonic() - t0) * 1000), - detail="claude returned empty", - ) - ) - continue - written = await _write_anchor_page( - wiki_root=wiki_root, - domain=domain, - scope_name=sc.scope.name, - suggested_kind=sc.scope.suggested_kind, - suggested_path=sc.suggested_path, - body_markdown=response.strip(), - today=today, - ) - results.append( - _root.DrainResult( - page_path=str(written) if written else sc.suggested_path, - gap=f"anchor:{sc.scope.name}", - status="filled" if written else "failed", - duration_ms=int((time.monotonic() - t0) * 1000), - detail="" if written else "page write failed", - ) - ) + frozen = [g for g in (meta.get("curation_gaps") or []) if isinstance(g, str)] + gaps = _live_audit_gaps(body, frozen) + if not gaps: + return [] + + ir, base_ms = await _invoke_page_fill(page_path, meta, gaps, invoke, _root) + response = ir.text + if not response: + return _no_response_results(page_path, gaps, base_ms, _root) + + new_body, filled_gaps, results = _fill_gaps_from_response( + response, gaps, body, page_path, base_ms, _root + ) + if filled_gaps: + await _persist_filled_gaps( + page_path, wiki_root, new_body, gaps, filled_gaps, results + ) return results diff --git a/mcp_server/handlers/consolidation/headless_authoring.py b/mcp_server/handlers/consolidation/headless_authoring.py index 93191848..dae1b0c9 100644 --- a/mcp_server/handlers/consolidation/headless_authoring.py +++ b/mcp_server/handlers/consolidation/headless_authoring.py @@ -1,67 +1,58 @@ """Headless authoring worker — drains the curation-gap queue. This is the actuator Meadows' leverage-point audit identified as -missing (2026-05-18). The gap detector knows what's missing; the -``curate_wiki`` tool builds prompts for the LLM; but until now the -loop terminated in a queue waiting for a human to open a Claude Code -session and consume the jobs interactively. The drain rate was zero -because the actuator was disconnected. - -The worker connects sensor → actuator. It: - - 1. Walks the wiki for pages with ``curation_gaps`` in frontmatter. - 2. For each page, picks the highest-leverage gap (the first one - listed — see ``FILE_DOC_SECTIONS`` for ordering) plus enough - source context that the LLM can answer it. - 3. Calls the user's Claude Code session via the ``claude -p`` CLI - to author the missing section. No API key configuration needed - — Claude Code's existing credentials carry through. - 4. Rewrites the page: the ``_(missing — needs: )_`` - marker is replaced with the authored content; the - ``curation_gaps`` frontmatter list shrinks; the ``lifecycle`` - promotes from ``needs-curation`` toward ``draft`` then - ``accepted`` as more gaps fill. - -The worker is per-cycle bounded — it drains at most ``MAX_DRAINS`` -pages per invocation so a single cycle doesn't monopolise the -session. Subsequent ``consolidate_background`` runs drain the rest. - -Failure handling: a failed LLM call leaves the page untouched. The -gap stays in the queue, the next cycle retries. The page is never -corrupted; the marker is only replaced after a successful LLM -response. - -Module layout (split 2026-06-30 to satisfy the 500-line size limit -without changing behaviour — strategy B in the refactor brief): this -module remains the stable public import surface. It defines the -constants, the dataclass types, and ``_claude_invoke`` (whose security -controls are kept verbatim here), then re-exports the candidate -scanners, drain operations, and the concurrent cycle from sibling -modules: - - * ``authoring_prompts`` — prompt builders, parsers, gap markers. - * ``page_io`` — frontmatter parse/rewrite, file reads, - anchor-page prompt + writer. - * ``candidate_scan`` — ``_scan_pages_with_gaps`` / - ``_collect_anchor_candidates``. - * ``drain_operations`` — ``drain_one`` / ``drain_all_gaps_on_page`` - / ``drain_missing_anchors``. - * ``cycle_orchestration`` — ``run_headless_authoring_cycle``. +missing (2026-05-18): the gap detector knows what's missing and +``curate_wiki`` builds prompts, but the loop terminated in a queue +waiting for a human to consume the jobs interactively. The drain rate +was zero because the actuator was disconnected. + +The worker connects sensor -> actuator: walks the wiki for pages with +``curation_gaps``, calls the user's Claude Code session via ``claude +-p`` to author the missing section (no API key needed — existing +credentials carry through), and rewrites the page (marker replaced, +``curation_gaps`` shrinks, ``lifecycle`` promotes toward ``accepted``). +Per-cycle bounded (``MAX_DRAINS_PER_CYCLE``); subsequent +``consolidate_background`` runs drain the rest. A failed LLM call +leaves the page untouched — never corrupted, only replaced on success. + +Module layout (split 2026-06-30, then again 2026-07-30 for #276, to +satisfy the size limit without changing behaviour): this module +remains the stable public import surface, defining the constants and +dataclass types, then re-exporting everything else: + + * ``authoring_prompts`` — prompt builders, parsers, gap markers. + * ``page_io`` — frontmatter parse/rewrite, file reads, + anchor-page prompt + writer. + * ``candidate_scan`` — ``_scan_pages_with_gaps`` / + ``_collect_anchor_candidates``. + * ``drain_operations`` — ``drain_one`` / ``drain_all_gaps_on_page``. + * ``anchor_authoring`` — ``drain_missing_anchors``. + * ``cycle_orchestration`` — ``run_headless_authoring_cycle``. + * ``claude_invoke`` — ``_claude_invoke`` (the ``claude -p`` + subprocess call; security controls live + in the argv/env builders in ``claude_cli``). The scanners and the cycle resolve the patchable names (``CORTEX_HEADLESS_*``, ``_collect_anchor_candidates``, ``_scan_pages_with_gaps``) as attributes of THIS module at call time, so ``monkeypatch.setattr(headless_authoring, ...)`` is observed. + +Import direction (fixed 2026-07-30, issue #237): the siblings above +used to import THIS module back at their own module top, deadlocking +any fresh interpreter that imported one of them first (partial-module +``ImportError`` — reproducible with e.g. ``python -c "import +mcp_server.handlers.consolidation.candidate_scan"``). Each sibling now +resolves ``_root`` with a deferred, function-scoped import instead +(``# noqa: PLC0415 — import cycle``, per pyproject.toml's named +exemption for this family) — the load-time back-reference is gone, the +call-time patchability above is unchanged. """ from __future__ import annotations -import asyncio -import json import logging import os -import time -from dataclasses import dataclass, field +from dataclasses import dataclass from .authoring_prompts import _delegation_hint logger = logging.getLogger(__name__) @@ -83,20 +74,14 @@ _CLAUDE_BIN = "claude" -# ── Environment-configured knobs ──────────────────────────────────────── -# -# Read at import time so values are stable for the process lifetime. -# All defaults are POLICY CAPS — operational safety rails bounding -# one cycle's resource consumption. They are NOT scientifically -# measured constants: tuning them to match your hardware and cost -# tolerance via env vars is expected and encouraged. +# ── Environment-configured knobs — read at import time so values stay +# stable for the process lifetime. All defaults are POLICY CAPS, not +# measured constants: tune via env vars to match your hardware/cost. ── def _env_int(name: str, default: int) -> int: """Return int from env var ``name``, or ``default`` when absent/invalid. - - Pre-condition: ``default`` is a non-negative int. - Post-condition: returns a valid int; never raises; logs on bad values. + Never raises; logs on a bad value. """ raw = os.environ.get(name, "").strip() if not raw: @@ -115,9 +100,7 @@ def _env_int(name: str, default: int) -> int: def _env_float(name: str, default: float) -> float: """Return float from env var ``name``, or ``default`` when absent/invalid. - - Pre-condition: ``default`` is a non-negative float. - Post-condition: returns a valid float; never raises; logs on bad values. + Never raises; logs on a bad value. """ raw = os.environ.get(name, "").strip() if not raw: @@ -193,77 +176,6 @@ class InvokeResult: cost_usd: float # client-side spend estimate; 0.0 when unavailable -@dataclass -class CycleBudget: - """Per-cycle wall-clock + USD budget tracker. - - Pre-condition: ``deadline`` is a ``time.monotonic()`` value in the - future; ``usd_cap`` is a float (<=0 means unlimited). - Invariant: ``usd_spent`` is monotonically non-decreasing. - - Concurrency note: with CORTEX_HEADLESS_CONCURRENCY > 1, multiple - coroutines can see ``exhausted() == False`` simultaneously before any - of them has charged the budget. The USD cap is therefore a SOFT - ceiling with overshoot of at most ``concurrency - 1`` calls beyond - the cap. This is acceptable for an operational safety rail. - """ - - deadline: float # time.monotonic() timestamp - usd_cap: float # <=0 means no USD cap - usd_spent: float = field(default=0.0) - - def time_left(self) -> float: - """Remaining seconds until deadline (negative when expired).""" - return self.deadline - time.monotonic() - - def exhausted(self) -> bool: - """True when wall-clock time is up OR the USD cap is reached.""" - if self.time_left() <= 0: - return True - return self.usd_cap > 0 and self.usd_spent >= self.usd_cap - - def charge(self, usd: float) -> None: - """Add ``usd`` to the running spend. - - Pre-condition: ``usd`` >= 0. - Post-condition: ``self.usd_spent`` incremented by ``usd``. - """ - self.usd_spent += usd - - -@dataclass -class DrainResult: - """One drain attempt's outcome.""" - - page_path: str - gap: str - status: str # "filled" | "failed" | "skipped" - duration_ms: int - detail: str = "" - - -@dataclass -class CycleSummary: - """Per-invocation roll-up with budget telemetry.""" - - pages_scanned: int - pages_with_gaps: int - drains_attempted: int - drains_filled: int - drains_failed: int - duration_ms: int - results: list[DrainResult] - # Budget telemetry fields (added in throttle refactor — callers that - # access only the original fields are not affected). - usd_spent: float = 0.0 - # wall_clock_ms is intentionally equal to duration_ms for the cycle: both - # measure the cycle's wall-clock span. Kept as a distinct, explicitly-named - # field because wiki_maintenance's telemetry dict emits both keys; the - # original duration_ms is retained for pre-throttle callers. - wall_clock_ms: int = 0 - skipped_budget: int = 0 - - @dataclass class _AnchorCandidate: """One missing groundable anchor to author (pre-screened, no I/O).""" @@ -277,121 +189,6 @@ class _AnchorCandidate: suggested_kind: str -async def _claude_invoke( - prompt: str, - *, - cwd: str | None = None, - source_root: str | None = None, - timeout: float | None = None, -) -> InvokeResult: - """Run ``claude -p`` asynchronously and return an InvokeResult. - - Uses ``asyncio.create_subprocess_exec`` + ``asyncio.wait_for`` so - the call is non-blocking on the event loop. On timeout the - subprocess is killed and an empty InvokeResult is returned. - - The argv and child environment — including the full audit-B-1 security - argument for both agents mode (the default, loading the user's zetetic - roster under a hard write/exec ceiling) and solo ``--safe-mode`` mode — - are built by ``claude_cli._build_argv`` / ``claude_cli._subprocess_env``. - Auth is subscription-by-default, ``CORTEX_HEADLESS_AUTH=api`` opt-in. - - Response parsing relies on ``--output-format json``: ``result`` (assistant - text) and ``total_cost_usd`` (client-side spend). ``usage`` / ``is_error`` - are NOT guaranteed in the CLI JSON — errors are detected via subprocess - returncode only. - - The prompt is fed via STDIN, not as a positional argv element: the - variadic ``--add-dir`` would otherwise swallow a trailing prompt and the - CLI would error "Input must be provided". See ``claude_cli._build_argv``. - """ - argv = _build_argv(source_root) - - call_timeout = timeout if timeout is not None else float(CLAUDE_CALL_TIMEOUT_SEC) - - # Subscription by default + hook-neutralising child flag; API key passes - # through only on CORTEX_HEADLESS_AUTH=api opt-in. See claude_cli. - child_env = _subprocess_env() - - try: - proc = await asyncio.create_subprocess_exec( - *argv, - stdin=asyncio.subprocess.PIPE, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - cwd=cwd, - env=child_env, - ) - except FileNotFoundError: - logger.warning("headless-authoring: claude binary not found on PATH") - return InvokeResult(text=None, cost_usd=0.0) - except Exception as exc: # noqa: BLE001 — last-resort boundary — failure is logged; degraded mode continues - logger.warning("headless-authoring: failed to start claude subprocess: %s", exc) - return InvokeResult(text=None, cost_usd=0.0) - - try: - stdout_bytes, stderr_bytes = await asyncio.wait_for( - proc.communicate(input=prompt.encode("utf-8")), timeout=call_timeout - ) - except asyncio.TimeoutError: - logger.warning( - "headless-authoring: claude -p timed out after %.0fs", call_timeout - ) - return InvokeResult(text=None, cost_usd=0.0) - except Exception as exc: # noqa: BLE001 — last-resort boundary — failure is logged; degraded mode continues - logger.warning("headless-authoring: claude -p communicate failed: %s", exc) - return InvokeResult(text=None, cost_usd=0.0) - finally: - # CancelledError is a BaseException — it escapes the except clauses above. - # Without this finally, a cancelled drain leaves a zombie subprocess. - # Covers the timeout path too (returncode is None after wait_for cancels - # communicate), so the kill lives in one place. - if proc.returncode is None: - try: - proc.kill() - except ProcessLookupError: - pass - await proc.wait() - - stdout = stdout_bytes.decode("utf-8", errors="replace") if stdout_bytes else "" - stderr = stderr_bytes.decode("utf-8", errors="replace") if stderr_bytes else "" - - if proc.returncode != 0: - logger.warning( - "headless-authoring: claude -p exit %d stderr=%r", - proc.returncode, - stderr[:300], - ) - return InvokeResult(text=None, cost_usd=0.0) - - stdout = stdout.strip() - if not stdout: - return InvokeResult(text=None, cost_usd=0.0) - - # Parse --output-format json response. - # Documented fields (source: code.claude.com/docs/en/headless): - # result (str) — the assistant text - # total_cost_usd (float) — client-side cost estimate - # ``usage`` and ``is_error`` are NOT guaranteed — use returncode only. - try: - data = json.loads(stdout) - text: str | None = data.get("result") or None - cost_usd = float(data.get("total_cost_usd") or 0.0) - except (json.JSONDecodeError, ValueError): - # Defensive: returncode==0 but JSON parse failed. This can happen - # if --output-format json isn't supported by an older claude CLI - # build. Treat raw stdout as the text so we degrade gracefully - # rather than losing a successful response. - logger.debug( - "headless-authoring: JSON parse failed (returncode=0); " - "treating raw stdout as text (cost unknown)" - ) - text = stdout or None - cost_usd = 0.0 - - return InvokeResult(text=text, cost_usd=cost_usd) - - def _delegation_hint_for(kind: str) -> str | None: """Return the Task-delegation prompt paragraph for ``kind``, or None. @@ -409,11 +206,14 @@ def _delegation_hint_for(kind: str) -> str | None: # ── Re-exports — the public import surface (see module docstring) ───────── # -# These siblings import THIS module as ``_root`` and read the patchable -# names off it at call time, so the imports below MUST come after every -# constant / type / ``_claude_invoke`` definition above. This is a -# deliberate, load-order-sensitive circular import. - +# These siblings resolve THIS module as ``_root`` via a deferred, +# function-scoped import (issue #237) and read the patchable names off it +# at call time — no back-reference at module scope, so nothing below is +# load-order-sensitive anymore. The imports stay after the constant/type +# definitions purely for readability (this module defines its own public +# surface before re-exporting the rest of it). + +from .cycle_types import CycleBudget, CycleSummary, DrainResult # noqa: E402 from .claude_cli import _build_argv, _subprocess_env # noqa: E402 from .candidate_scan import ( # noqa: E402 _collect_anchor_candidates, @@ -421,10 +221,11 @@ def _delegation_hint_for(kind: str) -> str | None: ) from .drain_operations import ( # noqa: E402 drain_all_gaps_on_page, - drain_missing_anchors, drain_one, ) +from .anchor_authoring import drain_missing_anchors # noqa: E402 from .cycle_orchestration import run_headless_authoring_cycle # noqa: E402 +from .claude_invoke import _claude_invoke # noqa: E402 __all__ = [ "InvokeResult", diff --git a/tests_py/handlers/consolidation/test_candidate_scan.py b/tests_py/handlers/consolidation/test_candidate_scan.py new file mode 100644 index 00000000..4bcc5e48 --- /dev/null +++ b/tests_py/handlers/consolidation/test_candidate_scan.py @@ -0,0 +1,112 @@ +"""Direct (non-mocked) coverage for candidate_scan's filesystem scan. + +Every other test in this package monkeypatches ``_scan_pages_with_gaps`` +out entirely (see test_headless_authoring_throttle.py / +test_import_cycle_237.py), so the real walking/filtering/tuple-building +logic in ``_gap_entry_for_page`` had zero test-suite coverage before +this file — a gap mutation testing surfaced during issue #276's Extract +Function pass over candidate_scan.py (mutmut showed survivors on the +dotfile-prefix check and the appended-tuple content, both invisible to +the mocked-out test suite). These tests exercise the real filesystem +walk against a tmp_path wiki tree. +""" + +from __future__ import annotations + +from pathlib import Path + +from mcp_server.handlers.consolidation.candidate_scan import _scan_pages_with_gaps + + +def _write_page( + path: Path, *, gaps: list[str] | None = None, extra_frontmatter: str = "" +) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + gaps_block = "" + if gaps: + gaps_block = "curation_gaps:\n" + "".join(f" - {g}\n" for g in gaps) + path.write_text( + f"---\ntitle: test page\n{extra_frontmatter}{gaps_block}---\nBody text.\n", + encoding="utf-8", + ) + + +def test_scan_pages_with_gaps_finds_frozen_gap_pages(tmp_path: Path) -> None: + """A page with a non-empty curation_gaps list is returned verbatim.""" + page = tmp_path / "architecture.md" + _write_page(page, gaps=["callers", "consumers"]) + + result = _scan_pages_with_gaps(tmp_path) + + assert len(result) == 1 + found_path, meta, body = result[0] + assert found_path == page + assert meta["curation_gaps"] == ["callers", "consumers"] + assert body.strip() == "Body text." + + +def test_scan_pages_with_gaps_skips_page_without_gaps(tmp_path: Path) -> None: + """A page with no curation_gaps and not kind=reference is excluded.""" + _write_page(tmp_path / "complete.md", gaps=None) + + assert _scan_pages_with_gaps(tmp_path) == [] + + +def test_scan_pages_with_gaps_skips_dotfile_and_underscore_dirs(tmp_path: Path) -> None: + """Pages under a dot- or underscore-prefixed directory are never scanned, + even when they carry curation_gaps (e.g. ``_drafts/`` or ``.trash/``). + """ + _write_page(tmp_path / "_drafts" / "wip.md", gaps=["callers"]) + _write_page(tmp_path / ".trash" / "old.md", gaps=["callers"]) + + assert _scan_pages_with_gaps(tmp_path) == [] + + +def test_scan_pages_with_gaps_returns_empty_for_missing_dir(tmp_path: Path) -> None: + """A non-existent wiki_root returns [] rather than raising.""" + assert _scan_pages_with_gaps(tmp_path / "does-not-exist") == [] + + +def test_scan_pages_with_gaps_multiple_pages_mixed(tmp_path: Path) -> None: + """Only the gap-carrying page is returned when siblings have none.""" + gapped = tmp_path / "with-gaps.md" + _write_page(gapped, gaps=["parameters"]) + _write_page(tmp_path / "without-gaps.md", gaps=None) + + result = _scan_pages_with_gaps(tmp_path) + + assert [p for p, _, _ in result] == [gapped] + + +def test_scan_pages_with_gaps_live_audits_file_docs_with_no_frozen_gaps( + tmp_path: Path, +) -> None: + """A kind=reference file-doc with a source_file_path but no frozen + curation_gaps is still included when the live section audit (added + after the page was generated) finds missing canonical sections. + """ + page = tmp_path / "reference.md" + _write_page( + page, + gaps=None, + extra_frontmatter="kind: reference\nsource_file_path: src/foo.py\n", + ) + + result = _scan_pages_with_gaps(tmp_path) + + assert [p for p, _, _ in result] == [page] + + +def test_scan_pages_with_gaps_skips_non_reference_kind_even_with_source_path( + tmp_path: Path, +) -> None: + """The live-audit axis only force-audits kind=reference file-docs — + an ADR/spec/guide with a source_file_path is never live-audited. + """ + _write_page( + tmp_path / "adr.md", + gaps=None, + extra_frontmatter="kind: adr\nsource_file_path: src/foo.py\n", + ) + + assert _scan_pages_with_gaps(tmp_path) == [] diff --git a/tests_py/handlers/consolidation/test_import_cycle_237.py b/tests_py/handlers/consolidation/test_import_cycle_237.py new file mode 100644 index 00000000..5b44cf6e --- /dev/null +++ b/tests_py/handlers/consolidation/test_import_cycle_237.py @@ -0,0 +1,294 @@ +"""Regression: the six consolidation submodules import standalone (#237). + +Before the fix, ``candidate_scan``, ``claude_cli``, ``cycle_orchestration``, +and ``drain_operations`` each did ``from . import headless_authoring as +_root`` at module top, while ``headless_authoring`` imported names back from +each of them at ITS module top. Importing ``headless_authoring`` first +resolved the cycle (which is why the test suite stayed green), but importing +any of the four directly in a fresh interpreter raised: + + ImportError: cannot import name '' from partially initialized + module '...' (most likely due to a circular import) + +``anchor_authoring`` and ``claude_invoke`` (issue #276 — split out of +``drain_operations`` / ``headless_authoring`` respectively to stay under the +file-size cap) carry the identical deferred-import pattern for the same +reason, so they join the list below. (``cycle_types``, also split out for +#276, has no dependency on ``headless_authoring`` at all — plain dataclasses +— so it carries no load-order risk and needs no entry here.) + +Each import below MUST run in its own subprocess — within a single +interpreter, the first successful import populates ``sys.modules`` and +masks the cycle for every subsequent import, exactly the false-green this +regression test exists to prevent (matching issue #233 criterion 1's +subprocess pattern for the same defect class). +""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path +from typing import Any + +import pytest + +_AFFECTED_MODULES = ( + "mcp_server.handlers.consolidation.candidate_scan", + "mcp_server.handlers.consolidation.claude_cli", + "mcp_server.handlers.consolidation.cycle_orchestration", + "mcp_server.handlers.consolidation.drain_operations", + "mcp_server.handlers.consolidation.anchor_authoring", + "mcp_server.handlers.consolidation.claude_invoke", +) + + +@pytest.mark.parametrize("module_name", _AFFECTED_MODULES) +def test_submodule_imports_standalone_in_fresh_interpreter(module_name: str) -> None: + """Each submodule imports successfully with no prior sibling import. + + Regression for issue #237: pre-fix, this subprocess call exits non-zero + with the partially-initialized-module ImportError quoted in the module + docstring above. + """ + result = subprocess.run( + [sys.executable, "-c", f"import {module_name}"], + capture_output=True, + text=True, + timeout=30, + ) + assert result.returncode == 0, ( + f"import {module_name} failed in a fresh interpreter:\n{result.stderr}" + ) + + +def test_headless_authoring_still_imports_first_and_re_exports() -> None: + """The normal (hub-first) import path is unaffected by the fix. + + Runs in-process (not a subprocess): this repo's conftest already + imported plenty of ``mcp_server`` modules by the time this test runs, + so this assertion is about the re-export surface, not fresh-interpreter + import order (that's covered by the parametrized test above). + """ + from mcp_server.handlers.consolidation import headless_authoring as ha + + assert callable(ha.run_headless_authoring_cycle) + assert callable(ha._collect_anchor_candidates) + assert callable(ha._scan_pages_with_gaps) + assert callable(ha.drain_one) + assert callable(ha.drain_all_gaps_on_page) + assert callable(ha.drain_missing_anchors) + assert callable(ha._build_argv) + assert callable(ha._subprocess_env) + + +# ── Sentinel-default resolution (part of the #237 fix, not just the import +# reordering) ──────────────────────────────────────────────────────────── +# +# Breaking the load-time cycle required converting the DI-seam parameters +# that used to be baked at function-DEFINITION time (``invoke: ... = +# _root._claude_invoke``, ``max_drains: int = _root.CORTEX_HEADLESS_MAX_ +# FILE_DRAINS``) into ``None`` sentinels resolved inside the function body +# instead — ``_root.X`` cannot appear in a parameter-list default when +# ``_root`` is only bound lazily inside the body. These tests pin that the +# sentinel resolution still reads the LIVE ``headless_authoring`` module +# attribute at call time (the same or better patchability than the old +# baked default, which could never observe a monkeypatch applied after +# the sibling module was first imported). + + +@pytest.mark.asyncio +async def test_drain_one_default_invoke_resolves_to_live_claude_invoke( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from mcp_server.handlers.consolidation import headless_authoring as ha + from mcp_server.handlers.consolidation import page_io + + calls: list[str] = [] + + async def fake_invoke(prompt: str, **_kw: Any) -> Any: + calls.append(prompt) + return ha.InvokeResult(text="filled content", cost_usd=0.0) + + async def fake_write(*_a: Any, **_kw: Any) -> dict[str, Any]: + return {} + + monkeypatch.setattr(ha, "_claude_invoke", fake_invoke) + monkeypatch.setattr(page_io, "write_governed_page", fake_write) + + page = tmp_path / "p.md" + page.write_text("_(missing — needs: What)_", encoding="utf-8") + meta = { + "curation_gaps": ["purpose"], + "domain": "d", + "kind": "reference", + "source_file_path": "x.py", + } + + # No `invoke=` kwarg: must resolve to the live ha._claude_invoke, not + # crash with AttributeError/NameError and not silently no-op. Whether + # the fill itself parses into "filled" is prompt-format detail this + # test doesn't care about — the sentinel-resolution contract is that + # `invoke` gets CALLED at all when omitted. + await ha.drain_one(page, meta, "_(missing — needs: What)_", wiki_root=tmp_path) + + assert calls, "omitting `invoke` must still call the live _claude_invoke" + + +@pytest.mark.asyncio +async def test_drain_all_gaps_on_page_default_invoke_resolves_to_live_claude_invoke( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from mcp_server.handlers.consolidation import headless_authoring as ha + from mcp_server.handlers.consolidation import page_io + + calls: list[str] = [] + + async def fake_invoke(prompt: str, **_kw: Any) -> Any: + calls.append(prompt) + return ha.InvokeResult(text="PURPOSE:\nfilled content\n", cost_usd=0.0) + + async def fake_write(*_a: Any, **_kw: Any) -> dict[str, Any]: + return {} + + monkeypatch.setattr(ha, "_claude_invoke", fake_invoke) + monkeypatch.setattr(page_io, "write_governed_page", fake_write) + + page = tmp_path / "p.md" + page.write_text("_(missing — needs: What)_", encoding="utf-8") + meta = { + "curation_gaps": ["purpose"], + "domain": "d", + "kind": "reference", + "source_file_path": "x.py", + } + + results = await ha.drain_all_gaps_on_page( + page, meta, "_(missing — needs: What)_", wiki_root=tmp_path + ) + + # As above, only the sentinel-resolution contract (invoke gets called) + # is under test here — not the prompt/response parsing format. + assert calls, "omitting `invoke` must still call the live _claude_invoke" + assert results, "a page with a live gap must produce at least one result" + + +def _make_fake_collect_anchor_candidates( + tmp_path: Path, ha: Any +) -> tuple[Any, list[int | None]]: + """Anchor-candidate stub; returns it plus the ``max_drains`` it saw.""" + anchor_max_drains_seen: list[int | None] = [] + + def fake_collect(_wiki_root: Path, max_drains: int | None) -> list[Any]: + anchor_max_drains_seen.append(max_drains) + return [ + ha._AnchorCandidate( + domain="proj", + scope_name=f"scope_{i}", + scope_title=f"Scope {i}", + scope_description=f"Description {i}", + source_root=str(tmp_path), + suggested_path=f"reference/proj/scope_{i}.md", + suggested_kind="reference", + ) + for i in range(3) + ][: max_drains if max_drains is not None else 3] + + return fake_collect, anchor_max_drains_seen + + +def _make_fake_scan_pages_with_gaps(tmp_path: Path) -> Any: + """Page-scan stub offering 3 gap-bearing pages for the same test.""" + + def fake_scan(_wiki_root: Path) -> list[Any]: + pages = [] + for i in range(3): + page = tmp_path / f"p{i}.md" + page.write_text("_(missing — needs: What)_", encoding="utf-8") + pages.append( + ( + page, + { + "curation_gaps": ["purpose"], + "domain": "d", + "kind": "reference", + "source_file_path": "x.py", + }, + "_(missing — needs: What)_", + ) + ) + return pages + + return fake_scan + + +def _patch_cycle_defaults_fixtures( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ha: Any, page_io: Any +) -> tuple[list[str], list[int | None]]: + """Apply every monkeypatch the defaults-resolution test needs; return + the ``calls``/``anchor_max_drains_seen`` trackers it asserts against. + """ + calls: list[str] = [] + + async def fake_invoke(prompt: str, **_kw: Any) -> Any: + calls.append(prompt) + return ha.InvokeResult(text="content", cost_usd=0.0) + + async def fake_write(*_a: Any, **_kw: Any) -> dict[str, Any]: + return {} + + monkeypatch.setattr(ha, "_claude_invoke", fake_invoke) + monkeypatch.setattr(ha, "CORTEX_HEADLESS_MAX_FILE_DRAINS", 1) + monkeypatch.setattr(ha, "CORTEX_HEADLESS_MAX_ANCHOR_DRAINS", 1) + monkeypatch.setattr(page_io, "write_governed_page", fake_write) + + fake_collect, anchor_max_drains_seen = _make_fake_collect_anchor_candidates( + tmp_path, ha + ) + monkeypatch.setattr(ha, "_collect_anchor_candidates", fake_collect) + monkeypatch.setattr( + ha, "_scan_pages_with_gaps", _make_fake_scan_pages_with_gaps(tmp_path) + ) + return calls, anchor_max_drains_seen + + +@pytest.mark.asyncio +async def test_run_headless_authoring_cycle_defaults_resolve_from_live_module( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Omitting ``invoke``/``max_drains``/``max_anchor_drains`` must resolve + them from the live ``headless_authoring`` attributes — this is the + production call shape (``wiki_maintenance.run_headless_authoring_cycle()`` + with no arguments). + + Pins the CAP itself, not just that a value got resolved: 3 file-doc + candidates and 3 anchor candidates are offered, ``CORTEX_HEADLESS_MAX_ + FILE_DRAINS``/``..._MAX_ANCHOR_DRAINS`` are patched to 1 each, and only + 1-of-3 must be processed on each side when the caller omits both caps + — a single spare candidate would let a broken sentinel (e.g. mutating + ``if max_drains is None`` to ``is not None``, which leaves the sentinel + unresolved at ``None`` and ``list[:None]`` silently means "no cap") slip + through undetected (mutation run, 2026-07-30: this exact mutant + survived a 1-candidate version of this test). + """ + from mcp_server.handlers.consolidation import headless_authoring as ha + from mcp_server.handlers.consolidation import page_io + + calls, anchor_max_drains_seen = _patch_cycle_defaults_fixtures( + monkeypatch, tmp_path, ha, page_io + ) + + # No `invoke=`, no `max_drains=`, no `max_anchor_drains=` — the exact + # shape wiki_maintenance.py's production call uses. + summary = await ha.run_headless_authoring_cycle(tmp_path) + + assert calls, "omitted invoke must resolve to the live _claude_invoke" + assert anchor_max_drains_seen == [1], ( + "omitted max_anchor_drains must resolve to the live " + "CORTEX_HEADLESS_MAX_ANCHOR_DRAINS (1), not stay unresolved at None" + ) + assert summary.pages_scanned == 3 + assert summary.pages_with_gaps == 1, ( + "omitted max_drains must cap file-doc candidates at the live " + "CORTEX_HEADLESS_MAX_FILE_DRAINS (1), not silently process all 3" + ) diff --git a/tests_py/handlers/test_ble001_sweep_handlers.py b/tests_py/handlers/test_ble001_sweep_handlers.py index 85d91f3f..c9520596 100644 --- a/tests_py/handlers/test_ble001_sweep_handlers.py +++ b/tests_py/handlers/test_ble001_sweep_handlers.py @@ -10,11 +10,6 @@ import asyncio from types import SimpleNamespace - -# Imported before candidate_scan anywhere in this module: candidate_scan -# and headless_authoring are mutually importing; headless_authoring must -# initialize first (matching the package's own import order). -import mcp_server.handlers.consolidation.headless_authoring # noqa: F401, E402 from unittest.mock import MagicMock import pytest diff --git a/tests_py/handlers/test_headless_authoring_throttle.py b/tests_py/handlers/test_headless_authoring_throttle.py index 1b66c918..56eda9e8 100644 --- a/tests_py/handlers/test_headless_authoring_throttle.py +++ b/tests_py/handlers/test_headless_authoring_throttle.py @@ -377,12 +377,14 @@ async def test_drain_missing_anchors_skips_ungroundable( fake_repo.canonical = "test-proj" fake_registry = MagicMock() fake_registry.repos = [fake_repo] + # drain_missing_anchors lives in anchor_authoring (issue #276 — + # split out of drain_operations to stay under the size limit). monkeypatch.setattr( - "mcp_server.handlers.consolidation.drain_operations._build_registry", + "mcp_server.handlers.consolidation.anchor_authoring._build_registry", lambda: fake_registry, ) monkeypatch.setattr( - "mcp_server.handlers.consolidation.drain_operations._project_source_root", + "mcp_server.handlers.consolidation.anchor_authoring._project_source_root", lambda d: str(tmp_path) if d == "test-proj" else None, ) @@ -393,7 +395,7 @@ async def test_drain_missing_anchors_skips_ungroundable( fake_cov = MagicMock() fake_cov.scopes = [groundable_sc, ungroundable_sc] monkeypatch.setattr( - "mcp_server.handlers.consolidation.drain_operations.audit_domain", + "mcp_server.handlers.consolidation.anchor_authoring.audit_domain", lambda wiki_root, domain: fake_cov, ) diff --git a/tests_py/handlers/test_s110_sweep_handlers.py b/tests_py/handlers/test_s110_sweep_handlers.py index 096825d2..ed44a941 100644 --- a/tests_py/handlers/test_s110_sweep_handlers.py +++ b/tests_py/handlers/test_s110_sweep_handlers.py @@ -206,9 +206,8 @@ def test_stage_timestamp_failure_is_logged(self, caplog): _assert_noted("consolidation.stage_timestamp", "pool closed") def test_source_root_failure_returns_none_and_is_logged(self, caplog, monkeypatch): - # headless_authoring first: drain_operations participates in a - # deliberate load-order-sensitive circular import (see its header). - from mcp_server.handlers.consolidation import headless_authoring # noqa: F401 + # drain_operations no longer back-imports headless_authoring at + # module scope (issue #237) — importing it standalone is safe now. from mcp_server.handlers.consolidation import drain_operations from mcp_server.handlers.consolidation.drain_operations import ( _optional_source_root, @@ -227,9 +226,8 @@ def broken(domain): _assert_noted("consolidation.project_source_root", "no source map") def test_source_root_missing_domain_is_silent(self, caplog): - # headless_authoring first: drain_operations participates in a - # deliberate load-order-sensitive circular import (see its header). - from mcp_server.handlers.consolidation import headless_authoring # noqa: F401 + # drain_operations no longer back-imports headless_authoring at + # module scope (issue #237) — importing it standalone is safe now. from mcp_server.handlers.consolidation.drain_operations import ( _optional_source_root, )