fix(qualification): the waiter parked because the OPERATION could not answer

ROOT CAUSE (run 11081, sha a604dd3): ci_event_wait defaulted to the
master_ci operation, whose authority (deploy_qualify) requires a
ci.yml@refs/heads/master run for the exact SHA. A candidate pushed to a
BRANCH has no such run, so independent_jobs_verdict matched nothing →
pending → combine(success, pending) = pending → the waiter re-armed until
it was killed BY HAND, while the runs/jobs API already showed every job
green. The event transport was never at fault: the event fired, matched
(matched=1) and the durable completion file was written BEFORE the kill.

Repairs, each at the authority that owns the decision:

1. OP-TYPE RESOLUTION (ci_event_wait.op_type_for_sha): when the caller does
   not name the operation, it is resolved from the SHA's OWN runs (master
   run → master_ci; otherwise branch_ci) and the choice is PRINTED, never
   assumed. deploy.yml's explicit master guard is untouched.

2. NO WAIT ON AN UNANSWERABLE QUESTION (op_policy_mismatch): when the op's
   declared ref policy excludes every run the SHA actually has, the waiter
   returns verdict=policy_mismatch with the reason instead of parking
   forever. It is neither success nor pending, so no caller can read it as
   qualification. No runs at all stays honestly pending (CI may not have
   started).

3. REPLAYED PENDING RE-ARMS: a replayed wake whose authority is still
   pending used to TERMINATE the wait as authority_unresolved. A replayed
   event proves an event happened, not that qualification is decidable —
   it now re-arms inside the same logical await.

4. LIFECYCLE RUNNER LEG WIRED (§19/§28): BuildLifecycle accepted a runner
   and never invoked it. qualify_candidate() now calls the canonical waiter
   for the EXACT candidate SHA, records the verdict durably on the issue
   link (a projection, not a second CI truth), and next_action sequences it:
   success → complete_gates, failure → repair_candidate (issue stays open),
   policy_mismatch → re_qualify, pending → wait_for_runner. ci_state()
   projects CI_NOT_STARTED/RUNNING/WAITING_FOR_EVENT/QUALIFIED/FAILED/
   RECONCILING for clients.

5. DEPENDENCY LOCK DETERMINISM (§14-§17): the gate compared committed pins
   against a FRESH resolve against the live index, so an unchanged candidate
   failed because wall-clock time passed (platformdirs 4.12.0 released
   minutes after regeneration). The lock now DECLARES a resolution boundary
   (uv --exclude-newer, recorded in uv's own header command line); the
   checker applies that same boundary and the printed remedy reproduces it,
   so the resolve is a pure function of (pyproject, boundary) and drift
   detection is undiminished (unpinned lines still fail loudly).

Evidence: with --op-type branch_ci the same SHA returns success in 0.4s; the
REAL red candidate 6fe6c90 returns failure/RС=1 (no manufactured red run);
an earlier boundary excludes today's releases (nodeenv 1.10.0,
platformdirs 4.11.14), proving the boundary actually binds.

Tests: tests/test_ci_event_correlation.py (17), test_dep_lock_determinism.py
(10), test_runner_qualification.py (9); affected suites 94 passed.

requirements-ci.txt

@@ -1,5 +1,5 @@
 # This file was autogenerated by uv via the following command:
-#    uv pip compile pyproject.toml -o requirements-ci.txt --extra dev --extra memory --python-version 3.11
+#    uv pip compile pyproject.toml -o requirements-ci.txt --extra dev --extra memory --python-version 3.11 --exclude-newer 2026-09-26T17:29:53Z
 aiohappyeyeballs==2.7.1
     # via aiohttp
 aiohttp==3.14.3

scripts/check_dep_lock.py

@@ -55,7 +55,41 @@ COMPILE_ARGS = [
     "--python-version",
     "3.11",
 ]
-REGEN_HINT = "run: uv pip compile pyproject.toml -o requirements-ci.txt --extra dev --extra memory --python-version 3.11 && commit"
+BASE_REGEN = ("uv pip compile pyproject.toml -o requirements-ci.txt "
+              "--extra dev --extra memory --python-version 3.11")
+
+
+def resolution_boundary(lock_text: str) -> str:
+    """The RESOLUTION BOUNDARY recorded in the lock's own uv header.
+
+    MEASURED DEFECT (2026-09-26): this gate compared committed pins against a
+    FRESH resolve against the live index, so the same unchanged candidate
+    failed qualification purely because wall-clock time passed and an
+    upstream package released (platformdirs 4.12.0 appeared mid-push). The
+    question was never "would the resolver choose these versions now?" — it
+    is "does the committed lock correspond to pyproject under the project's
+    DECLARED resolution policy?".
+
+    Policy: a resolution boundary (RFC3339/date) recorded in the lock header
+    and applied to BOTH the check and the documented regeneration, so the
+    comparison is stable forever and only a DECLARED change (pyproject edit,
+    or an explicit operator boundary advance) can move the pins. The boundary
+    travels in uv's own autogenerated command comment — one artifact, and no
+    way for the remedy to drift from the gate.
+    """
+    import re as _re
+    m = _re.search(r"--exclude-newer[= ](\S+)", lock_text or "")
+    return m.group(1) if m else ""
+
+
+def regen_hint(boundary: str = "") -> str:
+    """The regeneration command that reproduces EXACTLY what the gate
+    expects — same boundary included when the lock declares one."""
+    tail = f" --exclude-newer {boundary}" if boundary else ""
+    return f"run: {BASE_REGEN}{tail} && commit"
+
+
+REGEN_HINT = regen_hint()
 
 
 def uv_binary() -> str:
