fix(allocator): exclude vision/roadmap/umbrella coordination containers (#854)

#844 only caught epic-shaped child-only records. Live allocation still
selected product vision (#652), phased roadmap (#653), and umbrella (#655)
as implement targets. Extend pre-rank semantic classification with body
markers and container labels for those coordination records, keep title-
only and incidental mentions eligible, and add a live-equivalent canary.

Closes #854

allocator_service.py

@@ -133,10 +133,15 @@ ROLE_ACTIONS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
 
 
 # Body phrases that prove an issue is an implementation container, not a
-# unit of direct author work (#844). Matched case-insensitively against the
+# unit of direct author work (#844 / #854). Matched case-insensitively against
-# issue body. Title alone is never sufficient (ordinary issues may mention
+# the issue body. Title alone is never sufficient (ordinary issues may mention
-# "epic" incidentally).
+# "epic", "roadmap", "vision", or "umbrella" incidentally).
+#
+# #854 extends the #844 marker set so product-vision (#652), phased-roadmap
+# (#653), and umbrella (#655) coordination records — which do not use the word
+# "epic" — are classified with the same semantic exclusion as epic containers.
 _CHILD_ONLY_BODY_MARKERS: tuple[str, ...] = (
+    # Epic / child-only (#844, live #631)
     "implementation is delivered via child issues only",
     "implementation is delivered through child issues only",
     "implementation is delivered via child issues",
@@ -150,9 +155,26 @@ _CHILD_ONLY_BODY_MARKERS: tuple[str, ...] = (
     "coordination container",
     "child-only container",
     "implementation is delegated to child",
+    # Vision / roadmap / umbrella coordination (#854, live #652/#653/#655).
+    # Prefer authoritative non-implementation / child-only scope language over
+    # bare words like "roadmap" so ordinary implementable issues that mention
+    # a parent vision or roadmap stay eligible.
+    "do not implement features on this issue",
+    "implementing features on this roadmap issue",
+    "implementation is via linked children only",
+    "no product feature claimed complete on this issue alone",
+    "phased delivery roadmap and epic sequencing",
+    "this issue is the enduring source of truth",
+    "enduring source of truth for the",
+    "canonical product vision — enduring source of truth",
+    "canonical product vision - enduring source of truth",
+    "state: vision-active",
+    "state: roadmap-active",
 )
 
-# Explicit epic / umbrella labels (structured evidence preferred over title).
+# Explicit epic / umbrella / vision / roadmap labels (structured evidence
+# preferred over title). Tracker alone is *not* included — ordinary issues
+# may carry a tracker label without being non-implementable containers.
 _EPIC_LABELS: frozenset[str] = frozenset(
     {
         "type:epic",
@@ -161,6 +183,16 @@ _EPIC_LABELS: frozenset[str] = frozenset(
         "scope:epic",
         "type:umbrella",
         "umbrella",
+        "kind:umbrella",
+        "scope:umbrella",
+        "type:vision",
+        "vision",
+        "kind:vision",
+        "scope:vision",
+        "type:roadmap",
+        "roadmap",
+        "kind:roadmap",
+        "scope:roadmap",
     }
 )
 
@@ -222,16 +254,45 @@ class WorkCandidate:
         }
 
 
+def _title_container_prefix(title_l: str) -> str | None:
+    """Return a coordination-title prefix token if *title_l* uses one (#854).
+
+    Title prefixes alone never exclude; they only corroborate body/label
+    evidence. Ordinary issues may say "roadmap" or "vision" mid-title.
+    """
+    for prefix, token in (
+        ("epic:", "title_epic_prefix"),
+        ("epic ", "title_epic_prefix"),
+        ("umbrella:", "title_umbrella_prefix"),
+        ("umbrella ", "title_umbrella_prefix"),
+        ("roadmap:", "title_roadmap_prefix"),
+        ("roadmap ", "title_roadmap_prefix"),
+        ("product vision:", "title_vision_prefix"),
+        ("product vision ", "title_vision_prefix"),
+        ("vision:", "title_vision_prefix"),
+        ("vision ", "title_vision_prefix"),
+    ):
+        if title_l.startswith(prefix):
+            return token
+    return None
+
+
 def classify_epic_or_child_only_container(
     c: WorkCandidate,
 ) -> tuple[bool, str | None]:
-    """Return whether *c* is an epic / child-only implementation container (#844).
+    """Return whether *c* is a non-implementable coordination container (#844/#854).
 
     Exclusion uses structured evidence first (labels, body scope language).
-    A bare title containing the word "epic" is **not** enough — ordinary
+    A bare title containing the words "epic", "roadmap", "vision", or
-    implementable issues may mention epics incidentally. A title that is
+    "umbrella" is **not** enough — ordinary implementable issues may mention
-    explicitly prefixed ``Epic:`` only counts when the body also proves
+    those terms incidentally. Explicit title prefixes (``Epic:``, ``Roadmap:``,
-    child-only / no-direct-implementation scope (or an epic label is present).
+    ``Product vision:``, ``Umbrella:``) only count when the body also proves
+    child-only / no-direct-implementation scope (or a container label is
+    present).
+
+    Covers epic, product-vision, phased-roadmap, umbrella, and child-only
+    records so the allocator never assigns coordination containers as direct
+    author work.
 
     PRs are never classified as containers here (they already have a head).
     """
@@ -245,25 +306,28 @@ def classify_epic_or_child_only_container(
     title_l = title.lower()
 
     body_hits = [m for m in _CHILD_ONLY_BODY_MARKERS if m in body_l]
-    title_epic_prefix = title_l.startswith("epic:") or title_l.startswith("epic ")
+    title_prefix = _title_container_prefix(title_l)
 
     if epic_label:
         detail = f"label={epic_label[0]}"
         if body_hits:
             detail = f"{detail}; body_marker={body_hits[0]!r}"
+        if title_prefix:
+            detail = f"{title_prefix}; {detail}"
         return True, detail
 
     if body_hits:
-        # Body proves child-only / umbrella scope. Title "Epic:" is corroborating
+        # Body proves child-only / vision / roadmap / umbrella scope. Title
-        # but not required — containers without the word still exclude.
+        # prefixes are corroborating but not required — containers without the
+        # title word still exclude.
         detail = f"body_marker={body_hits[0]!r}"
-        if title_epic_prefix:
+        if title_prefix:
-            detail = f"title_epic_prefix; {detail}"
+            detail = f"{title_prefix}; {detail}"
         return True, detail
 
-    # Title-only "Epic:" without body scope evidence is insufficient (#844 AC:
+    # Title-only coordination prefix without body scope evidence is
-    # eligibility does not rely solely on the word "Epic" in a title).
+    # insufficient (#844/#854 AC: eligibility does not rely solely on a title
-    # Similarly, incidental "epic" mid-title without markers stays eligible.
+    # word). Incidental mid-title mentions without markers stay eligible.
     return False, None
 
 

drain_proof.py

@@ -1,726 +0,0 @@
-"""Pre-restart drain proof and hard gate (#661).
-
-A sanctioned MCP restart may only proceed after a machine-verifiable *drain
-proof* attests that unsafe work is clear and checkpoints are complete. Drain
-mode alone (#659) is not enough: without a proof, an apply path could still
-restart on a stale or false "ready" claim, dropping mutations and orphaning
-leases. This module defines the :class:`DrainProof` artifact, a fail-closed
-verifier, and the hard gate the sanctioned restart-apply path must consult.
-
-Design rules (mirror :mod:`restart_coordinator` / :mod:`lease_lifecycle`):
-
-* **Pure classification.** Every function here operates on already-gathered
-  inputs and returns a structured result. Nothing touches the network, the
-  filesystem, or a live process, so multi-session fixtures drive every branch.
-  This module never restarts anything; the gate only *authorizes or denies*.
-* **Fail closed.** A missing, expired, tampered, or unclean proof denies the
-  restart. Unknown checkpoint completeness is treated as *not complete*. An
-  incomplete impact report can never yield a clean proof.
-* **Non-forgeable within the process.** The proof id is a keyed hash over the
-  canonical proof contents using a per-process secret. A worker session cannot
-  hand-craft a passing proof without that secret, and a proof minted in a prior
-  daemon process will not verify after a restart (the secret is regenerated).
-* **No secrets leak.** The per-process secret never appears in a proof, an
-  ``as_dict``, an audit record, or an incident descriptor.
-
-Relationship to siblings (#655 umbrella):
-
-* **#658** ``restart_coordinator.evaluate_restart_impact`` — produces the
-  blast-radius impact report this proof consumes ("what would a restart
-  disrupt?"). ``mutations`` / ``critical_sections`` being empty is what the
-  no-in-flight-mutations check verifies.
-* **#659** graceful drain mode — performs the drain actions and calls
-  :func:`build_drain_proof` to mint the artifact once its checklist passes.
-* **#660** durable session checkpoints — supplies checkpoint completeness.
-  Because that schema may not yet be present, completeness is an *input* here,
-  never a hard table dependency; unknown fails closed.
-
-Non-goals (separate children): the emergency break-glass *workflow* (this gate
-only leaves a sanctioned bypass hole authorized elsewhere), the console UI, and
-the actual restart execution.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import hmac
-import json
-import os
-from dataclasses import dataclass
-from datetime import datetime, timedelta, timezone
-from typing import Any, Mapping, Sequence
-
-DRAIN_PROOF_VERSION = "1.0.0-issue-661"
-
-# Short default lifetime for a drain proof. A proof attests to a *point-in-time*
-# drained state; live work can resume the moment drain mode relaxes, so the
-# window in which a proof is honoured must be small (#661 security: short TTL).
-DEFAULT_PROOF_TTL_SECONDS = 120
-
-# Gate verdicts.
-GATE_ALLOW = "allow"
-GATE_DENY = "deny"
-GATE_BREAK_GLASS = "break_glass"
-
-# The mandatory drain checklist. A proof is *clean* only when every one of these
-# checks passed. Names are stable identifiers surfaced in audit + incidents.
-CHECK_NO_INFLIGHT_MUTATIONS = "no_inflight_mutations"
-CHECK_ASSIGNMENTS_STOPPED = "assignments_stopped"
-CHECK_CHECKPOINTS_COMPLETE = "checkpoints_complete"
-CHECK_HANDOFFS_OK = "handoffs_ok"
-CHECK_LEASES_HANDLED = "leases_handled"
-CHECK_ACKS_OR_TIMEOUT = "acks_or_timeout"
-
-REQUIRED_CHECKS: tuple[str, ...] = (
-    CHECK_NO_INFLIGHT_MUTATIONS,
-    CHECK_ASSIGNMENTS_STOPPED,
-    CHECK_CHECKPOINTS_COMPLETE,
-    CHECK_HANDOFFS_OK,
-    CHECK_LEASES_HANDLED,
-    CHECK_ACKS_OR_TIMEOUT,
-)
-
-
-def _utc_now() -> datetime:
-    return datetime.now(timezone.utc)
-
-
-def _parse_ts(value: str | None) -> datetime | None:
-    if not value:
-        return None
-    text = str(value).strip()
-    if not text:
-        return None
-    if text.endswith("Z"):
-        text = text[:-1] + "+00:00"
-    try:
-        dt = datetime.fromisoformat(text)
-    except ValueError:
-        return None
-    if dt.tzinfo is None:
-        dt = dt.replace(tzinfo=timezone.utc)
-    return dt
-
-
-def _canonical(payload: Any) -> str:
-    """Deterministic JSON encoding for hashing (stable key order, no spaces)."""
-
-    return json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
-
-
-# ---------------------------------------------------------------------------
-# Per-process secret. Generated once per daemon process; regenerated on restart.
-# Injectable for tests so build + verify share a secret. Never serialized.
-# ---------------------------------------------------------------------------
-_PROCESS_SECRET = os.urandom(32)
-
-
-def process_secret() -> bytes:
-    """Return the per-process proof-signing secret (never serialized)."""
-
-    return _PROCESS_SECRET
-
-
-def _resolve_secret(secret: bytes | None) -> bytes:
-    return secret if secret is not None else _PROCESS_SECRET
-
-
-@dataclass(frozen=True)
-class DrainCheck:
-    """One mandatory drain checklist result."""
-
-    name: str
-    passed: bool
-    detail: str
-
-    def as_dict(self) -> dict[str, Any]:
-        return {"name": self.name, "passed": self.passed, "detail": self.detail}
-
-
-@dataclass(frozen=True)
-class DrainProof:
-    """Machine-verifiable proof that a restart's blast radius has been drained.
-
-    The ``proof_id`` is a keyed hash over the canonical proof body; it is the
-    tamper-evident signature verified at the gate. ``clean`` is True only when
-    every required check passed. The proof is honoured only until ``expires_at``.
-    """
-
-    version: str
-    proof_id: str
-    clean: bool
-    issued_at: str
-    expires_at: str
-    requesting_session_id: str | None
-    impact_fingerprint: str
-    checks: list[DrainCheck]
-    failed_checks: list[str]
-
-    def as_dict(self) -> dict[str, Any]:
-        return {
-            "version": self.version,
-            "proof_id": self.proof_id,
-            "clean": self.clean,
-            "issued_at": self.issued_at,
-            "expires_at": self.expires_at,
-            "requesting_session_id": self.requesting_session_id,
-            "impact_fingerprint": self.impact_fingerprint,
-            "checks": [c.as_dict() for c in self.checks],
-            "failed_checks": list(self.failed_checks),
-        }
-
-
-@dataclass(frozen=True)
-class VerifyResult:
-    """Outcome of :func:`verify_drain_proof` (fail closed)."""
-
-    valid: bool
-    reasons: list[str]
-    proof_id: str | None
-    clean: bool
-    expired: bool
-    tampered: bool
-
-    def as_dict(self) -> dict[str, Any]:
-        return {
-            "valid": self.valid,
-            "reasons": list(self.reasons),
-            "proof_id": self.proof_id,
-            "clean": self.clean,
-            "expired": self.expired,
-            "tampered": self.tampered,
-        }
-
-
-@dataclass(frozen=True)
-class GateDecision:
-    """Outcome of :func:`gate_apply_restart`."""
-
-    allow: bool
-    verdict: str
-    reasons: list[str]
-    proof_id: str | None
-    break_glass: bool
-    incident: dict[str, Any] | None
-    audit_record: dict[str, Any]
-
-    def as_dict(self) -> dict[str, Any]:
-        return {
-            "allow": self.allow,
-            "verdict": self.verdict,
-            "reasons": list(self.reasons),
-            "proof_id": self.proof_id,
-            "break_glass": self.break_glass,
-            "incident": self.incident,
-            "audit_record": dict(self.audit_record),
-        }
-
-
-def impact_fingerprint(impact_report: Mapping[str, Any] | None) -> str:
-    """Stable fingerprint of the blast-radius state a proof was minted against.
-
-    Binds a proof to the specific impact evaluation. If the live state changes
-    (a new mutation appears) between minting and gate, the caller can pass the
-    fresh fingerprint and the proof will be rejected as stale.
-    """
-
-    report = impact_report or {}
-    counts = report.get("counts") or {}
-    material = {
-        "inventory_complete": bool(report.get("inventory_complete", False)),
-        "verdict": report.get("verdict"),
-        "affected_issues": sorted(report.get("affected_issues") or []),
-        "affected_prs": sorted(report.get("affected_prs") or []),
-        "mutations": sorted(
-            str(m.get("lease_id"))
-            for m in (report.get("mutations") or [])
-            if isinstance(m, Mapping)
-        ),
-        "critical_sections": sorted(
-            str(c.get("lease_id"))
-            for c in (report.get("critical_sections") or [])
-            if isinstance(c, Mapping)
-        ),
-        "terminal_lock": bool(report.get("terminal_lock")),
-        "counts": {
-            k: counts.get(k)
-            for k in (
-                "sessions_live_other",
-                "leases_disruptive",
-                "critical_sections",
-                "mutations",
-            )
-        },
-    }
-    return hashlib.sha256(_canonical(material).encode("utf-8")).hexdigest()
-
-
-def _sign(body: Mapping[str, Any], secret: bytes) -> str:
-    """Keyed (HMAC-SHA256) signature over the canonical proof body."""
-
-    return hmac.new(
-        secret, _canonical(body).encode("utf-8"), hashlib.sha256
-    ).hexdigest()
-
-
-def _proof_body(
-    *,
-    clean: bool,
-    issued_at: str,
-    expires_at: str,
-    requesting_session_id: str | None,
-    fingerprint: str,
-    checks: Sequence[DrainCheck],
-) -> dict[str, Any]:
-    """The exact fields covered by the signature. Order-independent (canonical)."""
-
-    return {
-        "version": DRAIN_PROOF_VERSION,
-        "clean": clean,
-        "issued_at": issued_at,
-        "expires_at": expires_at,
-        "requesting_session_id": requesting_session_id,
-        "impact_fingerprint": fingerprint,
-        "checks": [c.as_dict() for c in checks],
-    }
-
-
-def _bool_input(value: Any) -> bool:
-    """Strictly interpret a drain-state flag; anything not explicitly True fails."""
-
-    return value is True
-
-
-def _evaluate_checks(
-    impact_report: Mapping[str, Any],
-    drain_state: Mapping[str, Any],
-) -> list[DrainCheck]:
-    """Compute the mandatory checklist from the impact report + drain outcomes.
-
-    The report answers "is unsafe work still in flight?"; ``drain_state`` reports
-    the drain-mode actions the coordinator/#659 performed. Every check fails
-    closed when its evidence is missing or ambiguous.
-    """
-
-    checks: list[DrainCheck] = []
-
-    inventory_complete = bool(impact_report.get("inventory_complete", False))
-    mutations = list(impact_report.get("mutations") or [])
-    critical = list(impact_report.get("critical_sections") or [])
-    disruptive = [
-        l
-        for l in (impact_report.get("affected_leases") or [])
-        if isinstance(l, Mapping) and l.get("disruptive")
-    ]
-
-    # 1. No in-flight mutations. Derived from the authoritative impact report,
-    #    not self-reported: a proof cannot claim "no mutations" while the report
-    #    still shows mutations or unsevered critical sections.
-    if not inventory_complete:
-        checks.append(
-            DrainCheck(
-                CHECK_NO_INFLIGHT_MUTATIONS,
-                False,
-                "impact report inventory incomplete; cannot confirm mutations "
-                "cleared (fail closed)",
-            )
-        )
-    elif mutations or critical:
-        checks.append(
-            DrainCheck(
-                CHECK_NO_INFLIGHT_MUTATIONS,
-                False,
-                f"{len(mutations)} mutation(s) and {len(critical)} critical "
-                "section(s) still in flight",
-            )
-        )
-    else:
-        checks.append(
-            DrainCheck(
-                CHECK_NO_INFLIGHT_MUTATIONS,
-                True,
-                "no in-flight mutations or critical sections in impact report",
-            )
-        )
-
-    # 2. New assignment halted (maintenance-drain entered).
-    checks.append(
-        DrainCheck(
-            CHECK_ASSIGNMENTS_STOPPED,
-            _bool_input(drain_state.get("assignments_stopped")),
-            "new work assignment halted"
-            if _bool_input(drain_state.get("assignments_stopped"))
-            else "assignments not confirmed stopped (fail closed)",
-        )
-    )
-
-    # 3. Durable checkpoints complete (#660). Unknown => not complete.
-    cp = drain_state.get("checkpoints_complete")
-    checks.append(
-        DrainCheck(
-            CHECK_CHECKPOINTS_COMPLETE,
-            _bool_input(cp),
-            "all live sessions checkpointed"
-            if _bool_input(cp)
-            else "checkpoint completeness unconfirmed (fail closed)",
-        )
-    )
-
-    # 4. Handoffs verified.
-    checks.append(
-        DrainCheck(
-            CHECK_HANDOFFS_OK,
-            _bool_input(drain_state.get("handoffs_verified")),
-            "pending handoffs verified"
-            if _bool_input(drain_state.get("handoffs_verified"))
-            else "handoffs not verified (fail closed)",
-        )
-    )
-
-    # 5. Leases resolved/transferred/preserved AND none left disruptive. Requires
-    #    both the drain-mode assertion and the report showing no disruptive lease.
-    leases_asserted = _bool_input(drain_state.get("leases_handled"))
-    if not leases_asserted:
-        checks.append(
-            DrainCheck(
-                CHECK_LEASES_HANDLED,
-                False,
-                "lease disposition not asserted by drain (fail closed)",
-            )
-        )
-    elif disruptive:
-        checks.append(
-            DrainCheck(
-                CHECK_LEASES_HANDLED,
-                False,
-                f"{len(disruptive)} disruptive lease(s) still active in report",
-            )
-        )
-    else:
-        checks.append(
-            DrainCheck(
-                CHECK_LEASES_HANDLED,
-                True,
-                "leases resolved/transferred/preserved; none left disruptive",
-            )
-        )
-
-    # 6. Acknowledgements received, or an explicit timeout policy was applied.
-    acks = drain_state.get("acks") or {}
-    ack_values = list(acks.values()) if isinstance(acks, Mapping) else []
-    all_acked = bool(ack_values) and all(
-        str(v).strip().lower() in {"ack", "acked", "acknowledged"}
-        for v in ack_values
-    )
-    no_sessions_to_ack = isinstance(acks, Mapping) and len(ack_values) == 0
-    timeout_policy = _bool_input(drain_state.get("ack_timeout_policy_applied"))
-    acks_ok = all_acked or no_sessions_to_ack or timeout_policy
-    if acks_ok:
-        if timeout_policy and not all_acked:
-            detail = "explicit ack timeout policy applied"
-        elif no_sessions_to_ack:
-            detail = "no other live sessions required to acknowledge"
-        else:
-            detail = "all affected sessions acknowledged"
-    else:
-        detail = "outstanding acks with no timeout policy (fail closed)"
-    checks.append(DrainCheck(CHECK_ACKS_OR_TIMEOUT, acks_ok, detail))
-
-    return checks
-
-
-def build_drain_proof(
-    *,
-    impact_report: Mapping[str, Any],
-    drain_state: Mapping[str, Any],
-    requesting_session_id: str | None = None,
-    now: datetime | None = None,
-    ttl_seconds: int = DEFAULT_PROOF_TTL_SECONDS,
-    secret: bytes | None = None,
-) -> DrainProof:
-    """Mint a drain proof from an impact report and the drain-mode outcomes.
-
-    A successful drain (every checklist item passes) yields a *clean* proof with
-    a valid signature (AC#2). An unclean drain still yields a signed proof, but
-    with ``clean=False`` and the failing checks named — the gate will deny it —
-    so the artifact is auditable rather than silently absent.
-    """
-
-    moment = now or _utc_now()
-    ttl = max(1, int(ttl_seconds))
-    issued_at = moment.isoformat()
-    expires_at = (moment + timedelta(seconds=ttl)).isoformat()
-    fingerprint = impact_fingerprint(impact_report)
-
-    checks = _evaluate_checks(impact_report, drain_state)
-    failed = [c.name for c in checks if not c.passed]
-    clean = not failed
-
-    body = _proof_body(
-        clean=clean,
-        issued_at=issued_at,
-        expires_at=expires_at,
-        requesting_session_id=requesting_session_id,
-        fingerprint=fingerprint,
-        checks=checks,
-    )
-    proof_id = _sign(body, _resolve_secret(secret))
-
-    return DrainProof(
-        version=DRAIN_PROOF_VERSION,
-        proof_id=proof_id,
-        clean=clean,
-        issued_at=issued_at,
-        expires_at=expires_at,
-        requesting_session_id=requesting_session_id,
-        impact_fingerprint=fingerprint,
-        checks=checks,
-        failed_checks=failed,
-    )
-
-
-def verify_drain_proof(
-    proof: Mapping[str, Any] | None,
-    *,
-    now: datetime | None = None,
-    secret: bytes | None = None,
-    expected_impact_fingerprint: str | None = None,
-) -> VerifyResult:
-    """Verify a drain proof, failing closed on any doubt.
-
-    A proof is valid only when: it is present and well-formed; its signature
-    recomputes with the per-process secret (not forged/tampered, not minted in a
-    prior process); it has not expired; every required check is present and
-    passed; and — when ``expected_impact_fingerprint`` is supplied — it was
-    minted against the current blast-radius state.
-    """
-
-    moment = now or _utc_now()
-    reasons: list[str] = []
-
-    if not isinstance(proof, Mapping):
-        return VerifyResult(
-            valid=False,
-            reasons=["drain proof missing or not an object (fail closed)"],
-            proof_id=None,
-            clean=False,
-            expired=False,
-            tampered=False,
-        )
-
-    proof_id = proof.get("proof_id")
-    presented_clean = bool(proof.get("clean", False))
-
-    # Rebuild the signed body from the presented fields and re-sign. Any mutation
-    # of a covered field (including flipping ``clean`` to True) breaks the match.
-    raw_checks = proof.get("checks")
-    checks: list[DrainCheck] = []
-    checks_wellformed = isinstance(raw_checks, Sequence) and not isinstance(
-        raw_checks, (str, bytes)
-    )
-    if checks_wellformed:
-        for c in raw_checks:
-            if not isinstance(c, Mapping) or "name" not in c or "passed" not in c:
-                checks_wellformed = False
-                break
-            checks.append(
-                DrainCheck(
-                    name=str(c.get("name")),
-                    passed=bool(c.get("passed")),
-                    detail=str(c.get("detail") or ""),
-                )
-            )
-
-    tampered = False
-    if not checks_wellformed:
-        reasons.append("drain proof checks malformed (fail closed)")
-        tampered = True
-    else:
-        body = _proof_body(
-            clean=presented_clean,
-            issued_at=str(proof.get("issued_at") or ""),
-            expires_at=str(proof.get("expires_at") or ""),
-            requesting_session_id=proof.get("requesting_session_id"),
-            fingerprint=str(proof.get("impact_fingerprint") or ""),
-            checks=checks,
-        )
-        expected_sig = _sign(body, _resolve_secret(secret))
-        if not (
-            isinstance(proof_id, str)
-            and hmac.compare_digest(expected_sig, proof_id)
-        ):
-            tampered = True
-            reasons.append(
-                "drain proof signature mismatch: forged, tampered, or minted "
-                "in a prior process (fail closed)"
-            )
-
-    expires = _parse_ts(proof.get("expires_at"))
-    expired = expires is None or moment >= expires
-    if expires is None:
-        reasons.append("drain proof has no valid expiry (fail closed)")
-    elif expired:
-        reasons.append(f"drain proof expired at {proof.get('expires_at')}")
-
-    # Recompute cleanliness from the checks themselves — never trust the flag.
-    recomputed_failed = [c.name for c in checks if not c.passed]
-    present_names = {c.name for c in checks}
-    missing = [name for name in REQUIRED_CHECKS if name not in present_names]
-    recomputed_clean = checks_wellformed and not recomputed_failed and not missing
-    if missing:
-        reasons.append(f"drain proof missing required checks: {', '.join(missing)}")
-    if checks_wellformed and recomputed_failed:
-        reasons.append(
-            f"drain checks failed: {', '.join(sorted(set(recomputed_failed)))}"
-        )
-    if presented_clean and not recomputed_clean:
-        tampered = True
-        reasons.append("proof claims clean but its checks do not support it")
-
-    if expected_impact_fingerprint is not None:
-        if str(proof.get("impact_fingerprint") or "") != str(
-            expected_impact_fingerprint
-        ):
-            reasons.append(
-                "drain proof was minted against a different blast-radius state "
-                "(stale; fail closed)"
-            )
-
-    valid = (not tampered) and (not expired) and recomputed_clean and not reasons
-    return VerifyResult(
-        valid=valid,
-        reasons=reasons,
-        proof_id=proof_id if isinstance(proof_id, str) else None,
-        clean=recomputed_clean,
-        expired=expired,
-        tampered=tampered,
-    )
-
-
-def _incident_descriptor(
-    *,
-    reasons: Sequence[str],
-    requesting_session_id: str | None,
-    proof_id: str | None,
-    at: str,
-) -> dict[str, Any]:
-    """Durable incident descriptor for a denied restart (caller creates issue).
-
-    Kept as data (not a live Gitea call) so this module stays pure and testable;
-    the MCP tool layer turns it into a durable issue via the incident bridge.
-    """
-
-    return {
-        "kind": "restart_drain_gate_denied",
-        "title": "Restart denied: drain proof failed the hard gate",
-        "labels": ["mcp-health", "safety", "stale-runtime", "workflow-hardening"],
-        "reasons": list(reasons),
-        "requesting_session_id": requesting_session_id,
-        "proof_id": proof_id,
-        "at": at,
-        "drain_proof_version": DRAIN_PROOF_VERSION,
-    }
-
-
-def gate_apply_restart(
-    *,
-    proof: Mapping[str, Any] | None,
-    now: datetime | None = None,
-    secret: bytes | None = None,
-    break_glass: bool = False,
-    expected_impact_fingerprint: str | None = None,
-    requesting_session_id: str | None = None,
-) -> GateDecision:
-    """Hard gate for a sanctioned restart apply (#661 AC#1/#3).
-
-    Restart apply is authorized only with a valid, unexpired, clean drain proof.
-    A missing, expired, tampered, or unclean proof denies the restart and emits
-    a durable incident descriptor. ``break_glass`` is the *only* sanctioned
-    bypass — authorization for it is the caller's responsibility (the emergency
-    workflow is a separate child); when set, the gate allows without a proof but
-    records the bypass so it is never silent.
-    """
-
-    moment = now or _utc_now()
-    at = moment.isoformat()
-
-    if break_glass:
-        reasons = ["break-glass restart authorized; drain proof gate bypassed"]
-        audit = {
-            "event": "restart_gate_evaluated",
-            "drain_proof_version": DRAIN_PROOF_VERSION,
-            "evaluated_at": at,
-            "verdict": GATE_BREAK_GLASS,
-            "allow": True,
-            "break_glass": True,
-            "requesting_session_id": requesting_session_id,
-            "proof_id": None,
-        }
-        return GateDecision(
-            allow=True,
-            verdict=GATE_BREAK_GLASS,
-            reasons=reasons,
-            proof_id=None,
-            break_glass=True,
-            incident=None,
-            audit_record=audit,
-        )
-
-    result = verify_drain_proof(
-        proof,
-        now=moment,
-        secret=secret,
-        expected_impact_fingerprint=expected_impact_fingerprint,
-    )
-
-    if result.valid:
-        reasons = ["valid unexpired clean drain proof present; restart authorized"]
-        audit = {
-            "event": "restart_gate_evaluated",
-            "drain_proof_version": DRAIN_PROOF_VERSION,
-            "evaluated_at": at,
-            "verdict": GATE_ALLOW,
-            "allow": True,
-            "break_glass": False,
-            "requesting_session_id": requesting_session_id,
-            "proof_id": result.proof_id,
-        }
-        return GateDecision(
-            allow=True,
-            verdict=GATE_ALLOW,
-            reasons=reasons,
-            proof_id=result.proof_id,
-            break_glass=False,
-            incident=None,
-            audit_record=audit,
-        )
-
-    deny_reasons = ["restart denied: drain proof invalid (fail closed)"] + list(
-        result.reasons
-    )
-    incident = _incident_descriptor(
-        reasons=deny_reasons,
-        requesting_session_id=requesting_session_id,
-        proof_id=result.proof_id,
-        at=at,
-    )
-    audit = {
-        "event": "restart_gate_evaluated",
-        "drain_proof_version": DRAIN_PROOF_VERSION,
-        "evaluated_at": at,
-        "verdict": GATE_DENY,
-        "allow": False,
-        "break_glass": False,
-        "requesting_session_id": requesting_session_id,
-        "proof_id": result.proof_id,
-        "incident_kind": incident["kind"],
-    }
-    return GateDecision(
-        allow=False,
-        verdict=GATE_DENY,
-        reasons=deny_reasons,
-        proof_id=result.proof_id,
-        break_glass=False,
-        incident=incident,
-        audit_record=audit,
-    )

gitea_mcp_server.py

@@ -2068,7 +2068,6 @@ import lease_lifecycle  # noqa: E402
 import lease_policy  # noqa: E402
 import workflow_dashboard  # noqa: E402  # #605 live queue/lease dashboard
 import restart_coordinator  # noqa: E402  # #658 MCP restart coordinator/impact
-import drain_proof  # noqa: E402  # #661 pre-restart drain proof and hard gate
 import incident_bridge  # noqa: E402
 import sentry_observability  # noqa: E402  (#606 optional Sentry observability)
 import sentry_incident_bridge  # noqa: E402  (#607 Sentry→Gitea incident bridge)
@@ -22343,8 +22342,6 @@ def gitea_request_mcp_restart(
     request_override: bool = False,
     session_id: str | None = None,
     limit: int = 200,
-    drain_proof_json: str | None = None,
-    request_break_glass: bool = False,
 ) -> dict:
     """Evaluate a proposed MCP restart and return an impact preview (#658).
 
@@ -22354,15 +22351,10 @@ def gitea_request_mcp_restart(
     verdict, so the console (#642/#652) and operators can see what a restart
     would disrupt *before* any concurrent LLM work is destroyed.
 
-    This tool **never restarts a process.** In dry-run (the default) it returns
+    This tool is **dry-run and never restarts anything.** The mutative apply
-    only the impact preview. With ``dry_run=False`` it enforces the #661 hard
+    path is a separate child gated by a drain proof (non-goal here); calling
-    gate: the apply request must present a valid, unexpired, clean drain proof
+    with ``dry_run=False`` still performs no restart and reports that apply is
-    (``drain_proof_json``) or it is denied and a durable incident descriptor is
+    not yet available.
-    returned under ``incident``. Break-glass is the only bypass and is honoured
-    only when ``request_break_glass`` is set *and* the environment carries
-    ``GITEA_BREAKGLASS_RESTART_AUTHORIZATION``. Even an authorized gate performs
-    no restart here; actual execution is a further child. The gate outcome is
-    reported under ``apply_gate`` / ``apply_authorized``.
 
     Operator override authority is read from the process environment
     (``GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION``), never self-asserted by
@@ -22481,54 +22473,12 @@ def gitea_request_mcp_restart(
     payload["requesting_session_id"] = sid
     payload["operator_override_requested"] = bool(request_override)
     payload["operator_override_authorized"] = operator_authorized
-    # Actual restart execution remains a further child; this tool never restarts
-    # a process. What #661 adds is the *hard gate*: an apply request (dry_run
-    # False) must present a valid, unexpired, clean drain proof, or it is denied
-    # and a durable incident is raised. Break-glass is the only bypass and its
-    # authorization is read from the environment, never self-asserted.
     payload["apply_supported"] = False
     if not dry_run:
-        proof_obj: dict | None = None
+        payload["reasons"] = list(payload.get("reasons") or []) + [
-        proof_parse_error: str | None = None
+            "apply requested but not supported: sanctioned restart apply is "
-        if drain_proof_json:
+            "gated by a drain proof (separate child); no restart performed (#658)"
-            try:
+        ]
-                parsed = json.loads(drain_proof_json)
-                proof_obj = parsed if isinstance(parsed, dict) else None
-                if proof_obj is None:
-                    proof_parse_error = "drain_proof_json is not a JSON object"
-            except (ValueError, TypeError) as exc:
-                proof_parse_error = f"invalid drain_proof_json: {_redact(str(exc))}"
-
-        break_glass_authorized = bool(
-            (
-                os.environ.get("GITEA_BREAKGLASS_RESTART_AUTHORIZATION") or ""
-            ).strip()
-        )
-        break_glass = bool(request_break_glass and break_glass_authorized)
-
-        expected_fp = drain_proof.impact_fingerprint(report.as_dict())
-        gate = drain_proof.gate_apply_restart(
-            proof=proof_obj,
-            break_glass=break_glass,
-            expected_impact_fingerprint=expected_fp,
-            requesting_session_id=sid,
-        )
-        gate_payload = gate.as_dict()
-        if proof_parse_error and not break_glass:
-            gate_payload["reasons"] = [proof_parse_error] + list(
-                gate_payload.get("reasons") or []
-            )
-        payload["apply_gate"] = gate_payload
-        payload["apply_authorized"] = gate.allow
-        payload["break_glass_requested"] = bool(request_break_glass)
-        payload["break_glass_authorized"] = break_glass_authorized
-        # Even an authorized gate performs no restart here: execution is a later
-        # child. The gate proves the apply path *would* be permitted.
-        payload["reasons"] = list(payload.get("reasons") or []) + list(
-            gate_payload.get("reasons") or []
-        )
-        if not gate.allow and gate.incident is not None:
-            payload["incident"] = gate.incident
     return payload
 
 

tests/test_drain_proof.py

@@ -1,383 +0,0 @@
-"""Tests for the pre-restart drain proof and hard gate (#661).
-
-Covers the acceptance criteria:
-
-1. Restart apply without a proof fails closed.
-2. A successful drain produces a verifiable proof.
-3. An open unsafe mutation makes the proof fail (multi-session fixture).
-4. Pass / fail / expired verification paths.
-
-Plus the security posture: forged/tampered proofs are rejected, break-glass is
-the only bypass and is never silent, a stale blast-radius fingerprint rejects a
-proof, and no per-process secret ever leaks into a serialized artifact.
-"""
-
-from __future__ import annotations
-
-import os
-import unittest
-from datetime import datetime, timedelta, timezone
-
-import drain_proof as dp
-import restart_coordinator as rc
-
-
-NOW = datetime(2026, 7, 24, 6, 0, 0, tzinfo=timezone.utc)
-SECRET = b"unit-test-drain-proof-secret-0123456789abcdef"
-
-
-def _live_pid() -> int:
-    return os.getpid()
-
-
-def _clean_drain_state() -> dict:
-    """Every drain action succeeded, no sessions outstanding."""
-
-    return {
-        "assignments_stopped": True,
-        "checkpoints_complete": True,
-        "handoffs_verified": True,
-        "leases_handled": True,
-        "acks": {},  # no other live sessions to acknowledge
-        "ack_timeout_policy_applied": False,
-    }
-
-
-def _safe_report() -> dict:
-    """Impact report with no other live work: a restart here is safe."""
-
-    report = rc.evaluate_restart_impact(
-        {"sessions": [], "leases": [], "inventory_complete": True},
-        now=NOW,
-        requesting_session_id="prgs-controller-1-req",
-    )
-    return report.as_dict()
-
-
-def _unsafe_mutation_report() -> dict:
-    """Multi-session report: a second session holds a live author mutation."""
-
-    sessions = [
-        {
-            "session_id": "prgs-controller-1-req",
-            "role": "controller",
-            "profile": "prgs-controller",
-            "pid": _live_pid(),
-            "status": "active",
-            "last_heartbeat_at": NOW.isoformat(),
-        },
-        {
-            "session_id": "prgs-author-99",
-            "role": "author",
-            "profile": "prgs-author",
-            "pid": _live_pid(),
-            "status": "active",
-            "last_heartbeat_at": NOW.isoformat(),
-        },
-    ]
-    leases = [
-        {
-            "lease_id": "lease-mut",
-            "session_id": "prgs-author-99",
-            "role": "author",
-            "phase": "implementing",
-            "work_kind": "issue",
-            "work_number": 661,
-            "worktree_path": "branches/issue-661",
-            "freshness": {"freshness": "active"},
-        }
-    ]
-    report = rc.evaluate_restart_impact(
-        {"sessions": sessions, "leases": leases, "inventory_complete": True},
-        now=NOW,
-        requesting_session_id="prgs-controller-1-req",
-    )
-    return report.as_dict()
-
-
-class BuildDrainProofTests(unittest.TestCase):
-    def test_clean_drain_produces_verifiable_clean_proof(self):
-        """AC#2: a successful drain produces a verifiable proof."""
-
-        proof = dp.build_drain_proof(
-            impact_report=_safe_report(),
-            drain_state=_clean_drain_state(),
-            requesting_session_id="prgs-controller-1-req",
-            now=NOW,
-            secret=SECRET,
-        )
-        self.assertTrue(proof.clean)
-        self.assertEqual(proof.failed_checks, [])
-        self.assertEqual(
-            {c.name for c in proof.checks}, set(dp.REQUIRED_CHECKS)
-        )
-        result = dp.verify_drain_proof(
-            proof.as_dict(), now=NOW, secret=SECRET
-        )
-        self.assertTrue(result.valid, result.reasons)
-        self.assertFalse(result.expired)
-        self.assertFalse(result.tampered)
-
-    def test_open_mutation_makes_proof_unclean(self):
-        """AC#3: an unsafe mutation still in flight fails the proof."""
-
-        proof = dp.build_drain_proof(
-            impact_report=_unsafe_mutation_report(),
-            drain_state=_clean_drain_state(),
-            now=NOW,
-            secret=SECRET,
-        )
-        self.assertFalse(proof.clean)
-        self.assertIn(dp.CHECK_NO_INFLIGHT_MUTATIONS, proof.failed_checks)
-        # Leases-handled also fails: the report still shows a disruptive lease.
-        self.assertIn(dp.CHECK_LEASES_HANDLED, proof.failed_checks)
-        result = dp.verify_drain_proof(proof.as_dict(), now=NOW, secret=SECRET)
-        self.assertFalse(result.valid)
-
-    def test_incomplete_inventory_fails_no_mutations_check(self):
-        proof = dp.build_drain_proof(
-            impact_report={"inventory_complete": False},
-            drain_state=_clean_drain_state(),
-            now=NOW,
-            secret=SECRET,
-        )
-        self.assertFalse(proof.clean)
-        self.assertIn(dp.CHECK_NO_INFLIGHT_MUTATIONS, proof.failed_checks)
-
-    def test_missing_checkpoint_flag_fails_closed(self):
-        state = _clean_drain_state()
-        del state["checkpoints_complete"]
-        proof = dp.build_drain_proof(
-            impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
-        )
-        self.assertFalse(proof.clean)
-        self.assertIn(dp.CHECK_CHECKPOINTS_COMPLETE, proof.failed_checks)
-
-    def test_non_true_flags_fail_closed(self):
-        """A truthy-but-not-True value (e.g. the string 'yes') must not pass."""
-
-        state = _clean_drain_state()
-        state["assignments_stopped"] = "yes"
-        proof = dp.build_drain_proof(
-            impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
-        )
-        self.assertIn(dp.CHECK_ASSIGNMENTS_STOPPED, proof.failed_checks)
-
-    def test_ack_timeout_policy_satisfies_ack_check(self):
-        state = _clean_drain_state()
-        state["acks"] = {"prgs-author-99": "pending"}
-        state["ack_timeout_policy_applied"] = True
-        proof = dp.build_drain_proof(
-            impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
-        )
-        names = {c.name: c.passed for c in proof.checks}
-        self.assertTrue(names[dp.CHECK_ACKS_OR_TIMEOUT])
-
-    def test_outstanding_acks_without_timeout_fail(self):
-        state = _clean_drain_state()
-        state["acks"] = {"prgs-author-99": "pending"}
-        state["ack_timeout_policy_applied"] = False
-        proof = dp.build_drain_proof(
-            impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
-        )
-        self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks)
-
-    def test_all_acked_satisfies_ack_check(self):
-        state = _clean_drain_state()
-        state["acks"] = {"prgs-author-99": "acked", "prgs-author-2": "acknowledged"}
-        proof = dp.build_drain_proof(
-            impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
-        )
-        names = {c.name: c.passed for c in proof.checks}
-        self.assertTrue(names[dp.CHECK_ACKS_OR_TIMEOUT])
-
-
-class VerifyDrainProofTests(unittest.TestCase):
-    def _clean_proof_dict(self) -> dict:
-        return dp.build_drain_proof(
-            impact_report=_safe_report(),
-            drain_state=_clean_drain_state(),
-            now=NOW,
-            secret=SECRET,
-        ).as_dict()
-
-    def test_missing_proof_is_invalid(self):
-        result = dp.verify_drain_proof(None, now=NOW, secret=SECRET)
-        self.assertFalse(result.valid)
-        self.assertIsNone(result.proof_id)
-
-    def test_expired_proof_is_invalid(self):
-        """AC#4: an expired proof fails verification."""
-
-        proof = self._clean_proof_dict()
-        later = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS + 1)
-        result = dp.verify_drain_proof(proof, now=later, secret=SECRET)
-        self.assertFalse(result.valid)
-        self.assertTrue(result.expired)
-
-    def test_proof_valid_just_before_expiry(self):
-        proof = self._clean_proof_dict()
-        almost = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS - 1)
-        result = dp.verify_drain_proof(proof, now=almost, secret=SECRET)
-        self.assertTrue(result.valid, result.reasons)
-
-    def test_wrong_secret_rejected(self):
-        """A proof minted in a prior process (different secret) will not verify."""
-
-        proof = self._clean_proof_dict()
-        result = dp.verify_drain_proof(proof, now=NOW, secret=b"other-secret")
-        self.assertFalse(result.valid)
-        self.assertTrue(result.tampered)
-
-    def test_flipping_clean_flag_is_detected(self):
-        """Forging clean=True on an unclean proof breaks the signature."""
-
-        unclean = dp.build_drain_proof(
-            impact_report=_unsafe_mutation_report(),
-            drain_state=_clean_drain_state(),
-            now=NOW,
-            secret=SECRET,
-        ).as_dict()
-        self.assertFalse(unclean["clean"])
-        unclean["clean"] = True  # forge
-        result = dp.verify_drain_proof(unclean, now=NOW, secret=SECRET)
-        self.assertFalse(result.valid)
-        self.assertTrue(result.tampered)
-
-    def test_tampering_a_check_is_detected(self):
-        unclean = dp.build_drain_proof(
-            impact_report=_unsafe_mutation_report(),
-            drain_state=_clean_drain_state(),
-            now=NOW,
-            secret=SECRET,
-        ).as_dict()
-        for c in unclean["checks"]:
-            if c["name"] == dp.CHECK_NO_INFLIGHT_MUTATIONS:
-                c["passed"] = True  # forge the failing check to pass
-        result = dp.verify_drain_proof(unclean, now=NOW, secret=SECRET)
-        self.assertFalse(result.valid)
-        self.assertTrue(result.tampered)
-
-    def test_missing_required_check_rejected(self):
-        proof = self._clean_proof_dict()
-        proof["checks"] = [
-            c for c in proof["checks"] if c["name"] != dp.CHECK_HANDOFFS_OK
-        ]
-        result = dp.verify_drain_proof(proof, now=NOW, secret=SECRET)
-        self.assertFalse(result.valid)
-
-    def test_stale_fingerprint_rejected(self):
-        proof = self._clean_proof_dict()
-        result = dp.verify_drain_proof(
-            proof,
-            now=NOW,
-            secret=SECRET,
-            expected_impact_fingerprint="deadbeef",
-        )
-        self.assertFalse(result.valid)
-
-    def test_matching_fingerprint_accepted(self):
-        report = _safe_report()
-        proof = dp.build_drain_proof(
-            impact_report=report,
-            drain_state=_clean_drain_state(),
-            now=NOW,
-            secret=SECRET,
-        ).as_dict()
-        fp = dp.impact_fingerprint(report)
-        result = dp.verify_drain_proof(
-            proof, now=NOW, secret=SECRET, expected_impact_fingerprint=fp
-        )
-        self.assertTrue(result.valid, result.reasons)
-
-
-class GateApplyRestartTests(unittest.TestCase):
-    def _clean_proof_dict(self) -> dict:
-        return dp.build_drain_proof(
-            impact_report=_safe_report(),
-            drain_state=_clean_drain_state(),
-            now=NOW,
-            secret=SECRET,
-        ).as_dict()
-
-    def test_apply_without_proof_denied(self):
-        """AC#1: restart apply without a proof fails closed + raises incident."""
-
-        decision = dp.gate_apply_restart(proof=None, now=NOW, secret=SECRET)
-        self.assertFalse(decision.allow)
-        self.assertEqual(decision.verdict, dp.GATE_DENY)
-        self.assertIsNotNone(decision.incident)
-        self.assertEqual(
-            decision.incident["kind"], "restart_drain_gate_denied"
-        )
-
-    def test_apply_with_valid_proof_allowed(self):
-        decision = dp.gate_apply_restart(
-            proof=self._clean_proof_dict(), now=NOW, secret=SECRET
-        )
-        self.assertTrue(decision.allow)
-        self.assertEqual(decision.verdict, dp.GATE_ALLOW)
-        self.assertIsNone(decision.incident)
-
-    def test_apply_with_expired_proof_denied_with_incident(self):
-        later = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS + 5)
-        decision = dp.gate_apply_restart(
-            proof=self._clean_proof_dict(), now=later, secret=SECRET
-        )
-        self.assertFalse(decision.allow)
-        self.assertIsNotNone(decision.incident)
-
-    def test_apply_with_unclean_proof_denied(self):
-        """AC#3 at the gate: an unsafe-mutation proof is denied."""
-
-        unclean = dp.build_drain_proof(
-            impact_report=_unsafe_mutation_report(),
-            drain_state=_clean_drain_state(),
-            now=NOW,
-            secret=SECRET,
-        ).as_dict()
-        decision = dp.gate_apply_restart(proof=unclean, now=NOW, secret=SECRET)
-        self.assertFalse(decision.allow)
-        self.assertIsNotNone(decision.incident)
-
-    def test_break_glass_allows_without_proof_but_records_bypass(self):
-        decision = dp.gate_apply_restart(
-            proof=None, now=NOW, secret=SECRET, break_glass=True
-        )
-        self.assertTrue(decision.allow)
-        self.assertEqual(decision.verdict, dp.GATE_BREAK_GLASS)
-        self.assertTrue(decision.break_glass)
-        self.assertIsNone(decision.incident)
-        self.assertTrue(decision.audit_record["break_glass"])
-
-    def test_denied_gate_carries_stale_fingerprint_reason(self):
-        decision = dp.gate_apply_restart(
-            proof=self._clean_proof_dict(),
-            now=NOW,
-            secret=SECRET,
-            expected_impact_fingerprint="not-the-fingerprint",
-        )
-        self.assertFalse(decision.allow)
-
-
-class SecretHygieneTests(unittest.TestCase):
-    def test_secret_never_serialized(self):
-        proof = dp.build_drain_proof(
-            impact_report=_safe_report(),
-            drain_state=_clean_drain_state(),
-            now=NOW,
-            secret=SECRET,
-        )
-        blob = dp._canonical(proof.as_dict())
-        self.assertNotIn(SECRET.decode(), blob)
-        # The signature is a hex digest, not the raw secret.
-        self.assertNotIn(SECRET.hex(), blob)
-
-    def test_incident_descriptor_has_no_secret(self):
-        decision = dp.gate_apply_restart(proof=None, now=NOW, secret=SECRET)
-        blob = dp._canonical(decision.incident)
-        self.assertNotIn(SECRET.decode(), blob)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_issue_854_semantic_container_exclusion.py

@@ -0,0 +1,378 @@
+"""Allocator semantic container exclusion for vision/roadmap/umbrella (#854).
+
+#844 excluded epic / child-only containers (live #631) but product-vision
+(#652), phased-roadmap (#653), and umbrella (#655) coordination records still
+ranked as implementable work. This module is the live-equivalent canary:
+
+* #631 / #652 / #653 / #655-shaped records are all excluded in one inventory.
+* Independently executable children remain eligible and can be selected.
+* Ordinary issues that merely mention vision / roadmap / umbrella stay eligible.
+* Excluded containers never receive assignments or workflow leases.
+* Structured skip reason ``epic_or_child_only_container`` is reported.
+* Candidate-set fingerprint remains stable after exclusions.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+
+from allocator_service import (
+    OUTCOME_ASSIGNED,
+    OUTCOME_PREVIEW,
+    SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
+    WorkCandidate,
+    allocate_next_work,
+    candidate_set_fingerprint,
+    classify_epic_or_child_only_container,
+)
+from control_plane_db import ControlPlaneDB
+
+REMOTE = "prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+
+# Minimal bodies mirroring live coordination records (not full issue text).
+_EPIC_631_BODY = """
+## Scope (umbrella)
+
+This epic owns the **product roadmap and linkage** for the Web Console.
+Implementation is delivered via child issues only.
+
+## Explicit non-goals
+
+* Do not implement product features in this epic issue itself.
+* No product feature implementation is claimed complete solely on this epic.
+"""
+
+_VISION_652_BODY = """
+## Canonical product vision — enduring source of truth
+
+**This issue is the enduring source of truth for the MCP Control Plane Web Console product vision.**
+
+## Implementation linkage
+
+* **Do not implement features on this issue.**
+* Sequencing: roadmap issue + #631 children.
+
+## Canonical issue state
+
+```text
+STATE: vision-active
+WHO_IS_NEXT: controller (triage/ordering) / author (implementation of linked children only)
+```
+"""
+
+_ROADMAP_653_BODY = """
+## Purpose
+
+This issue is the **phased delivery roadmap and epic sequencing** for the MCP Control Plane Web Console.
+
+## Non-goals
+
+* Implementing features on this roadmap issue.
+* Deleting vision items by omitting them from phases without #652 change log.
+
+## Canonical issue state
+
+```text
+STATE: roadmap-active
+WHO_IS_NEXT: author
+```
+"""
+
+_UMBRELLA_655_BODY = """
+## Scope (umbrella)
+
+This issue owns the **canonical restart-governance program**. Implementation is via linked children only.
+
+## Acceptance criteria (umbrella)
+
+6. No product feature claimed complete on this issue alone.
+"""
+
+_CHILD_BODY = """
+## Problem
+
+Operators need a workflow-event timeline model for Phase 1.
+
+## Acceptance criteria
+
+- [ ] Timeline model API exists
+"""
+
+
+def _issue(
+    number: int,
+    *,
+    title: str = "",
+    body: str = "",
+    labels: tuple[str, ...] = ("status:ready", "type:feature"),
+    priority: int = 20,
+) -> WorkCandidate:
+    return WorkCandidate(
+        kind="issue",
+        number=number,
+        state="open",
+        labels=labels,
+        title=title or f"issue {number}",
+        body=body,
+        priority=priority,
+    )
+
+
+def _live_shaped_containers() -> list[WorkCandidate]:
+    return [
+        _issue(
+            631,
+            title="Epic: MCP Control Plane Web Console",
+            body=_EPIC_631_BODY,
+        ),
+        _issue(
+            652,
+            title="Product vision: MCP Control Plane Web Console (canonical)",
+            body=_VISION_652_BODY,
+        ),
+        _issue(
+            653,
+            title="Roadmap: MCP Control Plane Web Console (phased delivery)",
+            body=_ROADMAP_653_BODY,
+        ),
+        _issue(
+            655,
+            title="Umbrella: Governed MCP restart coordination and zero-disruption recovery",
+            body=_UMBRELLA_655_BODY,
+        ),
+    ]
+
+
+class ClassifySemanticContainersTest(unittest.TestCase):
+    def test_652_vision_is_container(self) -> None:
+        c = _issue(
+            652,
+            title="Product vision: MCP Control Plane Web Console (canonical)",
+            body=_VISION_652_BODY,
+        )
+        is_c, detail = classify_epic_or_child_only_container(c)
+        self.assertTrue(is_c)
+        self.assertIsNotNone(detail)
+        self.assertIn("body_marker", detail or "")
+
+    def test_653_roadmap_is_container(self) -> None:
+        c = _issue(
+            653,
+            title="Roadmap: MCP Control Plane Web Console (phased delivery)",
+            body=_ROADMAP_653_BODY,
+        )
+        is_c, detail = classify_epic_or_child_only_container(c)
+        self.assertTrue(is_c)
+        self.assertIn("body_marker", detail or "")
+
+    def test_655_umbrella_is_container(self) -> None:
+        c = _issue(
+            655,
+            title="Umbrella: Governed MCP restart coordination and zero-disruption recovery",
+            body=_UMBRELLA_655_BODY,
+        )
+        is_c, detail = classify_epic_or_child_only_container(c)
+        self.assertTrue(is_c)
+        self.assertIn("body_marker", detail or "")
+
+    def test_631_still_container_after_854(self) -> None:
+        c = _issue(
+            631,
+            title="Epic: MCP Control Plane Web Console",
+            body=_EPIC_631_BODY,
+        )
+        is_c, detail = classify_epic_or_child_only_container(c)
+        self.assertTrue(is_c)
+        self.assertIn("body_marker", detail or "")
+
+    def test_incidental_vision_roadmap_umbrella_words_not_container(self) -> None:
+        cases = (
+            (
+                "Document vision handoff conventions",
+                "Update docs so implementable issues that mention a vision "
+                "remain independently executable.",
+            ),
+            (
+                "Clarify roadmap sequencing notes",
+                "Write a short note about how the roadmap issue relates to children.",
+            ),
+            (
+                "Umbrella recovery checklist for authors",
+                "Authors should still implement the concrete recovery fix here.",
+            ),
+            (
+                "Product vision wording in the help text",
+                "Fix a typo in the operator-facing help string that says product vision.",
+            ),
+        )
+        for title, body in cases:
+            with self.subTest(title=title):
+                c = _issue(900, title=title, body=body)
+                is_c, detail = classify_epic_or_child_only_container(c)
+                self.assertFalse(is_c)
+                self.assertIsNone(detail)
+
+    def test_title_prefix_alone_not_container(self) -> None:
+        for title in (
+            "Epic: something mentioned only in title",
+            "Roadmap: title only without body scope",
+            "Product vision: title only without body scope",
+            "Umbrella: title only without body scope",
+        ):
+            with self.subTest(title=title):
+                c = _issue(
+                    901,
+                    title=title,
+                    body="Implement a concrete fix for the allocator skip list.",
+                )
+                is_c, detail = classify_epic_or_child_only_container(c)
+                self.assertFalse(is_c)
+                self.assertIsNone(detail)
+
+    def test_roadmap_label_alone_is_container(self) -> None:
+        c = _issue(
+            902,
+            title="Console delivery sequencing",
+            body="Track phased delivery only.",
+            labels=("status:ready", "type:roadmap"),
+        )
+        is_c, detail = classify_epic_or_child_only_container(c)
+        self.assertTrue(is_c)
+        self.assertIn("type:roadmap", detail or "")
+
+    def test_child_referencing_parent_policy_stays_eligible(self) -> None:
+        """Children may quote parent policy without becoming containers."""
+        c = _issue(
+            637,
+            title="Web Console: Workflow-event timeline model (Phase 1)",
+            body=(
+                _CHILD_BODY
+                + "\n\nParent #652 says do not implement on the vision issue; "
+                "this child is the implementable unit."
+            ),
+        )
+        is_c, _ = classify_epic_or_child_only_container(c)
+        self.assertFalse(is_c)
+
+
+class AllocateSemanticContainerExclusionTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def _alloc(self, candidates, **kwargs):
+        defaults = dict(
+            session_id="sess-854",
+            role="author",
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            profile_name="prgs-author",
+            username="jcwalker3",
+            claims={},
+            apply=False,
+        )
+        defaults.update(kwargs)
+        return allocate_next_work(self.db, candidates=candidates, **defaults)
+
+    def test_live_equivalent_canary_excludes_all_containers_selects_child(self) -> None:
+        containers = _live_shaped_containers()
+        child = _issue(
+            637,
+            title="Web Console: Workflow-event timeline model (Phase 1)",
+            body=_CHILD_BODY,
+        )
+        inventory = containers + [child]
+        res = self._alloc(inventory, apply=False)
+        self.assertTrue(res["success"], res)
+        self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(res["selected"]["number"], 637)
+
+        skipped = {s["number"]: s for s in res["skipped"]}
+        for number in (631, 652, 653, 655):
+            self.assertIn(number, skipped, res["skipped"])
+            self.assertEqual(
+                skipped[number]["reason_code"],
+                SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
+            )
+            self.assertIn(
+                SKIP_EPIC_OR_CHILD_ONLY_CONTAINER, skipped[number]["reason"]
+            )
+
+    def test_containers_cannot_receive_assignment_or_lease(self) -> None:
+        containers = _live_shaped_containers()
+        res = self._alloc(containers, apply=True)
+        self.assertTrue(res["success"], res)
+        self.assertNotEqual(res["outcome"], OUTCOME_ASSIGNED)
+        self.assertIsNone(res.get("assignment"))
+        self.assertIsNone(res.get("selected"))
+        skipped = {s["number"]: s for s in res["skipped"]}
+        for number in (631, 652, 653, 655):
+            self.assertEqual(
+                skipped[number]["reason_code"],
+                SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
+            )
+
+        leases = []
+        if hasattr(self.db, "list_active_leases"):
+            leases = self.db.list_active_leases(
+                remote=REMOTE, org=ORG, repo=REPO
+            )
+        if not leases and hasattr(self.db, "list_leases"):
+            leases = self.db.list_leases(remote=REMOTE, org=ORG, repo=REPO)
+        for lease in leases or []:
+            work_number = (
+                lease.get("work_number") if isinstance(lease, dict) else None
+            )
+            self.assertNotIn(work_number, {631, 652, 653, 655})
+
+    def test_apply_selects_child_not_container(self) -> None:
+        containers = _live_shaped_containers()
+        child = _issue(
+            637,
+            title="Web Console: Workflow-event timeline model (Phase 1)",
+            body=_CHILD_BODY,
+        )
+        res = self._alloc(containers + [child], apply=True)
+        self.assertTrue(res["success"], res)
+        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
+        self.assertEqual(res["selected"]["number"], 637)
+        self.assertEqual(res["assignment"]["work_number"], 637)
+
+    def test_fingerprint_stable_with_containers_present(self) -> None:
+        containers = _live_shaped_containers()
+        child = _issue(
+            637,
+            title="Web Console: Workflow-event timeline model (Phase 1)",
+            body=_CHILD_BODY,
+        )
+        inventory = containers + [child]
+        fp_before = candidate_set_fingerprint(inventory)
+        res = self._alloc(inventory, apply=False)
+        self.assertTrue(res["success"], res)
+        self.assertEqual(res["selected"]["number"], 637)
+        # Allocator reports the same CAS fingerprint for the full candidate set.
+        reported = res.get("candidate_set_fingerprint")
+        self.assertEqual(reported, fp_before)
+        # Re-fingerprint of the same inventory is byte-stable.
+        self.assertEqual(candidate_set_fingerprint(inventory), fp_before)
+
+    def test_incidental_mentions_remain_eligible(self) -> None:
+        ordinary = _issue(
+            700,
+            title="Document roadmap handoff conventions",
+            body="Write runbook text about vision vs roadmap vs child issues.",
+        )
+        res = self._alloc([ordinary], apply=False)
+        self.assertTrue(res["success"], res)
+        self.assertEqual(res["selected"]["number"], 700)
+        self.assertEqual(res["skipped"], [])
+
+
+if __name__ == "__main__":
+    unittest.main()