Author SHA1 Message Date
sysadmin 0a78da39e5 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
2026-07-24 17:17:05 -04:00
5 changed files with 467 additions and 1184 deletions
+81 -17
View File
@@ -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
# issue body. Title alone is never sufficient (ordinary issues may mention
# "epic" incidentally).
# unit of direct author work (#844 / #854). Matched case-insensitively against
# the issue body. Title alone is never sufficient (ordinary issues may mention
# "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
implementable issues may mention epics incidentally. A title that is
explicitly prefixed ``Epic:`` only counts when the body also proves
child-only / no-direct-implementation scope (or an epic label is present).
A bare title containing the words "epic", "roadmap", "vision", or
"umbrella" is **not** enough — ordinary implementable issues may mention
those terms incidentally. Explicit title prefixes (``Epic:``, ``Roadmap:``,
``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
# but not required — containers without the word still exclude.
# Body proves child-only / vision / roadmap / umbrella scope. Title
# 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:
detail = f"title_epic_prefix; {detail}"
if title_prefix:
detail = f"{title_prefix}; {detail}"
return True, detail
# Title-only "Epic:" without body scope evidence is insufficient (#844 AC:
# eligibility does not rely solely on the word "Epic" in a title).
# Similarly, incidental "epic" mid-title without markers stays eligible.
# Title-only coordination prefix without body scope evidence is
# insufficient (#844/#854 AC: eligibility does not rely solely on a title
# word). Incidental mid-title mentions without markers stay eligible.
return False, None
-726
View File
@@ -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,
)
+8 -58
View File
@@ -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
only the impact preview. With ``dry_run=False`` it enforces the #661 hard
gate: the apply request must present a valid, unexpired, clean drain proof
(``drain_proof_json``) or it is denied and a durable incident descriptor is
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``.
This tool is **dry-run and never restarts anything.** The mutative apply
path is a separate child gated by a drain proof (non-goal here); calling
with ``dry_run=False`` still performs no restart and reports that apply is
not yet available.
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
proof_parse_error: str | None = None
if drain_proof_json:
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
payload["reasons"] = list(payload.get("reasons") or []) + [
"apply requested but not supported: sanctioned restart apply is "
"gated by a drain proof (separate child); no restart performed (#658)"
]
return payload
-383
View File
@@ -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()
@@ -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()