@@ -87,6 +121,11 @@ def run_uv_compile(out_path: Path, root: Path = ROOT) -> int:
     the compile fails — on success it is noise on CI logs (#150 gate review).
     """
     args = [uv_binary()] + [part.format(out=str(out_path)) for part in COMPILE_ARGS]
+    boundary = resolution_boundary(LOCK_FILE.read_text()) if LOCK_FILE.exists() else ""
+    if boundary:
+        # SAME policy as the gate's expectation: candidates are limited to
+        # those published before the declared boundary.
+        args += ["--exclude-newer", boundary]
     try:
         proc = subprocess.run(args, cwd=str(root), check=False, capture_output=True, text=True)
     except FileNotFoundError:
@@ -128,7 +167,9 @@ def lock_requirements(path: Path) -> dict[str, str]:
                 " wildcards, or bare names) — otherwise the same commit resolves\n"
                 "differently by date, which is exactly what this lock exists to"
                 " prevent.\n"
-                f"Fix the line or regenerate the lock: {REGEN_HINT}",
+                "Fix the line or regenerate the lock: "
+                + regen_hint(resolution_boundary(
+                    LOCK_FILE.read_text() if LOCK_FILE.exists() else "")),
                 file=sys.stderr,
             )
             raise SystemExit(1)
@@ -152,7 +193,10 @@ def compare(committed: Path, regenerated: Path) -> str | None:
             problems.append(f"  {name}: {old[name]} (lock) -> {new[name]} (regenerated)")
     if not problems:
         return None
-    return "requirements-ci.txt is out of sync with pyproject.toml:\n" + "\n".join(problems) + f"\n{REGEN_HINT}"
+    hint = regen_hint(resolution_boundary(
+        committed.read_text() if committed.exists() else ""))
+    return ("requirements-ci.txt is out of sync with pyproject.toml:\n"
+            + "\n".join(problems) + f"\n{hint}")
 
 
 def main() -> int:
@@ -175,7 +219,10 @@ def main() -> int:
             return 1
         if code != 0:
             print(
-                f"FAIL: dependency-lock regeneration failed (uv pip compile exited non-zero). {REGEN_HINT}",
+                "FAIL: dependency-lock regeneration failed (uv pip compile "
+                "exited non-zero). "
+                + regen_hint(resolution_boundary(
+                    LOCK_FILE.read_text() if LOCK_FILE.exists() else "")),
                 file=sys.stderr,
             )
             return 1

scripts/ci_event_wait.py

@@ -270,6 +270,92 @@ def op_ci_alive(op_type: str, sha: str) -> bool:
                for r in runs)
 
 
+def op_type_for_sha(sha: str, explicit: str = "") -> str:
+    """WHICH OPERATION does this SHA's CI belong to?
+
+    MEASURED DEFECT (2026-09-26, run 11081): the waiter defaulted to
+    `master_ci`, whose authority (`deploy_qualify`) requires a
+    `ci.yml@refs/heads/master` run for the exact SHA. A candidate pushed to
+    a BRANCH has no such run, so `independent_jobs_verdict` matched nothing
+    and returned pending — forever. The waiter parked although every job was
+    green; it was killed by hand while the runs/jobs API already showed
+    success. The event path was never the problem: the OPERATION was.
+
+    Resolution is EVIDENCE-BASED, never assumed: if the exact SHA has a
+    ci.yml run whose head_branch is master, it is a master operation →
+    `master_ci`; otherwise it is that ref's own operation → `branch_ci`.
+    An explicit op_type still wins (deploy.yml's master guard relies on it).
+    """
+    if explicit:
+        return explicit
+    try:
+        sys.path.insert(0, str(_SCRIPT_DIR))
+        import deploy_qualify as dq  # noqa: E402
+        import gitea_auth  # noqa: E402
+        server = os.environ.get(
+            "GITEA_SERVER", getattr(gitea_auth, "DEFAULT_SERVER", "")).rstrip("/")
+        repo = os.environ.get("GITEA_REPO",
+                              getattr(gitea_auth, "DEFAULT_REPO", ""))
+        token = gitea_auth.get_token() if gitea_auth.has_token() else ""
+        if not (server and repo and sha and token):
+            return "branch_ci"
+        runs = dq._fetch_runs(server, repo, sha, token) or []
+        for r in runs:
+            if (str(r.get("path", "")).startswith("ci.yml@")
+                    and str(r.get("head_branch", "")) == "master"):
+                return "master_ci"
+        return "branch_ci"
+    except Exception:  # noqa: BLE001 — resolution failure must not invent master scope
+        return "branch_ci"
+
+
+def op_policy_mismatch(op_type: str, sha: str) -> str:
+    """Is the op's run-set filter IMPOSSIBLE for this SHA?
+
+    Returns "" when the op can still resolve (runs exist or CI has not
+    started), or a human reason when the op's declared ref policy excludes
+    every run this SHA actually has — the state that parked the waiter
+    forever. A question whose answer can never change must not be waited on.
+    """
+    spec = op_spec(op_type)
+    branches = spec.get("head_branches")
+    wfp = spec.get("workflow_path")
+    if branches is None and wfp is None:
+        return ""
+    try:
+        sys.path.insert(0, str(_SCRIPT_DIR))
+        import deploy_qualify as dq  # noqa: E402
+        import gitea_auth  # noqa: E402
+        server = os.environ.get(
+            "GITEA_SERVER", getattr(gitea_auth, "DEFAULT_SERVER", "")).rstrip("/")
+        repo = os.environ.get("GITEA_REPO",
+                              getattr(gitea_auth, "DEFAULT_REPO", ""))
+        token = gitea_auth.get_token() if gitea_auth.has_token() else ""
+        if not (server and repo and sha and token):
+            return ""
+        runs = dq._fetch_runs(server, repo, sha, token) or []
+        if not runs:
+            return ""  # CI may not have started; pending is honest
+        in_scope = runs
+        if wfp is not None:
+            in_scope = [r for r in in_scope
+                        if str(r.get("path", "")).startswith(wfp + "@")]
+        if branches is not None:
+            in_scope = [r for r in in_scope
+                        if str(r.get("head_branch", "")) in branches]
+        if in_scope:
+            return ""
+        # runs exist, none can ever satisfy the op's filter
+        seen = sorted({f"{r.get('path')} (head_branch="
+                       f"{r.get('head_branch')})" for r in runs})
+        wanted = (f"workflow {wfp!r} on branch(es) {list(branches)}"
+                  if branches is not None else f"workflow {wfp!r}")
+        return (f"operation {op_type!r} requires {wanted}, but {sha[:12]} "
+                f"has only: {'; '.join(seen)}")
+    except Exception:  # noqa: BLE001 — a detector failure must not fake a mismatch
+        return ""
+
+
 def authority_after_wake(op_type: str = DEFAULT_OP_TYPE, sha: str = "",
                          max_tries: int = 3, gap_secs: int = 4) -> str:
     """(sha is EXPLICIT — ambient env never selects the target.)"""
@@ -374,26 +460,34 @@ def run(*, sha: str, deadline_secs: int,
         verdict = normalize_verdict(authority_after_wake(op_type, sha))
         if verdict == "pending" and not op_ci_alive(op_type, sha):
             verdict = normalize_verdict(op_authority_verdict(op_type, sha))
-        mode = ("event_replayed" if verdict != "pending"
-                else "authority_unresolved")
-        verdict_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
-        out = {
-            "sha": sha,
-            "op_type": op_type,
-            "authority": spec["authority"],
-            "registered_at": registered_at,
-            "event_received_at": event_at,
-            "controller_resumed_at": resumed_at,
-            "verdict_at": verdict_at,
-            "mode": mode,
-            "rearm_reason": "completion event replayed from durable log "
-                            "(arrived before registration)",
-            "verdict": verdict,
-            "wall_secs": round(time.time() - t0, 3),
-        }
-        print(json.dumps(out))
-        return 0 if verdict == "success" else 1
-    rec = ew.register_process_wait(
+        if verdict != "pending":
+            mode = "event_replayed"
+            verdict_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+            out = {
+                "sha": sha,
+                "op_type": op_type,
+                "authority": spec["authority"],
+                "registered_at": registered_at,
+                "event_received_at": event_at,
+                "controller_resumed_at": resumed_at,
+                "verdict_at": verdict_at,
+                "mode": mode,
+                "rearm_reason": "completion event replayed from durable log "
+                                "(arrived before registration)",
+                "verdict": verdict,
+                "wall_secs": round(time.time() - t0, 3),
+            }
+            print(json.dumps(out))
+            return 0 if verdict == "success" else 1
+        # MEASURED DEFECT (2026-09-26): a REPLAYED event whose authority is
+        # still pending used to TERMINATE the wait here
+        # (mode="authority_unresolved"). A replayed wake is evidence that an
+        # event happened, NOT that qualification is decidable — pending must
+        # re-arm inside the same logical await, exactly like the live path.
+        replayed_rec = rec
+    else:
+        replayed_rec = None
+    rec = replayed_rec or ew.register_process_wait(
         op, project="rune",
         objective=f"event-driven {op_type} qualification for {sha}",
         watched_for="Gitea CI completion event (wake; verdict re-read "
@@ -465,6 +559,12 @@ def run(*, sha: str, deadline_secs: int,
                 if verdict != "pending":
                     mode = "event"
                     break
+                mismatch = op_policy_mismatch(op_type, sha)
+                if mismatch:
+                    verdict = "policy_mismatch"
+                    mode = "policy_mismatch"
+                    rearm_reason = mismatch
+                    break
                 rearm_count += 1
                 rearm_reason = ("completion file seen; authority "
                                 "pending — re-subscribing")
@@ -477,6 +577,12 @@ def run(*, sha: str, deadline_secs: int,
             if verdict != "pending":
                 mode = "watchdog_authority_recovered"
                 break
+            mismatch = op_policy_mismatch(op_type, sha)
+            if mismatch:
+                verdict = "policy_mismatch"
+                mode = "policy_mismatch"
+                rearm_reason = mismatch
+                break
             rearm_count += 1
             rearm_reason = ("watchdog expiry; op pending (not started "
                             "or lagging) — re-subscribing inotify")
@@ -547,7 +653,7 @@ def main() -> int:
     sha = os.environ.get("DEPLOY_SHA", "")
     args = sys.argv[1:]
     deadline = 900
-    op_type = DEFAULT_OP_TYPE
+    op_type = ""  # resolved from evidence below when the caller omits it
     watchdog = 300
     task_id = ""
     i = 0
@@ -575,6 +681,21 @@ def main() -> int:
     if not sha:
         print(json.dumps({"error": "no sha (DEPLOY_SHA or --sha required)"}))
         return 1
+    # OP-TYPE RESOLUTION (measured defect): deploy.yml passes --op-type
+    # implicitly via the master_ci default, which is correct FOR ITS OWN
+    # master-guarded job. A branch candidate under that default can never
+    # resolve (its authority requires a master run) and parked the waiter
+    # forever. When the caller does not name the operation, resolve it from
+    # the SHA's actual runs and SAY SO — never assume master scope silently.
+    resolved_from = "explicit"
+    if not op_type:
+        op_type = op_type_for_sha(sha, explicit="")
+        resolved_from = "evidence"
+    if resolved_from == "evidence":
+        print(json.dumps({"op_type_resolution": {
+            "sha": sha, "op_type": op_type, "source": "gitea runs",
+            "note": "no --op-type given; resolved from the SHA's own runs"}}),
+            file=sys.stderr)
     # A-PATCH fix: run()'s authority subprocess (deploy_qualify.py) reads
     # DEPLOY_SHA from the environment — a CLI `--sha` consumer previously
     # got a silent "pending" verdict forever because the var was never

src/rune/lifecycle.py

@@ -159,8 +159,10 @@ class BuildLifecycle:
                     .get("verification_obligations", []),
                 }
                 # candidate + runner truth
-                state["candidate_sha"] = ext.get("candidate_sha", "")
-                state["runner"] = ext.get("runner_result") or None
+                state["candidate_sha"] = (
+                    (link or {}).get("candidate_sha", "")
+                    or ext.get("candidate_sha", ""))
+                state["runner"] = self.ci_state(task_id)
             state["contract_id"] = contract_id
         # the ONE question
         try:
@@ -205,11 +207,159 @@ class BuildLifecycle:
         if contract_id:
             act = self._contract_next_action(contract_id)
             if act is not None:
+                # CANDIDATE → RUNNER (§19): once local work produced an exact
+                # candidate SHA, the runner leg IS the next action — and a
+                # recorded verdict sequences past it without anyone noticing
+                # CI finished.
+                ci = self.ci_state(task_id)
+                if act.action == "candidate" and ci["sha"]:
+                    if ci["verdict"] == "success":
+                        return LifecycleAction(
+                            "complete_gates",
+                            reason="candidate qualified by the runner "
+                                   f"({ci['op_type'] or 'op'} via "
+                                   f"{ci['mode'] or 'authority'})",
+                            detail={"ci": ci})
+                    if ci["verdict"] == "failure":
+                        return LifecycleAction(
+                            "repair_candidate",
+                            reason="runner qualification FAILED for the exact "
+                                   "candidate; the issue stays open",
+                            detail={"ci": ci})
+                    if ci["verdict"] == "policy_mismatch":
+                        return LifecycleAction(
+                            "re_qualify",
+                            reason="runner operation does not match the "
+                                   "candidate's ref — re-qualify with the "
+                                   "resolved operation",
+                            detail={"ci": ci})
+                    return LifecycleAction(
+                        "qualify_candidate",
+                        reason="candidate committed; the durable runner wait "
+                               "decides qualification",
+                        detail={"ci": ci, "sha": ci["sha"]})
                 return act
 
         return LifecycleAction("execute", reason="no contract governs this task; "
                                                  "operator model path executes")
 
+    # -- candidate qualification (§19/§28: the runner leg) ----------------------
+
+    def qualify_candidate(self, task_id: str, *, sha: str = "",
+                          op_type: str = "", deadline_secs: int = 900,
+                          watchdog_secs: int = 300) -> dict:
+        """Qualify the EXACT candidate SHA through the canonical waiter.
+
+        ONE authority: this calls scripts/ci_event_wait.py, which re-reads the
+        canonical qualification verdict after the wake — the event is
+        transport, qualification is truth. The operation type is resolved from
+        the SHA's own runs when the caller does not name it, because a branch
+        candidate under master scope can never resolve (measured: run 11081
+        parked the waiter while every job was green).
+
+        The verdict is recorded durably on the issue link (a PROJECTION of the
+        runner's answer, not a second CI truth) so a fresh process resumes.
+        """
+        link = (self._issues.get(task_id) if self._issues else None) or {}
+        sha = sha or link.get("candidate_sha", "")
+        if not sha:
+            return {"verdict": "refused",
+                    "reason": "no candidate SHA — nothing to qualify",
+                    "next_action": "commit_candidate"}
+        runner = self._runner
+        if runner is not None:
+            # injected backend (tests, alternate transports): same contract
+            out = runner(task_id=task_id, sha=sha, op_type=op_type,
+                         deadline_secs=deadline_secs,
+                         watchdog_secs=watchdog_secs) or {}
+        else:
+            out = self._run_canonical_waiter(
+                sha=sha, op_type=op_type, deadline_secs=deadline_secs,
+                watchdog_secs=watchdog_secs, task_id=task_id)
+        verdict = str(out.get("verdict", "pending"))
+        if self._issues is not None:
+            self._issues.record_qualification(
+                task_id, verdict=verdict, sha=sha,
+                op_type=out.get("op_type", op_type),
+                mode=out.get("mode", ""),
+                reason=out.get("reason", "") or out.get("rearm_reason", ""))
+        return {
+            "verdict": verdict,
+            "sha": sha,
+            "op_type": out.get("op_type", op_type),
+            "mode": out.get("mode", ""),
+            "authority": out.get("authority", "ci_runs"),
+            "next_action": self._after_qualification(verdict),
+        }
+
+    def _run_canonical_waiter(self, *, sha: str, op_type: str,
+                              deadline_secs: int, watchdog_secs: int,
+                              task_id: str = "") -> dict:
+        """Invoke the canonical waiter as a subprocess — the production
+        transport. Its stdout is one JSON line (the §16D envelope)."""
+        import subprocess as _sp
+        script = (Path(__file__).resolve().parent.parent.parent
+                  / "scripts" / "ci_event_wait.py")
+        if not script.is_file():
+            return {"verdict": "refused",
+                    "reason": f"canonical waiter not found at {script}"}
+        cmd = [sys.executable, str(script), "--sha", sha,
+               "--deadline", str(deadline_secs),
+               "--watchdog", str(watchdog_secs)]
+        if op_type:
+            cmd += ["--op-type", op_type]
+        if task_id:
+            cmd += ["--task-id", task_id]
+        try:
+            proc = _sp.run(cmd, capture_output=True, text=True,
+                           timeout=deadline_secs + 120)
+        except (_sp.TimeoutExpired, OSError) as exc:
+            return {"verdict": "pending",
+                    "reason": f"waiter transport unavailable: {exc}"}
+        for line in reversed((proc.stdout or "").splitlines()):
+            line = line.strip()
+            if line.startswith("{") and '"verdict"' in line:
+                return json.loads(line)
+        return {"verdict": "pending",
+                "reason": f"waiter produced no verdict envelope "
+                          f"(rc={proc.returncode})"}
+
+    @staticmethod
+    def _after_qualification(verdict: str) -> str:
+        """The next action the lifecycle sequences from a runner verdict."""
+        if verdict == "success":
+            return "complete_gates"
+        if verdict == "failure":
+            return "repair_candidate"
+        if verdict == "policy_mismatch":
+            return "re_qualify"      # wrong operation, correctable
+        return "wait_for_runner"     # pending: keep the durable wait
+
+    def ci_state(self, task_id: str) -> dict:
+        """CI state for CLIENTS — a projection of the recorded runner verdict
+        plus the candidate reference. Never an independent status authority:
+        a recorded success is reported as QUALIFIED even if some transport is
+        still reconciling."""
+        link = (self._issues.get(task_id) if self._issues else None) or {}
+        sha = link.get("candidate_sha", "")
+        res = link.get("runner_result") or {}
+        verdict = res.get("verdict", "")
+        if verdict == "success":
+            state = "CI_QUALIFIED"
+        elif verdict == "failure":
+            state = "CI_FAILED"
+        elif verdict == "policy_mismatch":
+            state = "CI_RECONCILING"
+        elif verdict == "pending":
+            state = "CI_WAITING_FOR_EVENT"
+        elif sha:
+            state = "CI_RUNNING"
+        else:
+            state = "CI_NOT_STARTED"
+        return {"state": state, "sha": sha, "verdict": verdict,
+                "mode": res.get("mode", ""), "op_type": res.get("op_type", ""),
+                "at": res.get("at", ""), "reason": res.get("reason", "")}
+
     def _contract_for_task(self, task_id: str) -> str:
         """Resolve the contract reference for a task from durable links —
         the IssueLink holds contract_id; no second mapping exists."""
@@ -470,6 +620,26 @@ class IssueLink:
         return {"updated": True, "issue_number": link["issue_number"],
                 "candidate_sha": sha}
 
+    def record_qualification(self, task_id: str, *, verdict: str,
+                             sha: str, op_type: str = "", mode: str = "",
+                             reason: str = "") -> dict:
+        """Record the canonical runner verdict FOR THIS CANDIDATE.
+
+        A PROJECTION of the runner authority's answer (never a second CI
+        truth): the lifecycle reads it back to sequence the next action, and
+        clients project it. Idempotent per (sha, verdict)."""
+        doc = self._load()
+        link = doc["links"].get(task_id)
+        if not link:
+            return {"recorded": False, "reason": "no issue link for task"}
+        link["runner_result"] = {
+            "verdict": verdict, "sha": sha, "op_type": op_type,
+            "mode": mode, "reason": reason[:400] if reason else "",
+            "at": _now(),
+        }
+        self._save(doc)
+        return {"recorded": True, "verdict": verdict, "sha": sha}
+
     def get(self, task_id: str) -> dict | None:
         return self._load()["links"].get(task_id)
 

tests/test_ci_event_correlation.py

@@ -0,0 +1,225 @@
+"""§21 event-correlation attack tests + §6 lost/early/late-event robustness.
+
+MEASURED DEFECT BEING PINNED (run 11081): the waiter defaulted to the
+`master_ci` operation, whose authority requires a
+`ci.yml@refs/heads/master` run for the exact SHA. A candidate pushed to a
+BRANCH has no such run, so the jobs leg matched nothing → pending forever →
+the waiter parked although every job was green. The event transport was
+never at fault; the OPERATION was. These tests hold that shut.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+sys.path.insert(0, str(ROOT / "src"))
+
+import ci_event_wait as cew  # noqa: E402
+import deploy_qualify as dq  # noqa: E402
+import completion_file as cf  # noqa: E402
+
+SHA = "a604dd379e7d0fcd7d64c6be45f5dfcdec750e70"
+OTHER_SHA = "6fe6c90c41671f367d3e132eb3e23dfb214f76bb"
+
+
+def _run(path: str, sha: str, branch: str, status="completed", conclusion="success", rid=1):
+    return {"id": rid, "path": path, "head_sha": sha, "head_branch": branch, "status": status, "conclusion": conclusion}
+
+
+def _job(conclusion="success", name="unit-tests"):
+    return {"name": name, "status": "completed", "conclusion": conclusion}
+
+
+@pytest.fixture()
+def fake_runs(monkeypatch):
+    """Inject the run set; never touch a live Gitea."""
+    state = {"runs": [], "jobs": {}}
+
+    def _fetch_runs(server, repo, sha, token):
+        return [r for r in state["runs"] if r.get("head_sha") == sha]
+
+    def _jobs_fetch(rid):  # 1-arg callable, as independent_jobs_verdict expects
+        return state["jobs"].get(rid, [_job()])
+
+    monkeypatch.setattr(dq, "_fetch_runs", _fetch_runs)
+    monkeypatch.setenv("GITEA_SERVER", "http://fake")
+    monkeypatch.setenv("GITEA_REPO", "conductor/rune")
+    monkeypatch.setattr("gitea_auth.has_token", lambda: True)
+    monkeypatch.setattr("gitea_auth.get_token", lambda: "t")
+    state["jobs_fetch"] = _jobs_fetch
+    return state
+
+
+# ── the defect: op-type resolution must follow the SHA's OWN runs ──────────
+
+
+def test_op_type_resolves_branch_candidate_to_branch_ci(fake_runs):
+    fake_runs["runs"] = [_run("ci.yml@refs/heads/m10-derive-refresh", SHA, "m10-derive-refresh")]
+    assert cew.op_type_for_sha(SHA) == "branch_ci"
+
+
+def test_op_type_resolves_master_run_to_master_ci(fake_runs):
+    fake_runs["runs"] = [_run("ci.yml@refs/heads/master", SHA, "master")]
+    assert cew.op_type_for_sha(SHA) == "master_ci"
+
+
+def test_op_type_explicit_wins_deploy_semantics_preserved(fake_runs):
+    """deploy.yml's master guard passes master_ci explicitly — unchanged."""
+    fake_runs["runs"] = [_run("ci.yml@refs/heads/m10-derive-refresh", SHA, "m10-derive-refresh")]
+    assert cew.op_type_for_sha(SHA, explicit="master_ci") == "master_ci"
+
+
+def test_op_type_defaults_to_candidate_scope_when_unresolvable(fake_runs):
+    """No runs yet / no Gitea: the DEFAULT must be candidate scope, never
+    master — a wrong master default is what parked the waiter forever."""
+    fake_runs["runs"] = []
+    assert cew.op_type_for_sha(SHA) == "branch_ci"
+
+
+# ── the unanswerable question must be REPORTED, not waited on forever ──────
+
+
+def test_policy_mismatch_detected_for_master_op_on_branch_sha(fake_runs):
+    fake_runs["runs"] = [_run("ci.yml@refs/heads/m10-derive-refresh", SHA, "m10-derive-refresh")]
+    reason = cew.op_policy_mismatch("master_ci", SHA)
+    assert "master_ci" in reason and "m10-derive-refresh" not in reason.split("has only")[0]
+    assert "ci.yml@refs/heads/m10-derive-refresh" in reason
+
+
+def test_no_mismatch_when_a_run_can_still_satisfy_the_op(fake_runs):
+    fake_runs["runs"] = [_run("ci.yml@refs/heads/master", SHA, "master")]
+    assert cew.op_policy_mismatch("master_ci", SHA) == ""
+
+
+def test_no_mismatch_when_ci_has_not_started(fake_runs):
+    """No runs at all is genuinely pending — CI may not have started.
+    Reporting a mismatch there would abort a legitimate wait."""
+    fake_runs["runs"] = []
+    assert cew.op_policy_mismatch("master_ci", SHA) == ""
+
+
+def test_policy_mismatch_is_not_a_success_and_not_pending(fake_runs):
+    """The verdict vocabulary must keep the three apart: a mismatch is a
+    refuse-with-reason, so no caller can read it as qualification."""
+    assert cew.normalize_verdict("policy_mismatch") == "policy_mismatch"
+    assert cew.normalize_verdict("pending (probe failed)") == "pending"
+    assert cew.normalize_verdict("policy_mismatch") != "success"
+
+
+# ── exact-SHA fail-closed cases (§11) ──────────────────────────────────────
+
+
+def test_jobs_verdict_requires_every_job_success(fake_runs):
+    """The deploy leg's real input: a master run whose job failed."""
+    fake_runs["runs"] = [_run("ci.yml@refs/heads/master", SHA, "master", rid=7)]
+    fake_runs["jobs"] = {7: [_job(), _job(conclusion="failure", name="lint")]}
+    assert dq.independent_jobs_verdict(fake_runs["runs"], fake_runs["jobs_fetch"], sha=SHA) == "failure"
+
+
+def test_deploy_jobs_leg_refuses_a_branch_sha(fake_runs):
+    """THE DEFECT'S OTHER HALF, PINNED AS CORRECT BEHAVIOUR.
+
+    deploy_qualify requires a ci.yml@refs/heads/master run for the exact
+    SHA — that is release safety (a branch must never be promoted), so
+    pending for a branch SHA is RIGHT here. The consequence is the one that
+    parked the waiter: this authority can NEVER resolve a branch candidate,
+    so the waiter must select `branch_ci` for it (op_type_for_sha) instead
+    of defaulting every SHA into master scope."""
+    fake_runs["runs"] = [_run("ci.yml@refs/heads/m10-derive-refresh", SHA, "m10-derive-refresh", rid=7)]
+    fake_runs["jobs"] = {7: [_job()]}
+    assert dq.independent_jobs_verdict(fake_runs["runs"], fake_runs["jobs_fetch"], sha=SHA) == "pending"
+    # and the waiter resolves the operation that CAN answer for this SHA
+    assert cew.op_type_for_sha(SHA) == "branch_ci"
+
+
+def test_jobs_verdict_pending_for_other_sha_run(fake_runs):
+    """A green run for a DIFFERENT SHA must never qualify this SHA."""
+    fake_runs["runs"] = [_run("ci.yml@refs/heads/master", OTHER_SHA, "master")]
+    assert dq.independent_jobs_verdict([r for r in fake_runs["runs"]], fake_runs["jobs_fetch"], sha=SHA) == "pending"
+
+
+def test_jobs_verdict_pending_when_run_incomplete(fake_runs):
+    fake_runs["runs"] = [_run("ci.yml@refs/heads/master", SHA, "master", status="in_progress", conclusion="")]
+    assert dq.independent_jobs_verdict(fake_runs["runs"], fake_runs["jobs_fetch"], sha=SHA) == "pending"
+
+
+def test_jobs_verdict_ignores_tag_runs_and_sibling_workflows(fake_runs):
+    fake_runs["runs"] = [
+        _run("ci.yml@refs/tags/v9.9.9", SHA, "v9.9.9", conclusion="failure", rid=1),
+        _run("deploy.yml@refs/heads/master", SHA, "master", conclusion="skipped", rid=2),
+        _run("ci.yml@refs/heads/master", SHA, "master", conclusion="success", rid=3),
+    ]
+    fake_runs["jobs"] = {1: [_job(conclusion="failure")], 2: [], 3: [_job()]}
+    assert dq.independent_jobs_verdict(fake_runs["runs"], fake_runs["jobs_fetch"], sha=SHA) == "success"
+
+
+def test_retried_run_for_same_sha_uses_latest_completed(fake_runs):
+    """§6.H: a retried run for the same SHA — the LATEST completed attempt
+    decides (existing policy), and an older failing attempt cannot veto it."""
+    fake_runs["runs"] = [
+        {
+            "id": 1,
+            "path": "ci.yml@refs/heads/master",
+            "head_sha": SHA,
+            "head_branch": "master",
+            "status": "completed",
+            "conclusion": "failure",
+            "created_at": "2026-09-26T10:00:00Z",
+        },
+        {
+            "id": 2,
+            "path": "ci.yml@refs/heads/master",
+            "head_sha": SHA,
+            "head_branch": "master",
+            "status": "completed",
+            "conclusion": "success",
+            "created_at": "2026-09-26T11:00:00Z",
+        },
+    ]
+    fake_runs["jobs"] = {1: [_job(conclusion="failure")], 2: [_job()]}
+    assert dq.independent_jobs_verdict(fake_runs["runs"], fake_runs["jobs_fetch"], sha=SHA) == "success"
+
+
+# ── no cross-wake: correlation identity is the exact SHA ───────────────────
+
+
+def test_completion_file_is_keyed_by_exact_operation(tmp_path, monkeypatch):
+    """§6.F/G: a stale event (old SHA) or another task's event cannot wake
+    this candidate — the durable filename IS the correlation identity."""
+    monkeypatch.setenv("WR_ROOT", str(tmp_path))
+    (tmp_path / "resumes").mkdir(parents=True, exist_ok=True)
+    a = cf.completion_path(f"ci-{SHA}")
+    b = cf.completion_path(f"ci-{OTHER_SHA}")
+    assert a != b
+    assert a.name == f"ci-{SHA}.done"
+    # a file for the other SHA is not this operation's wake
+    b.write_text(json.dumps({"operation_id": f"ci-{OTHER_SHA}"}))
+    assert not a.exists()
+
+
+def test_short_sha_cannot_alias_a_full_sha_op(tmp_path, monkeypatch):
+    """A 12-char prefix must never be treated as the 40-char operation."""
+    monkeypatch.setenv("WR_ROOT", str(tmp_path))
+    assert cf.completion_path(f"ci-{SHA[:12]}") != cf.completion_path(f"ci-{SHA}")
+
+
+# ── duplicate / replay idempotence ─────────────────────────────────────────
+
+
+def test_replayed_event_is_idempotent_through_the_durable_log(tmp_path, monkeypatch):
+    """§6.E: the durable event log is the replay source; the same event read
+    twice yields the same match (no double-effect state)."""
+    monkeypatch.setenv("WR_ROOT", str(tmp_path))
+    (tmp_path / "resumes").mkdir(parents=True, exist_ok=True)
+    op = f"ci-{SHA}"
+    cf.write_completion(op, {"operation_id": op, "sha": SHA, "exit_code": 0})
+    first = cf.wait_for_completion(op, timeout_secs=1)
+    second = cf.wait_for_completion(op, timeout_secs=1)
+    assert first is not None and second is not None
+    assert first.get("operation_id") == second.get("operation_id") == op

tests/test_dep_lock_determinism.py

@@ -0,0 +1,157 @@
+"""§22 dependency-lock attack tests — determinism vs the wall clock.
+
+MEASURED DEFECT (2026-09-26): the gate compared committed pins against a
+FRESH resolve against the live index, so the same unchanged candidate failed
+purely because time passed and an upstream package released (platformdirs
+4.12.0 appeared mid-push, minutes after the lock was regenerated). The gate
+must answer "does the lock correspond to pyproject under the DECLARED
+resolution policy?", not "would the resolver choose these versions now?".
+
+Policy under test: a resolution boundary recorded in the lock's own uv
+header, applied identically by the check and the documented remedy, so the
+resolve is a pure function of (pyproject, boundary).
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+
+import check_dep_lock as cdl  # noqa: E402
+
+LOCK = """# This file was autogenerated by uv via the following command:
+#    uv pip compile pyproject.toml -o requirements-ci.txt --extra dev --extra memory --python-version 3.11 --exclude-newer 2026-09-26T17:29:53Z
+aiohappyeyeballs==2.7.1
+    # via aiohttp
+diff-cover==10.6.0
+    # via rune (pyproject.toml)
+platformdirs==4.12.0
+"""
+
+
+def _write(tmp: Path, name: str, body: str) -> Path:
+    p = tmp / name
+    p.write_text(body, encoding="utf-8")
+    return p
+
+
+# ── boundary parse + command convergence ──────────────────────────────────
+
+
+def test_boundary_is_parsed_from_the_lock_header():
+    assert cdl.resolution_boundary(LOCK) == "2026-09-26T17:29:53Z"
+    assert cdl.resolution_boundary("no header here\nfoo==1.0\n") == ""
+
+
+def test_remedy_reproduces_exactly_what_the_gate_expects():
+    """§17: no command in an error message may instruct the operator to
+    generate a different result from what the gate expects. The remedy
+    carries the SAME boundary the checker applies."""
+    boundary = cdl.resolution_boundary(LOCK)
+    hint = cdl.regen_hint(boundary)
+    assert f"--exclude-newer {boundary}" in hint
+    # and with no declared boundary the remedy stays flag-free (legacy locks)
+    assert "--exclude-newer" not in cdl.regen_hint("")
+
+
+def test_check_applies_the_locks_own_boundary(monkeypatch, tmp_path):
+    """The checker's regeneration argv must include the boundary — server-side
+    proof that the comparison is time-stable, no network needed."""
+    lock = _write(tmp_path, "requirements-ci.txt", LOCK)
+    monkeypatch.setattr(cdl, "LOCK_FILE", lock)
+    captured = {}
+
+    def _fake_run(args, cwd=None, **kw):
+        captured["args"] = args
+
+        class _P:
+            returncode = 0
+            stdout = ""
+            stderr = ""
+
+        return _P()
+
+    monkeypatch.setattr(cdl.subprocess, "run", _fake_run)
+    monkeypatch.setattr(cdl, "uv_binary", lambda: "uv")
+    cdl.run_uv_compile(tmp_path / "regen.txt")
+    assert "--exclude-newer" in captured["args"]
+    assert "2026-09-26T17:29:53Z" in captured["args"]
+
+
+def test_check_omits_the_flag_for_a_legacy_lock(monkeypatch, tmp_path):
+    """Backward compatibility: a lock with no declared boundary keeps the
+    old fresh-resolve comparison (strict drift detection unchanged)."""
+    lock = _write(tmp_path, "requirements-ci.txt", "foo==1.0\n")
+    monkeypatch.setattr(cdl, "LOCK_FILE", lock)
+    captured = {}
+
+    class _Proc:
+        returncode = 0
+        stdout = ""
+        stderr = ""
+
+    def _fake_run(args, cwd=None, **kw):
+        captured["args"] = args
+        return _Proc()
+
+    monkeypatch.setattr(cdl.subprocess, "run", _fake_run)
+    monkeypatch.setattr(cdl, "uv_binary", lambda: "uv")
+    cdl.run_uv_compile(tmp_path / "regen.txt")
+    assert "--exclude-newer" not in captured["args"]
+
+
+# ── drift detection is preserved (not weakened) ───────────────────────────
+
+
+def test_unchanged_pins_pass(tmp_path):
+    a = _write(tmp_path, "a.txt", LOCK)
+    b = _write(tmp_path, "b.txt", LOCK)
+    assert cdl.compare(a, b) is None
+
+
+def test_declaration_change_makes_a_stale_lock_fail(tmp_path):
+    a = _write(tmp_path, "a.txt", "foo==1.0\nbar==2.0\n")
+    b = _write(tmp_path, "b.txt", "foo==1.0\nbar==2.5\n")
+    drift = cdl.compare(a, b)
+    assert drift and "bar: 2.0 (lock) -> 2.5 (regenerated)" in drift
+
+
+def test_boundary_advance_without_regeneration_fails(tmp_path):
+    """§22: advancing the resolution boundary WITHOUT regenerating leaves the
+    lock stale — reported, never silently accepted. The report carries the
+    boundary-aware remedy."""
+    a = _write(tmp_path, "old.txt", LOCK)
+    advanced = LOCK.replace("2026-09-26T17:29:53Z", "2026-10-01T00:00:00Z")
+    b = _write(tmp_path, "new.txt", advanced.replace("platformdirs==4.12.0", "platformdirs==4.13.0"))
+    drift = cdl.compare(a, b)
+    assert drift and "platformdirs" in drift
+    assert "--exclude-newer 2026-09-26T17:29:53Z" in drift
+
+
+def test_boundary_advance_with_regenerated_lock_passes(tmp_path):
+    a = _write(tmp_path, "old.txt", LOCK)
+    b = _write(tmp_path, "new.txt", LOCK.replace("2026-09-26T17:29:53Z", "2026-10-01T00:00:00Z"))
+    assert cdl.compare(a, b) is None  # comment/boundary changes are not pins
+
+
+def test_unpinned_requirement_is_still_refused(tmp_path):
+    """The strictness the gate exists for is untouched: a floor instead of an
+    exact pin must fail loudly."""
+    bad = _write(tmp_path, "bad.txt", "# header\nfoo>=1.0\n")
+    with pytest.raises(SystemExit):
+        cdl.lock_requirements(bad)
+
+
+def test_committed_lock_declares_a_boundary():
+    """The repository's own lock carries the policy (this is the state that
+    makes the candidate reproducible)."""
+    text = (ROOT / "requirements-ci.txt").read_text()
+    assert cdl.resolution_boundary(text), (
+        "requirements-ci.txt must declare a resolution boundary "
+        "(--exclude-newer) or the gate is wall-clock dependent again"
+    )

tests/test_runner_qualification.py

@@ -0,0 +1,162 @@
+"""§19/§28 runner → lifecycle qualification wiring.
+
+The runner leg was MISSING from the lifecycle: `BuildLifecycle` accepted an
+injected runner and never invoked it, so a pushed candidate had no step that
+qualifies it. These tests hold the sequence shut: candidate → qualify
+(recorded durably) → next action, including a FRESH PROCESS that reads only
+durable state.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+from rune.construction_interface import ConstructionInterface  # noqa: E402
+from rune.lifecycle import BuildLifecycle, IssueLink  # noqa: E402
+from rune.orchestration.atomic import AtomicStore  # noqa: E402
+from rune.orchestration.conversation_state import TaskStateRepository  # noqa: E402
+from rune.orchestration.event_store import DurableEventStore  # noqa: E402
+from rune.scaffold import Scaffold  # noqa: E402
+
+SHA = "a604dd379e7d0fcd7d64c6be45f5dfcdec750e70"
+CID = "ctx-run"
+
+
+def _env(tmp_path, runner=None):
+    store = tmp_path / "scaffold"
+    store.mkdir()
+    repo = TaskStateRepository(DurableEventStore(str(tmp_path / "e.jsonl")), AtomicStore(str(tmp_path / "hot")))
+    rec = repo.create("qualify the candidate", "c1")
+    link = IssueLink(state_dir=tmp_path / "issues")
+    iface = ConstructionInterface(builder="t", project_store=str(store), project_id="issue161-harness")
+    iface.scaffold = Scaffold(str(store), "issue161-harness")
+    doc = iface.scaffold.load()
+    iface.scaffold.new_contract(
+        doc, contract_id=CID, need="n", origin="user", scope=["src/rune"], requirement="r", resolution="BUILD"
+    )
+    iface.scaffold.save(doc)
+    lc = BuildLifecycle(task_state=repo, construction_store=str(store), issue_link=link, runner=runner)
+    return {
+        "task": rec.task_id,
+        "store": store,
+        "link": link,
+        "lc": lc,
+        "repo": repo,
+        "iface": iface,
+        "issues": tmp_path / "issues",
+    }
+
+
+def test_qualify_candidate_records_and_sequences_success(tmp_path):
+    calls = {}
+
+    def runner(*, task_id, sha, op_type, deadline_secs, watchdog_secs):
+        calls.update({"task_id": task_id, "sha": sha, "op_type": op_type})
+        return {"verdict": "success", "op_type": op_type or "branch_ci", "mode": "event", "authority": "ci_runs"}
+
+    env = _env(tmp_path, runner=runner)
+    env["link"].attach_explicit(env["task"], issue_number=9, contract_id=CID)
+    env["link"].update_candidate(env["task"], sha=SHA)
+    out = env["lc"].qualify_candidate(env["task"])
+    assert out["verdict"] == "success"
+    assert out["next_action"] == "complete_gates"
+    assert calls["sha"] == SHA  # the EXACT candidate, taken from the link
+    # recorded durably on the link (a projection, not a second CI truth)
+    link = env["link"].get(env["task"])
+    assert link["runner_result"]["verdict"] == "success"
+    assert link["runner_result"]["sha"] == SHA
+
+
+def test_qualify_candidate_refuses_without_a_candidate_sha(tmp_path):
+    env = _env(tmp_path, runner=lambda **kw: {"verdict": "success"})
+    env["link"].attach_explicit(env["task"], issue_number=9, contract_id=CID)
+    out = env["lc"].qualify_candidate(env["task"])
+    assert out["verdict"] == "refused"
+    assert out["next_action"] == "commit_candidate"
+
+
+@pytest.mark.parametrize(
+    "verdict,expected",
+    [
+        ("success", "complete_gates"),
+        ("failure", "repair_candidate"),
+        ("policy_mismatch", "re_qualify"),
+        ("pending", "wait_for_runner"),
+    ],
+)
+def test_verdict_sequences_the_next_action(tmp_path, verdict, expected):
+    env = _env(tmp_path, runner=lambda **kw: {"verdict": verdict})
+    env["link"].attach_explicit(env["task"], issue_number=9, contract_id=CID)
+    env["link"].update_candidate(env["task"], sha=SHA)
+    assert env["lc"].qualify_candidate(env["task"])["next_action"] == expected
+
+
+def test_ci_state_vocabulary_is_a_projection(tmp_path):
+    env = _env(tmp_path)
+    env["link"].attach_explicit(env["task"], issue_number=9, contract_id=CID)
+    assert env["lc"].ci_state(env["task"])["state"] == "CI_NOT_STARTED"
+    env["link"].update_candidate(env["task"], sha=SHA)
+    assert env["lc"].ci_state(env["task"])["state"] == "CI_RUNNING"
+    for verdict, state in (
+        ("success", "CI_QUALIFIED"),
+        ("failure", "CI_FAILED"),
+        ("policy_mismatch", "CI_RECONCILING"),
+        ("pending", "CI_WAITING_FOR_EVENT"),
+    ):
+        env["link"].record_qualification(env["task"], verdict=verdict, sha=SHA)
+        assert env["lc"].ci_state(env["task"])["state"] == state
+
+
+def test_next_action_sequenced_by_a_recorded_verdict(tmp_path):
+    """The §19 advance: a contract at the candidate stage plus a RECORDED
+    success sequences completion gates without anyone announcing that CI
+    finished."""
+    env = _env(tmp_path)
+    env["link"].attach_explicit(env["task"], issue_number=9, contract_id=CID)
+    env["link"].update_candidate(env["task"], sha=SHA)
+    # locally complete construction so the contract action is `candidate`
+    doc = env["iface"].scaffold.load()
+    for ph in ("approved", "building"):
+        env["iface"].scaffold.transition_contract(doc, CID, ph, evidence_ref="t")
+    env["iface"].scaffold.save(doc)
+    act = env["lc"].next_action(env["task"])
+    assert act.action == "qualify_candidate", act  # no verdict recorded yet
+    env["link"].record_qualification(env["task"], verdict="success", sha=SHA, op_type="branch_ci", mode="event")
+    act = env["lc"].next_action(env["task"])
+    assert act.action == "complete_gates", act
+    env["link"].record_qualification(env["task"], verdict="failure", sha=SHA)
+    act = env["lc"].next_action(env["task"])
+    assert act.action == "repair_candidate", act
+
+
+def test_fresh_process_resumes_from_durable_state_only(tmp_path):
+    """§28: a NEW lifecycle object over the same stores — no session state —
+    reads the candidate and the recorded verdict and sequences the same
+    next action."""
+    env = _env(tmp_path)
+    env["link"].attach_explicit(env["task"], issue_number=9, contract_id=CID)
+    env["link"].update_candidate(env["task"], sha=SHA)
+    env["link"].record_qualification(env["task"], verdict="success", sha=SHA, op_type="branch_ci", mode="event")
+    # the durable contract state a fresh process would find: past local work
+    doc = env["iface"].scaffold.load()
+    for ph in ("approved", "building"):
+        env["iface"].scaffold.transition_contract(doc, CID, ph, evidence_ref="t")
+    env["iface"].scaffold.save(doc)
+    fresh = BuildLifecycle(
+        task_state=TaskStateRepository(
+            DurableEventStore(str(tmp_path / "e.jsonl")), AtomicStore(str(tmp_path / "hot"))
+        ),
+        construction_store=str(env["store"]),
+        issue_link=IssueLink(state_dir=env["issues"]),
+    )
+    state = fresh.present(env["task"])
+    assert state["candidate_sha"] == SHA
+    assert state["runner"]["state"] == "CI_QUALIFIED"
+    assert state["next_action"]["action"] == "complete_gates"