diff --git a/.console/log.md b/.console/log.md index 04f54d1..05c190c 100644 --- a/.console/log.md +++ b/.console/log.md @@ -1,4 +1,42 @@ # Log +## 2026-07-16 — feat: D3 P3 two-phase write-once consequence writer (human --apply) + +New `context_engine/attribution_apply.py` (460 lines; the `_load` sibling +loader is imported from attribution.py, not cloned — custodian D11 flagged +the copy on first push): the P3 writer that +dispatches the P2 plan into cold-item frontmatter — `apply_attribution(root, +plan, ...)` → `AppliedResult{applied, flipped, skipped}` (auditable, +machine-readable reasons). Two phases, both MONOTONE and RE-CHECKED at write +time (the plan may be stale; disk is re-read per item): Phase A re-loads each +planned slug fresh and skips unless still cold-tier AND `acted_on_commit` not +already a real sha (write-once at write time, not just plan time), then +`dataclasses.replace` + `cold.write_item` (atomic whole-file rewrite) with the +planned `tests_green` VERBATIM — a planned "unknown" is written "unknown", +never upgraded; nonstandard degrades to "unknown", never True. Phase B scans +disk for cold items with real sha + tests_green=="unknown" (incl. fresh +Phase-A writes — an immediate §3.5 arrival) and flips ONLY on a literal +True/False from the injectable CI seam; "unknown"→True|False at most once, +True/False never touched, no re-flip (guarded in code + tests). Never raises: +per-item failures recorded, batch continues. Phase-B identity: NO argv secret, +NO env reads — injectable providers default to `gh auth token` (gh = the auth +source of truth) and `git remote get-url origin` parsed to owner/repo +(https/ssh/scp forms); any failure ⇒ None ⇒ CI resolves "unknown" ⇒ +safe-inert. Wired INSIDE the existing reviewed `--apply` boundary: +`plan_consolidation` gains additive `attribution_runner=None` (default = every +existing caller unchanged) + `ConsolidationPlan.attribution`; the CLI passes +`run_attribution` — dry-run now RENDERS the attribution plan + PENDING-CI-FLIP +worklist for the human reviewer (zero writes, fail-soft: attribution failure +never breaks the consolidation output; verified in a scaffolded-consumer sim +where attribution modules are absent), and under `--apply` the writer runs +FIRST, before the cold index loads, so `gate_promotions` sees fresh +consequences in the SAME pass (§4.3) — apply-then-gate covered both ways +(Phase-A True promotes; Phase-B flip promotes). The plan path stays +token-less (Phase A writes verbatim-as-reviewed; Phase B owns CI). NO +autonomous apply: P4 is an explicit operator decision — no new CLI, no new +mutation surface outside the human interlock. New +`tests/test_attribution_apply.py` (33 tests, all seams faked). Full suite +498 pass; ruff clean. + ## 2026-07-16 — feat: D3 P2 pure attribution planner (Context-Used trailer → consequence plan) New `context_engine/attribution.py`: `plan_attribution(root, ...)` → diff --git a/src/context_lifecycle/context_engine/attribution_apply.py b/src/context_lifecycle/context_engine/attribution_apply.py new file mode 100644 index 0000000..5cc3046 --- /dev/null +++ b/src/context_lifecycle/context_engine/attribution_apply.py @@ -0,0 +1,460 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 ProtocolWarden +"""D3 P3 — two-phase, write-once consequence WRITER (human ``--apply`` only). + +Dispatches a P2 :class:`attribution.AttributionPlan` into cold-item +frontmatter via ``dataclasses.replace`` + ``cold.write_item`` (the single +writer — every write is an atomic whole-file rewrite). This module runs ONLY +inside the reviewed ``consolidate.py --apply`` boundary (spec §4.5): the human +interlock stands; there is deliberately NO CLI here — a standalone entrypoint +would be a second, unreviewed mutation surface. Autonomous apply is P4, an +explicit operator decision, NOT this module. + +Two phases (spec §3.5), both MONOTONE and re-checked at write time (the plan +may be stale — disk is re-read per item, never trusted from plan time): + + **Phase A — new attributions.** For each planned ``(slug, sha, tests_green)``: + re-load the item fresh from disk; skip (with a recorded reason) unless it is + still cold-tier AND ``acted_on_commit`` is not already a real sha (§4.4 + write-once, enforced at write time, not just plan time). Then write + ``consequence.acted_on_commit=`` and ``consequence.tests_green`` EXACTLY + as planned — verbatim ``True | False | "unknown"``; a planned ``"unknown"`` + is written as ``"unknown"``, never upgraded, and any nonstandard planned + value degrades to ``"unknown"``, NEVER to ``True``. + + **Phase B — CI flip for attributed items.** For each cold item on disk with + a real ``acted_on_commit`` sha AND ``tests_green == "unknown"`` (including + items Phase A just wrote — an immediate CI verdict is just an early §3.5 + "arrival"): resolve CI for THAT sha through an injectable seam (default: + ``attribution._default_resolve_tests_green`` over ``ci_status``, fail-closed). + Only a literal ``True`` / ``False`` is ever written; anything else leaves the + item untouched. The transition is ``"unknown" -> True|False`` at most once: + ``True``/``False`` items never enter the worklist and are re-checked before + the write, so NEVER ``True -> anything``, NEVER ``False -> anything``, never + a re-flip. No path defaults to ``True``. + +Token/repo identity for Phase B in the human-run CLI: NO secret on argv and NO +environment reads. ``token`` comes from an injectable provider whose default +shells to ``gh auth token`` (gh is this repo's auth source of truth — +``ci_status`` already queries through ``gh``); ``github_repo`` from a provider +that parses ``git remote get-url origin`` (https/ssh/scp forms). Both are +bounded subprocesses; ANY failure ⇒ ``None`` ⇒ the CI seam resolves +``"unknown"`` ⇒ Phase B leaves items inert (safe — ``"unknown"`` still fails +the promotion gate). + +Guarantees (the house apply idiom, ``consolidate.plan_consolidation``): the +apply path may write but NEVER raises to the caller — every per-item failure +is recorded and the batch continues; the returned :class:`AppliedResult` lists +applied / flipped / skipped with machine-readable reasons (auditable). + +``run_attribution`` is the one integration entrypoint ``consolidate`` calls: +dry-run computes the plan + the pending Phase-B worklist for HUMAN visibility +(zero writes); ``apply=True`` runs both phases. Fail-soft throughout — in a +scaffolded consumer where the attribution modules are absent, everything +degrades to "no attribution", never breaking the consolidation pass. +""" + +from __future__ import annotations + +import importlib.util +import re +import subprocess +import sys +from dataclasses import dataclass, field, replace +from pathlib import Path +from typing import Callable + +# The verbatim sentinel cold.py stores when CI cannot be resolved (matches +# ci_status.UNKNOWN / attribution.UNKNOWN; duplicated so this module loads +# standalone next to cold/route). +UNKNOWN = "unknown" + +# Machine-readable skip reasons (write-time re-checks + per-item failures). +SKIP_ITEM_MISSING = "item_missing" +SKIP_NOT_COLD_TIER = "not_cold_tier" +SKIP_ALREADY_ATTRIBUTED = "already_attributed" +SKIP_ALREADY_FLIPPED = "already_flipped" +SKIP_WRITE_FAILED = "write_failed" +SKIP_CI_UNRESOLVED = "ci_unresolved" +SKIP_FLIP_FAILED = "flip_failed" + +# Bounded so a slow git/gh never hangs the human's --apply run (mirrors +# attribution._GIT_TIMEOUT_SECONDS). +_SUBPROCESS_TIMEOUT_SECONDS = 30 + +# The same sha shape-check the promotion gate applies (consolidate._is_real_sha). +_SHA_RE = re.compile(r"^[0-9a-f]{7,40}$") + +# origin URL forms -> owner/repo: https://host/owner/repo(.git), +# ssh://git@host/owner/repo(.git), git@host:owner/repo(.git). +_REMOTE_RE = re.compile( + r"^(?:[a-z+]+://[^/]+/|[^/@]+@[^/:]+:)([^/:]+)/([^/]+?)(?:\.git)?/?$" +) + + +# The sibling-module loader is REUSED from attribution.py, not cloned (the +# dual-mode import-shim body is identical by necessity and custodian D11 +# rightly flags copies): package/standalone import first, file-location +# fallback — the same dual-mode discipline as cold.py's route import. +try: # pragma: no cover - import shim + from attribution import _load +except ImportError: # pragma: no cover + _p = Path(__file__).resolve().parent / "attribution.py" + _spec = importlib.util.spec_from_file_location("cl_attribution", _p) + assert _spec and _spec.loader + _mod = importlib.util.module_from_spec(_spec) + sys.modules.setdefault("cl_attribution", _mod) + _spec.loader.exec_module(_mod) + _load = _mod._load + + +# ---- Result shapes — mirror attribution.AttributionPlan's audit posture. ---- # + + +@dataclass(frozen=True) +class AppliedWrite: + """One consequence write that happened: Phase A (``applied``) or Phase B + (``flipped``). ``tests_green`` is the verbatim value now on disk. + """ + + slug: str + sha: str + tests_green: object + + +@dataclass(frozen=True) +class SkippedWrite: + """One planned/pending write that did NOT happen, with the reason code.""" + + slug: str + reason: str + + +@dataclass(frozen=True) +class PendingFlip: + """One Phase-B worklist entry: a cold item already carrying a real + ``acted_on_commit`` sha whose ``tests_green`` is still ``"unknown"``. + """ + + slug: str + sha: str + + +@dataclass(frozen=True) +class AppliedResult: + applied: list[AppliedWrite] = field(default_factory=list) + flipped: list[AppliedWrite] = field(default_factory=list) + skipped: list[SkippedWrite] = field(default_factory=list) + + def is_empty(self) -> bool: + return not self.applied and not self.flipped and not self.skipped + + +@dataclass(frozen=True) +class AttributionOutcome: + """What ``consolidate`` renders/carries: the P2 plan (None when the planner + is unavailable/failed), the apply result (None on dry-run), and the + Phase-B worklist that remains pending on disk AFTER any writes. + """ + + plan: object | None = None + applied: object | None = None + pending: list[PendingFlip] = field(default_factory=list) + + +# ---- Injectable identity providers (no argv secrets, no env reads). -------- # + + +def _default_token_provider() -> str | None: + """Obtain a GitHub token from ``gh auth token`` (bounded subprocess). + + ``gh`` is the auth source of truth in this repo (ci_status queries through + it); this keeps the secret OFF argv and out of any env read of our own. + Any failure — gh missing, not logged in, timeout — ⇒ None (Phase B stays + safe-inert at ``"unknown"``). + """ + try: + proc = subprocess.run( # noqa: S603 — fixed argv, no shell + ["gh", "auth", "token"], + capture_output=True, + text=True, + timeout=_SUBPROCESS_TIMEOUT_SECONDS, + ) + token = proc.stdout.strip() + if proc.returncode != 0 or not token: + return None + return token + except Exception: + return None + + +def _parse_remote(url: str) -> str | None: + """Parse an origin URL to ``owner/repo`` (https / ssh:// / scp-style), or + None on any unrecognized shape — fail-closed, never a guess. + """ + m = _REMOTE_RE.match(url.strip()) + if not m: + return None + return f"{m.group(1)}/{m.group(2)}" + + +def _default_repo_provider(root: Path) -> str | None: + """Derive ``owner/repo`` from ``git remote get-url origin`` in ``root`` + (bounded subprocess). Any failure ⇒ None ⇒ CI resolves ``"unknown"``. + """ + try: + proc = subprocess.run( # noqa: S603 — fixed argv, no shell + ["git", "remote", "get-url", "origin"], + cwd=root, + capture_output=True, + text=True, + timeout=_SUBPROCESS_TIMEOUT_SECONDS, + ) + if proc.returncode != 0: + return None + return _parse_remote(proc.stdout.strip()) + except Exception: + return None + + +def _default_resolve_tests_green( + github_repo: str | None, sha: str, token: str | None +) -> object: + """Default CI seam: delegate to the P2 planner's fail-closed resolver + (``ci_status`` underneath). Either identity absent ⇒ ``"unknown"``. + """ + attribution = _load("attribution") + return attribution._default_resolve_tests_green(github_repo, sha, token) + + +# ---- Pure readers. ---------------------------------------------------------- # + + +def _real_sha(value: object) -> str | None: + sha = str(value or "").strip() + return sha if _SHA_RE.match(sha) else None + + +def pending_flips(root: Path) -> list[PendingFlip]: + """The Phase-B worklist, read fresh from disk: cold-tier items whose + ``acted_on_commit`` is a real sha and whose ``tests_green`` is still the + verbatim ``"unknown"``. ``True``/``False`` items never appear here — that + exclusion IS the first monotonicity guard (never re-flip a resolved value). + Pure reader; [] on any failure. + """ + try: + cold = _load("cold") + out: list[PendingFlip] = [] + for item in cold.load_index(root / ".context" / "knowledge"): + if item.tier != "cold" or item.tests_green != UNKNOWN: + continue + sha = _real_sha(item.acted_on_commit) + if sha is None: + continue + out.append(PendingFlip(item.slug, sha)) + return out + except Exception: + return [] + + +# --------------------------------------------------------------------------- # +# The writer. # +# --------------------------------------------------------------------------- # + + +def apply_attribution( + root: Path, + plan, + *, + github_repo: str | None = None, + token: str | None = None, + repo_provider: Callable[[Path], str | None] = _default_repo_provider, + token_provider: Callable[[], str | None] = _default_token_provider, + resolve_tests_green: Callable[ + [str | None, str, str | None], object + ] = _default_resolve_tests_green, +) -> AppliedResult: + """Dispatch ``plan`` (Phase A), then flip resolvable pending items + (Phase B). MUTATES cold-item files — call this ONLY from inside the + reviewed ``--apply`` boundary. NEVER raises: every per-item failure is + recorded in ``skipped`` and the batch continues. + + ``github_repo``/``token`` may be supplied directly (tests, wired callers); + when None they are resolved lazily through the injectable providers — and + only if there is at least one pending flip, so a plain Phase-A-only pass + never spawns a subprocess. + """ + try: + cold = _load("cold") + knowledge_dir = root / ".context" / "knowledge" + applied: list[AppliedWrite] = [] + flipped: list[AppliedWrite] = [] + skipped: list[SkippedWrite] = [] + + # ---- Phase A: planned attributions, re-checked at write time. ---- + for planned in getattr(plan, "attributions", None) or []: + slug = planned.slug + try: + item = cold.parse_item(knowledge_dir / f"{slug}.md") + if item is None: + skipped.append(SkippedWrite(slug, SKIP_ITEM_MISSING)) + continue + if item.tier != "cold": + skipped.append(SkippedWrite(slug, SKIP_NOT_COLD_TIER)) + continue + if _real_sha(item.acted_on_commit) is not None: + skipped.append(SkippedWrite(slug, SKIP_ALREADY_ATTRIBUTED)) + continue + tests_green = planned.tests_green + # Verbatim contract: True | False | "unknown" only; a planned + # "unknown" is written as "unknown" (never upgraded here — the + # flip is Phase B's job); nonstandard degrades to "unknown". + if tests_green is not True and tests_green is not False: + tests_green = UNKNOWN + cold.write_item( + knowledge_dir, + replace( + item, + acted_on_commit=planned.sha, + tests_green=tests_green, + ), + ) + applied.append(AppliedWrite(slug, planned.sha, tests_green)) + except Exception: + skipped.append(SkippedWrite(slug, SKIP_WRITE_FAILED)) + continue + + # ---- Phase B: CI flips for attributed-but-unknown items. ---- + worklist = pending_flips(root) + if worklist: + if github_repo is None: + try: + github_repo = repo_provider(root) + except Exception: + github_repo = None + if token is None: + try: + token = token_provider() + except Exception: + token = None + for pending in worklist: + try: + # Monotone, re-checked at write time: only a fresh, still-cold, + # still-"unknown", still-same-shaped item is ever flipped. + item = cold.parse_item(knowledge_dir / f"{pending.slug}.md") + if item is None: + skipped.append(SkippedWrite(pending.slug, SKIP_ITEM_MISSING)) + continue + if item.tier != "cold": + skipped.append(SkippedWrite(pending.slug, SKIP_NOT_COLD_TIER)) + continue + if item.tests_green != UNKNOWN: + skipped.append(SkippedWrite(pending.slug, SKIP_ALREADY_FLIPPED)) + continue + sha = _real_sha(item.acted_on_commit) + if sha is None: + skipped.append(SkippedWrite(pending.slug, SKIP_CI_UNRESOLVED)) + continue + verdict = resolve_tests_green(github_repo, sha, token) + # Literal True/False ONLY ever gets written; "unknown" or any + # nonstandard value leaves the item untouched (never default + # True — §3.5). + if verdict is not True and verdict is not False: + skipped.append(SkippedWrite(pending.slug, SKIP_CI_UNRESOLVED)) + continue + cold.write_item(knowledge_dir, replace(item, tests_green=verdict)) + flipped.append(AppliedWrite(pending.slug, sha, verdict)) + except Exception: + skipped.append(SkippedWrite(pending.slug, SKIP_FLIP_FAILED)) + continue + + return AppliedResult(applied=applied, flipped=flipped, skipped=skipped) + except Exception: + # Belt-and-suspenders: the apply path never raises to its caller. + return AppliedResult() + + +# --------------------------------------------------------------------------- # +# The consolidate integration entrypoint + renderer. # +# --------------------------------------------------------------------------- # + + +def run_attribution( + root: Path, + *, + apply: bool = False, + default_branch: str = "main", + plan_fn: Callable[[Path], object] | None = None, + apply_fn: Callable[[Path, object], AppliedResult] | None = None, +) -> AttributionOutcome: + """The one call ``consolidate`` makes, in BOTH modes. + + Dry-run (``apply=False``): compute the P2 plan + the pending Phase-B + worklist for human review — ZERO writes. ``apply=True``: dispatch the plan + through :func:`apply_attribution` (both phases), then report what remains + pending. The plan is always computed WITHOUT a token (``tests_green`` + plans as ``"unknown"``): CI resolution is Phase B's job under the gh-auth + seam, so the reviewed dry-run plan and the applied Phase-A writes match + verbatim. Never raises; total failure ⇒ an empty outcome. + """ + try: + if plan_fn is None: + attribution = _load("attribution") + + def plan_fn(r: Path): + # No token/repo on the plan path: a secret does not belong on + # argv (the P2 CLI choice) and Phase A writes verbatim-as- + # reviewed; Phase B owns CI. + return attribution.plan_attribution( + r, default_branch=default_branch, github_repo=None, token=None + ) + + try: + plan = plan_fn(root) + except Exception: + plan = None + + applied = None + if apply and plan is not None: + applied = (apply_fn or apply_attribution)(root, plan) + + return AttributionOutcome( + plan=plan, applied=applied, pending=pending_flips(root) + ) + except Exception: + return AttributionOutcome() + + +def render_outcome(outcome: AttributionOutcome) -> str: + """Render an outcome for the reviewed plan output (consolidate._render). + + Dry-run visibility is the point: the human sees exactly what ``--apply`` + would write (the plan) plus which attributed items are awaiting a CI flip. + Returns '' when there is nothing attribution-related to show, so the + existing consolidation output is unchanged. Never raises. + """ + try: + lines: list[str] = [] + plan = outcome.plan + if plan is not None and not plan.is_empty(): + attribution = _load("attribution") + lines.append(attribution._render(plan)) + for pending in outcome.pending: + lines.append(f"PENDING-CI-FLIP {pending.slug} @ {pending.sha}") + result = outcome.applied + if result is not None: + for write in result.applied: + lines.append( + f"ATTRIBUTED {write.slug} -> {write.sha} " + f"(tests_green={write.tests_green!r})" + ) + for write in result.flipped: + lines.append( + f"FLIPPED {write.slug} tests_green={write.tests_green!r} " + f"@ {write.sha}" + ) + for skip in result.skipped: + lines.append(f" skip {skip.slug} ({skip.reason})") + return "\n".join(lines) + except Exception: + return "" diff --git a/src/context_lifecycle/context_engine/consolidate.py b/src/context_lifecycle/context_engine/consolidate.py index 934ccdf..c6eb76a 100644 --- a/src/context_lifecycle/context_engine/consolidate.py +++ b/src/context_lifecycle/context_engine/consolidate.py @@ -95,6 +95,10 @@ class ConsolidationPlan: prunes: list[PlannedAction] = field(default_factory=list) rejections: list[PlannedAction] = field(default_factory=list) keeps: list[PlannedAction] = field(default_factory=list) + # D3 P3: the attribution outcome carried alongside the consolidation + # actions (attribution_apply.AttributionOutcome | None). Additive with a + # safe default so every existing constructor/caller is untouched. + attribution: object | None = None def all_actions(self) -> list[PlannedAction]: return ( @@ -337,13 +341,27 @@ def decay_warm( def plan_consolidation( - root: Path, campaign_id: str | None = None, *, apply: bool = False + root: Path, + campaign_id: str | None = None, + *, + apply: bool = False, + attribution_runner=None, ) -> ConsolidationPlan: """Compute (and, with apply=True, execute) the campaign-close pass. Always builds the ConsolidationPlan FIRST via pure readers (zero writes). With apply=True, dispatches each action to its mutator with per-action isolation. Returns an empty plan on any top-level failure (never raises). + + ``attribution_runner`` (D3 P3, additive — default None keeps every + existing caller byte-for-byte unchanged) is a ``(root, apply) -> outcome`` + callable (attribution_apply.run_attribution's shape). Dry-run it is a pure + reader whose outcome is rendered for the human reviewer; under apply=True + it runs the consequence WRITER (Phase A attributions + Phase B CI flips) + FIRST, before the cold index is loaded, so gate_promotions sees the + freshly-written consequences in this SAME pass (D3 spec §4.3). Fail-soft: + a failing runner degrades to no-attribution and never breaks the + consolidation plan/output. """ try: cold = _load("cold") @@ -356,6 +374,13 @@ def plan_consolidation( sessions_dir = context_dir / "sessions" archived_dir = context_dir / "archived" + attribution_outcome = None + if attribution_runner is not None: + try: + attribution_outcome = attribution_runner(root, apply) + except Exception: + attribution_outcome = None + items = cold.load_index(knowledge_dir) # Scope to a campaign_id when one is given (select that campaign's cold @@ -416,6 +441,7 @@ def plan_consolidation( prunes=prunes, rejections=rejections, keeps=keeps, + attribution=attribution_outcome, ) if not apply: @@ -502,6 +528,17 @@ def _today() -> str: def _render(plan: ConsolidationPlan) -> str: lines: list[str] = [] + # D3 P3 visibility: the attribution plan / pending CI flips / applied + # writes are shown FIRST — they mirror execution order under --apply + # (writer before gate). Fail-soft: a render failure must never break the + # existing consolidation output. + if plan.attribution is not None: + try: + extra = _load("attribution_apply").render_outcome(plan.attribution) + if extra: + lines.append(extra) + except Exception: + pass for a in plan.promotions: lines.append(f"PROMOTE {a.target} ({a.reason})") for a in plan.rejections: @@ -517,6 +554,16 @@ def _render(plan: ConsolidationPlan) -> str: return "\n".join(lines) +def _attribution_runner(root: Path, apply: bool): + """CLI wiring for the D3 P3 consequence writer, INSIDE this reviewed + boundary: dry-run computes the plan + pending CI flips for the human; + --apply runs the two-phase writer first (attribution_apply docstring). + Lazily loaded so a scaffolded consumer without the attribution modules + degrades to no-attribution (the caller's try/except catches the load). + """ + return _load("attribution_apply").run_attribution(root, apply=apply) + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Phase 5 consolidation engine") parser.add_argument("--root", default=".", help="repo root (CL_ANCHOR)") @@ -529,7 +576,10 @@ def main(argv: list[str] | None = None) -> int: args = parser.parse_args(argv) try: plan = plan_consolidation( - Path(args.root), args.campaign, apply=args.apply + Path(args.root), + args.campaign, + apply=args.apply, + attribution_runner=_attribution_runner, ) sys.stdout.write(_render(plan) + "\n") except Exception: # never blocks: any failure -> nothing actionable, exit 0 diff --git a/tests/test_attribution_apply.py b/tests/test_attribution_apply.py new file mode 100644 index 0000000..b1c66fb --- /dev/null +++ b/tests/test_attribution_apply.py @@ -0,0 +1,602 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 ProtocolWarden +"""Tests for the D3 P3 two-phase consequence writer (attribution_apply). + +Every seam is faked (no git, no gh, no network). Covered: + + - Phase A happy path: frontmatter after == planned values VERBATIM (True, + False, and a planned "unknown" written as "unknown" — never upgraded; + nonstandard planned values degrade to "unknown", never True); + - write-once RE-CHECKED at write time (stale plan): item attributed on disk + between plan and apply ⇒ skipped; item promoted to warm ⇒ skipped; item + deleted ⇒ skipped; + - Phase B: "unknown" -> True and "unknown" -> False flips; unresolved / + nonstandard seam values leave "unknown"; provider (token/repo) failure ⇒ + Phase B no-op (safe-inert); + - MONOTONE guards: True stays True, False stays False, no re-flip — even + when the seam contradicts the resolved value; + - per-item isolation: a seam raising mid-batch records that item and the + rest still process; a write failure is recorded, never raised; + - run_attribution: dry-run is byte-pure and carries plan + pending flips; + plan failure degrades to an empty outcome; + - consolidate integration: dry-run RENDERS the attribution plan + pending + Phase-B flips with zero writes; a failing runner never breaks the + consolidation output; and apply-then-gate — a Phase-A True write and a + Phase-B flip each pass gate_promotions in the SAME + plan_consolidation(apply=True) pass. +""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +from context_lifecycle.context_engine import attribution_apply, cold, consolidate +from context_lifecycle.context_engine.attribution import ( + AttributionPlan, + PlannedAttribution, +) +from context_lifecycle.context_engine.attribution_apply import ( + SKIP_ALREADY_ATTRIBUTED, + SKIP_CI_UNRESOLVED, + SKIP_FLIP_FAILED, + SKIP_ITEM_MISSING, + SKIP_NOT_COLD_TIER, + SKIP_WRITE_FAILED, + AttributionOutcome, + PendingFlip, + apply_attribution, + pending_flips, + run_attribution, +) + +SHA_A = "a" * 40 +SHA_B = "b" * 40 +SHA_C = "c" * 40 + +ROUTES_YAML = """engine_compat: ">=0.2 <0.3" +budget: + max_docs_per_edit: 3 +routes: + - match: "src/platform_manifest/loader.py" + inject: ["docs/inject/loader.md"] + priority: 10 +""" + + +# --------------------------------------------------------------------------- # +# fixtures / helpers # +# --------------------------------------------------------------------------- # + + +def _item(slug: str, **over) -> cold.ColdItem: + base = dict( + topic=over.pop("topic", slug), + paths=("src/**",), + created="2026-07-01", + campaign_id="c-2026-07-01-test", + acted_on_commit=None, + tests_green="unknown", + tier="cold", + pinned=False, + last_injected=None, + finding=f"Finding for {slug}.", + ) + base.update(over) + base["slug"] = slug + return cold.ColdItem(**base) + + +def _put(root: Path, slug: str, **over) -> Path: + return cold.write_item(root / ".context" / "knowledge", _item(slug, **over)) + + +def _get(root: Path, slug: str) -> cold.ColdItem | None: + return cold.parse_item(root / ".context" / "knowledge" / f"{slug}.md") + + +def _plan(*attributions: PlannedAttribution) -> AttributionPlan: + return AttributionPlan(attributions=list(attributions)) + + +def _ci(value: object): + return lambda repo, sha, token: value + + +def _apply(root: Path, plan=None, *, ci: object = "unknown", **kw): + kw.setdefault("github_repo", "o/r") + kw.setdefault("token", "t") + kw.setdefault("resolve_tests_green", ci if callable(ci) else _ci(ci)) + return apply_attribution(root, plan if plan is not None else _plan(), **kw) + + +def _skips(result) -> set[tuple[str, str]]: + return {(s.slug, s.reason) for s in result.skipped} + + +def _snapshot(root: Path) -> dict[str, bytes]: + return { + str(p.relative_to(root)): p.read_bytes() + for p in sorted(root.rglob("*")) + if p.is_file() + } + + +# --------------------------------------------------------------------------- # +# Phase A — verbatim writes, re-checked write-once # +# --------------------------------------------------------------------------- # + + +def test_phase_a_writes_planned_true_verbatim(tmp_path): + _put(tmp_path, "s1") + result = _apply(tmp_path, _plan(PlannedAttribution("s1", SHA_A, True))) + assert [(w.slug, w.sha, w.tests_green) for w in result.applied] == [ + ("s1", SHA_A, True) + ] + item = _get(tmp_path, "s1") + assert item.acted_on_commit == SHA_A + assert item.tests_green is True + text = (tmp_path / ".context/knowledge/s1.md").read_text(encoding="utf-8") + assert f"acted_on_commit: {SHA_A}" in text + assert "tests_green: true" in text + + +def test_phase_a_writes_planned_false_verbatim(tmp_path): + _put(tmp_path, "s1") + result = _apply(tmp_path, _plan(PlannedAttribution("s1", SHA_A, False))) + assert result.applied[0].tests_green is False + assert _get(tmp_path, "s1").tests_green is False + + +def test_phase_a_planned_unknown_written_as_unknown_never_upgraded(tmp_path): + _put(tmp_path, "s1") + result = _apply( + tmp_path, _plan(PlannedAttribution("s1", SHA_A, "unknown")), ci="unknown" + ) + assert result.applied[0].tests_green == "unknown" + item = _get(tmp_path, "s1") + assert item.acted_on_commit == SHA_A + assert item.tests_green == "unknown" + text = (tmp_path / ".context/knowledge/s1.md").read_text(encoding="utf-8") + assert "tests_green: unknown" in text + + +def test_phase_a_nonstandard_planned_value_degrades_to_unknown(tmp_path): + for i, weird in enumerate(("true", 1, None, "green")): + slug = f"s{i}" + _put(tmp_path, slug) + result = _apply( + tmp_path, _plan(PlannedAttribution(slug, SHA_A, weird)), ci="unknown" + ) + assert result.applied[-1].tests_green == "unknown" + assert _get(tmp_path, slug).tests_green == "unknown" + + +def test_stale_plan_item_attributed_on_disk_is_skipped(tmp_path): + # Attributed (to a DIFFERENT sha) between plan time and apply time: + # write-once is re-checked at write time — first attribution wins. + _put(tmp_path, "s1", acted_on_commit=SHA_B) + before = _snapshot(tmp_path) + result = _apply(tmp_path, _plan(PlannedAttribution("s1", SHA_A, True))) + assert result.applied == [] + assert ("s1", SKIP_ALREADY_ATTRIBUTED) in _skips(result) + assert _get(tmp_path, "s1").acted_on_commit == SHA_B + assert _snapshot(tmp_path) == before + + +def test_stale_plan_item_promoted_to_warm_is_skipped(tmp_path): + _put(tmp_path, "s1", tier="warm") + before = _snapshot(tmp_path) + result = _apply(tmp_path, _plan(PlannedAttribution("s1", SHA_A, True))) + assert result.applied == [] + assert ("s1", SKIP_NOT_COLD_TIER) in _skips(result) + assert _snapshot(tmp_path) == before + + +def test_stale_plan_item_deleted_is_skipped(tmp_path): + (tmp_path / ".context" / "knowledge").mkdir(parents=True) + result = _apply(tmp_path, _plan(PlannedAttribution("gone", SHA_A, True))) + assert result.applied == [] + assert ("gone", SKIP_ITEM_MISSING) in _skips(result) + + +def test_phase_a_write_failure_recorded_never_raised(tmp_path): + p1 = _put(tmp_path, "s1") + _put(tmp_path, "s2") + p1.chmod(0o444) # reads fine, the whole-file rewrite fails + try: + result = _apply( + tmp_path, + _plan( + PlannedAttribution("s1", SHA_A, True), + PlannedAttribution("s2", SHA_B, True), + ), + ) + finally: + p1.chmod(0o644) + # s1's failure is recorded; the batch continued and s2 still applied. + assert ("s1", SKIP_WRITE_FAILED) in _skips(result) + assert [(w.slug, w.sha) for w in result.applied] == [("s2", SHA_B)] + assert _get(tmp_path, "s1").acted_on_commit is None + assert _get(tmp_path, "s2").acted_on_commit == SHA_B + + +# --------------------------------------------------------------------------- # +# Phase B — CI flips, fail-closed and monotone # +# --------------------------------------------------------------------------- # + + +def test_phase_b_flips_unknown_to_true(tmp_path): + _put(tmp_path, "s1", acted_on_commit=SHA_A) + result = _apply(tmp_path, ci=True) + assert [(w.slug, w.sha, w.tests_green) for w in result.flipped] == [ + ("s1", SHA_A, True) + ] + assert _get(tmp_path, "s1").tests_green is True + + +def test_phase_b_flips_unknown_to_false(tmp_path): + _put(tmp_path, "s1", acted_on_commit=SHA_A) + result = _apply(tmp_path, ci=False) + assert result.flipped[0].tests_green is False + assert _get(tmp_path, "s1").tests_green is False + + +def test_phase_b_unresolved_leaves_unknown(tmp_path): + _put(tmp_path, "s1", acted_on_commit=SHA_A) + before = _snapshot(tmp_path) + result = _apply(tmp_path, ci="unknown") + assert result.flipped == [] + assert ("s1", SKIP_CI_UNRESOLVED) in _skips(result) + assert _snapshot(tmp_path) == before + + +def test_phase_b_nonstandard_seam_value_never_written(tmp_path): + # Only LITERAL True/False are ever written — truthy imposters are not. + _put(tmp_path, "s1", acted_on_commit=SHA_A) + before = _snapshot(tmp_path) + for weird in ("true", 1, None, "green", "false", 0): + result = _apply(tmp_path, ci=weird) + assert result.flipped == [] + assert ("s1", SKIP_CI_UNRESOLVED) in _skips(result) + assert _snapshot(tmp_path) == before + + +def test_monotone_true_stays_true(tmp_path): + _put(tmp_path, "s1", acted_on_commit=SHA_A, tests_green=True) + before = _snapshot(tmp_path) + result = _apply(tmp_path, ci=False) # a contradicting seam changes nothing + assert result.flipped == [] and result.applied == [] and result.skipped == [] + assert _snapshot(tmp_path) == before + + +def test_monotone_false_stays_false(tmp_path): + _put(tmp_path, "s1", acted_on_commit=SHA_A, tests_green=False) + before = _snapshot(tmp_path) + result = _apply(tmp_path, ci=True) + assert result.flipped == [] + assert _snapshot(tmp_path) == before + + +def test_monotone_no_reflip_across_passes(tmp_path): + # "unknown" -> True|False happens AT MOST ONCE: a second pass with a + # contradicting seam is a no-op. + _put(tmp_path, "s1", acted_on_commit=SHA_A) + assert _apply(tmp_path, ci=True).flipped != [] + after_flip = _snapshot(tmp_path) + result = _apply(tmp_path, ci=False) + assert result.flipped == [] + assert _snapshot(tmp_path) == after_flip + assert _get(tmp_path, "s1").tests_green is True + + +def test_phase_b_seam_raising_mid_batch_isolated(tmp_path): + _put(tmp_path, "s1", acted_on_commit=SHA_A) + _put(tmp_path, "s2", acted_on_commit=SHA_B) + + def seam(repo, sha, token): + if sha == SHA_A: + raise RuntimeError("ci exploded") + return True + + result = _apply(tmp_path, ci=seam) + assert ("s1", SKIP_FLIP_FAILED) in _skips(result) + assert [(w.slug, w.tests_green) for w in result.flipped] == [("s2", True)] + assert _get(tmp_path, "s1").tests_green == "unknown" + assert _get(tmp_path, "s2").tests_green is True + + +def test_provider_failures_leave_phase_b_inert(tmp_path): + # No token / no repo identity ⇒ the default fail-closed resolver answers + # "unknown" without any network attempt ⇒ nothing flips (safe-inert). + _put(tmp_path, "s1", acted_on_commit=SHA_A) + before = _snapshot(tmp_path) + for repo_p, token_p in ( + (lambda root: None, lambda: None), + (lambda root: (_ for _ in ()).throw(OSError("no git")), lambda: "t"), + (lambda root: "o/r", lambda: (_ for _ in ()).throw(OSError("no gh"))), + ): + result = apply_attribution( + tmp_path, + _plan(), + repo_provider=repo_p, + token_provider=token_p, + ) + assert result.flipped == [] + assert ("s1", SKIP_CI_UNRESOLVED) in _skips(result) + assert _snapshot(tmp_path) == before + + +def test_providers_not_consulted_without_pending_flips(tmp_path): + _put(tmp_path, "s1") # unattributed: Phase B worklist is empty + calls: list[str] = [] + result = apply_attribution( + tmp_path, + _plan(), + repo_provider=lambda root: calls.append("repo"), + token_provider=lambda: calls.append("token"), + ) + assert calls == [] + assert result.is_empty() + + +def test_phase_a_fresh_unknown_is_phase_b_eligible_same_pass(tmp_path): + # A Phase-A write of sha+"unknown" whose CI verdict is ALREADY resolvable + # flips in the same apply pass — an immediate §3.5 "arrival". + _put(tmp_path, "s1") + result = _apply( + tmp_path, _plan(PlannedAttribution("s1", SHA_A, "unknown")), ci=True + ) + assert result.applied[0].tests_green == "unknown" # Phase A verbatim + assert result.flipped[0].tests_green is True # Phase B flip + assert _get(tmp_path, "s1").tests_green is True + + +def test_pending_flips_lists_only_attributed_unknown_cold_items(tmp_path): + _put(tmp_path, "pend", acted_on_commit=SHA_A) + _put(tmp_path, "unattributed") + _put(tmp_path, "resolved", acted_on_commit=SHA_B, tests_green=True) + _put(tmp_path, "warm", tier="warm", acted_on_commit=SHA_C) + _put(tmp_path, "fake", acted_on_commit="see-pr") + assert pending_flips(tmp_path) == [PendingFlip("pend", SHA_A)] + assert pending_flips(tmp_path / "nowhere") == [] + + +# --------------------------------------------------------------------------- # +# default identity providers (subprocess seams faked) # +# --------------------------------------------------------------------------- # + + +def test_parse_remote_forms(): + for url in ( + "https://github.com/o/r", + "https://github.com/o/r.git", + "https://github.com/o/r/", + "ssh://git@github.com/o/r.git", + "git@github.com:o/r.git", + "git@github.com:o/r", + ): + assert attribution_apply._parse_remote(url) == "o/r" + for bad in ("", "not a url", "https://github.com/only-owner", "o/r"): + assert attribution_apply._parse_remote(bad) is None + + +def _completed(returncode: int, stdout: str) -> subprocess.CompletedProcess: + return subprocess.CompletedProcess( + args=[], returncode=returncode, stdout=stdout, stderr="" + ) + + +def test_default_token_provider_success_and_failure(monkeypatch): + monkeypatch.setattr( + attribution_apply.subprocess, "run", lambda *a, **k: _completed(0, "tok\n") + ) + assert attribution_apply._default_token_provider() == "tok" + monkeypatch.setattr( + attribution_apply.subprocess, "run", lambda *a, **k: _completed(1, "") + ) + assert attribution_apply._default_token_provider() is None + + def boom(*a, **k): + raise FileNotFoundError("gh not installed") + + monkeypatch.setattr(attribution_apply.subprocess, "run", boom) + assert attribution_apply._default_token_provider() is None + + +def test_default_repo_provider_success_and_failure(tmp_path, monkeypatch): + monkeypatch.setattr( + attribution_apply.subprocess, + "run", + lambda *a, **k: _completed(0, "git@github.com:o/r.git\n"), + ) + assert attribution_apply._default_repo_provider(tmp_path) == "o/r" + monkeypatch.setattr( + attribution_apply.subprocess, + "run", + lambda *a, **k: _completed(128, ""), + ) + assert attribution_apply._default_repo_provider(tmp_path) is None + + +# --------------------------------------------------------------------------- # +# run_attribution — the consolidate entrypoint # +# --------------------------------------------------------------------------- # + + +def test_run_attribution_dry_run_is_pure_and_visible(tmp_path): + _put(tmp_path, "s1") + _put(tmp_path, "pend", acted_on_commit=SHA_B) + plan = _plan(PlannedAttribution("s1", SHA_A, "unknown")) + before = _snapshot(tmp_path) + outcome = run_attribution(tmp_path, apply=False, plan_fn=lambda r: plan) + assert _snapshot(tmp_path) == before # byte-identical: ZERO writes + assert outcome.plan is plan + assert outcome.applied is None + assert outcome.pending == [PendingFlip("pend", SHA_B)] + + +def test_run_attribution_apply_dispatches_and_reports_remaining(tmp_path): + _put(tmp_path, "s1") + _put(tmp_path, "pend", acted_on_commit=SHA_B) + plan = _plan(PlannedAttribution("s1", SHA_A, "unknown")) + outcome = run_attribution( + tmp_path, + apply=True, + plan_fn=lambda r: plan, + apply_fn=lambda r, p: apply_attribution( + r, p, github_repo="o/r", token="t", resolve_tests_green=_ci("unknown") + ), + ) + assert [(w.slug, w.sha) for w in outcome.applied.applied] == [("s1", SHA_A)] + # Nothing resolved, so BOTH attributed items remain pending on disk. + assert set(outcome.pending) == { + PendingFlip("pend", SHA_B), + PendingFlip("s1", SHA_A), + } + + +def test_run_attribution_plan_failure_degrades_no_apply(tmp_path): + _put(tmp_path, "pend", acted_on_commit=SHA_B) + + def boom(root): + raise RuntimeError("planner exploded") + + before = _snapshot(tmp_path) + outcome = run_attribution( + tmp_path, + apply=True, + plan_fn=boom, + apply_fn=lambda r, p: (_ for _ in ()).throw(AssertionError("must not run")), + ) + assert outcome.plan is None + assert outcome.applied is None + assert outcome.pending == [PendingFlip("pend", SHA_B)] + assert _snapshot(tmp_path) == before + + +def test_render_outcome_shows_plan_pending_and_writes(): + outcome = AttributionOutcome( + plan=_plan(PlannedAttribution("s1", SHA_A, "unknown")), + applied=attribution_apply.AppliedResult( + applied=[attribution_apply.AppliedWrite("s1", SHA_A, "unknown")], + flipped=[attribution_apply.AppliedWrite("s2", SHA_B, True)], + skipped=[attribution_apply.SkippedWrite("s3", SKIP_CI_UNRESOLVED)], + ), + pending=[PendingFlip("s3", SHA_C)], + ) + text = attribution_apply.render_outcome(outcome) + assert f"ATTRIBUTE s1 -> {SHA_A} (tests_green='unknown')" in text + assert f"PENDING-CI-FLIP s3 @ {SHA_C}" in text + assert f"ATTRIBUTED s1 -> {SHA_A} (tests_green='unknown')" in text + assert f"FLIPPED s2 tests_green=True @ {SHA_B}" in text + assert f"skip s3 ({SKIP_CI_UNRESOLVED})" in text + assert attribution_apply.render_outcome(AttributionOutcome()) == "" + + +# --------------------------------------------------------------------------- # +# consolidate integration — inside the reviewed --apply boundary # +# --------------------------------------------------------------------------- # + + +def _populate(root: Path) -> None: + ctx = root / ".context" + (ctx / "knowledge").mkdir(parents=True) + (ctx / "routes.yaml").write_text(ROUTES_YAML, encoding="utf-8") + (root / ".console").mkdir(parents=True) + (root / ".console" / "task.md").write_text( + "---\ncampaign_id: c-2026-07-01-test\nstatus: active\n" + "started: 2026-07-01\n---\n\n# Task\n\n## Objective\n\nD3 P3.\n", + encoding="utf-8", + ) + + +def _runner(plan: AttributionPlan, ci: object): + def runner(root: Path, apply: bool): + return run_attribution( + root, + apply=apply, + plan_fn=lambda r: plan, + apply_fn=lambda r, p: apply_attribution( + r, p, github_repo="o/r", token="t", resolve_tests_green=_ci(ci) + ), + ) + + return runner + + +def test_consolidate_dry_run_renders_attribution_with_zero_writes(tmp_path): + _populate(tmp_path) + _put(tmp_path, "s1") + _put(tmp_path, "pend", acted_on_commit=SHA_B) + plan = _plan(PlannedAttribution("s1", SHA_A, "unknown")) + before = _snapshot(tmp_path) + cplan = consolidate.plan_consolidation( + tmp_path, apply=False, attribution_runner=_runner(plan, True) + ) + assert _snapshot(tmp_path) == before # byte-identical tree + text = consolidate._render(cplan) + assert f"ATTRIBUTE s1 -> {SHA_A} (tests_green='unknown')" in text + assert f"PENDING-CI-FLIP pend @ {SHA_B}" in text + # The existing consolidation output is intact alongside. + assert "reject" in text + + +def test_consolidate_attribution_failure_never_breaks_dry_run(tmp_path): + _populate(tmp_path) + _put(tmp_path, "s1") + + def boom(root, apply): + raise RuntimeError("attribution exploded") + + cplan = consolidate.plan_consolidation( + tmp_path, apply=False, attribution_runner=boom + ) + assert cplan.attribution is None + assert {a.target for a in cplan.rejections} == {"s1"} + assert "reject s1" in consolidate._render(cplan) + + +def test_consolidate_without_runner_is_unchanged(tmp_path): + _populate(tmp_path) + _put(tmp_path, "s1") + cplan = consolidate.plan_consolidation(tmp_path, apply=False) + assert cplan.attribution is None + + +def test_apply_then_gate_phase_a_true_promotes_same_pass(tmp_path): + # The keystone: a Phase-A (sha, True) write makes the item pass + # gate_promotions inside the SAME plan_consolidation(apply=True) pass. + _populate(tmp_path) + _put(tmp_path, "s1") + plan = _plan(PlannedAttribution("s1", SHA_A, True)) + cplan = consolidate.plan_consolidation( + tmp_path, apply=True, attribution_runner=_runner(plan, "unknown") + ) + assert {a.target for a in cplan.promotions} == {"s1"} + assert (tmp_path / "docs/inject/s1.md").exists() # materialized + item = _get(tmp_path, "s1") + assert item.tier == "warm" + assert item.acted_on_commit == SHA_A + assert item.tests_green is True + + +def test_apply_then_gate_phase_b_flip_promotes_same_pass(tmp_path): + # An item attributed in an EARLIER pass ("unknown") whose CI has since + # resolved green: Phase B flips it and the gate promotes it, same pass. + _populate(tmp_path) + _put(tmp_path, "s1", acted_on_commit=SHA_A) + cplan = consolidate.plan_consolidation( + tmp_path, apply=True, attribution_runner=_runner(_plan(), True) + ) + assert {a.target for a in cplan.promotions} == {"s1"} + assert _get(tmp_path, "s1").tier == "warm" + assert _get(tmp_path, "s1").tests_green is True + + +def test_consolidate_cli_default_runner_dry_run_never_breaks(tmp_path, capsys): + # The CLI wires the default runner; on an empty root (no git, no items) + # the whole attribution path degrades silently and exit stays 0. + assert consolidate.main(["--root", str(tmp_path)]) == 0 + out = capsys.readouterr().out + assert "consolidate: nothing to do" in out