Compare commits
12
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3a9d634c17 | ||
|
|
2068bae341 | ||
|
|
7af40fb5ff | ||
|
|
9517834913 | ||
|
|
824c42f7e3 | ||
|
|
578c44b685 | ||
|
|
3b68d15593 | ||
|
|
a4c73766f4 | ||
|
|
9f686253eb | ||
|
|
b2e28428a4 | ||
|
|
1cbbde0089 | ||
|
|
0a78da39e5 |
+81
-17
@@ -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
|
||||
|
||||
|
||||
|
||||
+1015
File diff suppressed because it is too large
Load Diff
+58
-8
@@ -2068,6 +2068,7 @@ 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)
|
||||
@@ -22342,6 +22343,8 @@ 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).
|
||||
|
||||
@@ -22351,10 +22354,15 @@ 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 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.
|
||||
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``.
|
||||
|
||||
Operator override authority is read from the process environment
|
||||
(``GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION``), never self-asserted by
|
||||
@@ -22473,12 +22481,54 @@ 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:
|
||||
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)"
|
||||
]
|
||||
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
|
||||
return payload
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,907 @@
|
||||
"""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)
|
||||
|
||||
|
||||
def _drained_report_with_live_sessions(count: int) -> dict:
|
||||
"""Report with ``count`` other live sessions but nothing in flight.
|
||||
|
||||
Every other checklist item passes against this report, so a failure
|
||||
isolates the acknowledgement check rather than tripping on mutations.
|
||||
"""
|
||||
|
||||
sessions = [
|
||||
{
|
||||
"session_id": "prgs-controller-1-req",
|
||||
"role": "controller",
|
||||
"profile": "prgs-controller",
|
||||
"pid": _live_pid(),
|
||||
"status": "active",
|
||||
"last_heartbeat_at": NOW.isoformat(),
|
||||
}
|
||||
]
|
||||
for index in range(count):
|
||||
sessions.append(
|
||||
{
|
||||
"session_id": f"prgs-author-{index}",
|
||||
"role": "author",
|
||||
"profile": "prgs-author",
|
||||
"pid": _live_pid(),
|
||||
"status": "active",
|
||||
"last_heartbeat_at": NOW.isoformat(),
|
||||
}
|
||||
)
|
||||
report = rc.evaluate_restart_impact(
|
||||
{"sessions": sessions, "leases": [], "inventory_complete": True},
|
||||
now=NOW,
|
||||
requesting_session_id="prgs-controller-1-req",
|
||||
)
|
||||
return report.as_dict()
|
||||
|
||||
|
||||
class AcknowledgementFailClosedTests(unittest.TestCase):
|
||||
"""Acknowledgement evidence must fail closed unless explicitly verified.
|
||||
|
||||
Regression cover for the reviewed fail-open on PR #882: an absent ``acks``
|
||||
key collapsed to ``{}`` and was read as "no other live sessions required to
|
||||
acknowledge", so a proof minted clean and the restart gate allowed while the
|
||||
impact report still showed other live sessions.
|
||||
"""
|
||||
|
||||
def _state(self, **overrides) -> dict:
|
||||
state = _clean_drain_state()
|
||||
state.pop("acks", None)
|
||||
state["ack_timeout_policy_applied"] = False
|
||||
state.update(overrides)
|
||||
return state
|
||||
|
||||
def _acks_check(self, proof) -> dp.DrainCheck:
|
||||
return next(c for c in proof.checks if c.name == dp.CHECK_ACKS_OR_TIMEOUT)
|
||||
|
||||
def _build(self, report: dict, state: dict):
|
||||
return dp.build_drain_proof(
|
||||
impact_report=report, drain_state=state, now=NOW, secret=SECRET
|
||||
)
|
||||
|
||||
def assertAcksFailClosed(self, report: dict, state: dict) -> None:
|
||||
proof = self._build(report, state)
|
||||
self.assertFalse(self._acks_check(proof).passed)
|
||||
self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks)
|
||||
self.assertFalse(proof.clean)
|
||||
|
||||
# --- missing / null / empty / malformed ------------------------------
|
||||
|
||||
def test_missing_acks_key_with_live_sessions_fails_closed(self):
|
||||
"""The exact reviewed defect: absent key, three other live sessions."""
|
||||
report = _drained_report_with_live_sessions(3)
|
||||
self.assertEqual(report["counts"]["sessions_live_other"], 3)
|
||||
state = self._state()
|
||||
self.assertNotIn("acks", state)
|
||||
proof = self._build(report, state)
|
||||
check = self._acks_check(proof)
|
||||
self.assertFalse(check.passed)
|
||||
self.assertNotIn("no other live sessions", check.detail)
|
||||
self.assertIn("fail closed", check.detail)
|
||||
self.assertFalse(proof.clean)
|
||||
self.assertEqual(proof.failed_checks, [dp.CHECK_ACKS_OR_TIMEOUT])
|
||||
|
||||
def test_none_acks_with_live_sessions_fails_closed(self):
|
||||
self.assertAcksFailClosed(
|
||||
_drained_report_with_live_sessions(2), self._state(acks=None)
|
||||
)
|
||||
|
||||
def test_empty_acks_with_live_sessions_fails_closed(self):
|
||||
self.assertAcksFailClosed(
|
||||
_drained_report_with_live_sessions(1), self._state(acks={})
|
||||
)
|
||||
|
||||
def test_malformed_acks_fail_closed(self):
|
||||
for malformed in ([], "ack", 7, ("ack",), True):
|
||||
with self.subTest(malformed=malformed):
|
||||
self.assertAcksFailClosed(
|
||||
_drained_report_with_live_sessions(1),
|
||||
self._state(acks=malformed),
|
||||
)
|
||||
|
||||
# --- stale / unproven values -----------------------------------------
|
||||
|
||||
def test_stale_or_unproven_ack_values_fail_closed(self):
|
||||
for value in ("pending", "stale", "unknown", "", None, True, 1, NOW):
|
||||
with self.subTest(value=value):
|
||||
self.assertAcksFailClosed(
|
||||
_drained_report_with_live_sessions(1),
|
||||
self._state(acks={"prgs-author-0": value}),
|
||||
)
|
||||
|
||||
def test_partial_coverage_fails_closed(self):
|
||||
"""Fewer acknowledgements than the report's live-session count."""
|
||||
self.assertAcksFailClosed(
|
||||
_drained_report_with_live_sessions(3),
|
||||
self._state(acks={"prgs-author-0": "ack"}),
|
||||
)
|
||||
|
||||
def test_one_unacked_entry_among_many_fails_closed(self):
|
||||
self.assertAcksFailClosed(
|
||||
_drained_report_with_live_sessions(2),
|
||||
self._state(acks={"prgs-author-0": "ack", "prgs-author-1": "pending"}),
|
||||
)
|
||||
|
||||
def test_unproven_live_session_count_fails_closed(self):
|
||||
"""A missing or malformed count cannot prove nobody had to acknowledge."""
|
||||
malformed_counts = (
|
||||
None,
|
||||
{},
|
||||
{"sessions_live_other": None},
|
||||
{"sessions_live_other": "3"},
|
||||
{"sessions_live_other": -1},
|
||||
{"sessions_live_other": True},
|
||||
)
|
||||
for counts in malformed_counts:
|
||||
with self.subTest(counts=counts):
|
||||
report = _drained_report_with_live_sessions(0)
|
||||
if counts is None:
|
||||
report.pop("counts", None)
|
||||
else:
|
||||
report["counts"] = counts
|
||||
self.assertAcksFailClosed(report, self._state())
|
||||
|
||||
# --- valid evidence still passes -------------------------------------
|
||||
|
||||
def test_complete_valid_acks_pass(self):
|
||||
report = _drained_report_with_live_sessions(2)
|
||||
state = self._state(
|
||||
acks={"prgs-author-0": "ack", "prgs-author-1": "acknowledged"}
|
||||
)
|
||||
proof = self._build(report, state)
|
||||
self.assertTrue(self._acks_check(proof).passed)
|
||||
self.assertTrue(proof.clean)
|
||||
self.assertEqual(proof.failed_checks, [])
|
||||
|
||||
def test_no_other_live_sessions_still_passes(self):
|
||||
"""Intended behavior retained: zero live sessions needs no acks."""
|
||||
report = _drained_report_with_live_sessions(0)
|
||||
self.assertEqual(report["counts"]["sessions_live_other"], 0)
|
||||
proof = self._build(report, self._state())
|
||||
check = self._acks_check(proof)
|
||||
self.assertTrue(check.passed)
|
||||
self.assertIn("sessions_live_other=0", check.detail)
|
||||
self.assertTrue(proof.clean)
|
||||
|
||||
# --- timeout policy cannot become a second fail-open ------------------
|
||||
|
||||
def test_unproven_timeout_policy_cannot_open_the_gate(self):
|
||||
for value in (None, "true", "yes", 1, "True", [], {}):
|
||||
with self.subTest(value=value):
|
||||
self.assertAcksFailClosed(
|
||||
_drained_report_with_live_sessions(2),
|
||||
self._state(ack_timeout_policy_applied=value),
|
||||
)
|
||||
|
||||
def test_explicit_timeout_policy_permits(self):
|
||||
proof = self._build(
|
||||
_drained_report_with_live_sessions(2),
|
||||
self._state(ack_timeout_policy_applied=True),
|
||||
)
|
||||
check = self._acks_check(proof)
|
||||
self.assertTrue(check.passed)
|
||||
self.assertIn("timeout policy", check.detail)
|
||||
self.assertTrue(proof.clean)
|
||||
|
||||
# --- the gate itself must deny ---------------------------------------
|
||||
|
||||
def test_failed_ack_check_denies_the_restart_gate(self):
|
||||
report = _drained_report_with_live_sessions(3)
|
||||
proof = self._build(report, self._state())
|
||||
self.assertFalse(proof.clean)
|
||||
decision = dp.gate_apply_restart(
|
||||
proof=proof.as_dict(),
|
||||
now=NOW,
|
||||
secret=SECRET,
|
||||
expected_impact_fingerprint=dp.impact_fingerprint(report),
|
||||
)
|
||||
self.assertFalse(decision.allow)
|
||||
self.assertEqual(decision.verdict, dp.GATE_DENY)
|
||||
self.assertIsNotNone(decision.incident)
|
||||
|
||||
def test_unclean_ack_proof_fails_verification(self):
|
||||
report = _drained_report_with_live_sessions(3)
|
||||
proof = self._build(report, self._state())
|
||||
result = dp.verify_drain_proof(
|
||||
proof.as_dict(),
|
||||
now=NOW,
|
||||
secret=SECRET,
|
||||
expected_impact_fingerprint=dp.impact_fingerprint(report),
|
||||
)
|
||||
self.assertFalse(result.valid)
|
||||
self.assertFalse(result.clean)
|
||||
|
||||
|
||||
def _identity_report(*, requester: str, others: tuple[str, ...]) -> dict:
|
||||
"""Report with explicitly named requester and other live sessions.
|
||||
|
||||
Unlike :func:`_drained_report_with_live_sessions`, the session ids are
|
||||
chosen by the caller so a test can supply acknowledgements for the *wrong*
|
||||
identities while keeping the count correct.
|
||||
"""
|
||||
|
||||
sessions = [
|
||||
{
|
||||
"session_id": requester,
|
||||
"role": "controller",
|
||||
"profile": "prgs-controller",
|
||||
"pid": _live_pid(),
|
||||
"status": "active",
|
||||
"last_heartbeat_at": NOW.isoformat(),
|
||||
}
|
||||
]
|
||||
for session_id in others:
|
||||
sessions.append(
|
||||
{
|
||||
"session_id": session_id,
|
||||
"role": "author",
|
||||
"profile": "prgs-author",
|
||||
"pid": _live_pid(),
|
||||
"status": "active",
|
||||
"last_heartbeat_at": NOW.isoformat(),
|
||||
}
|
||||
)
|
||||
report = rc.evaluate_restart_impact(
|
||||
{"sessions": sessions, "leases": [], "inventory_complete": True},
|
||||
now=NOW,
|
||||
requesting_session_id=requester,
|
||||
)
|
||||
return report.as_dict()
|
||||
|
||||
|
||||
class AcknowledgementIdentityBindingTests(unittest.TestCase):
|
||||
"""Acknowledgement coverage must be bound to session identity, not counted.
|
||||
|
||||
Regression cover for the second reviewed fail-open on PR #882 (review 582,
|
||||
blocker B1): coverage compared ``acked_count`` against
|
||||
``counts.sessions_live_other``, so acknowledgements supplied for the
|
||||
requesting session and for ids that do not exist satisfied the obligations
|
||||
of the live sessions that never answered. The required identities are
|
||||
carried by the report itself — ``ack_state`` keys and ``affected_sessions``
|
||||
filtered on ``live and not is_requester`` — and only an acknowledgement
|
||||
keyed by one of those ids may count for it.
|
||||
"""
|
||||
|
||||
def _state(self, **overrides) -> dict:
|
||||
state = _clean_drain_state()
|
||||
state.pop("acks", None)
|
||||
state["ack_timeout_policy_applied"] = False
|
||||
state.update(overrides)
|
||||
return state
|
||||
|
||||
def _acks_check(self, proof) -> dp.DrainCheck:
|
||||
return next(c for c in proof.checks if c.name == dp.CHECK_ACKS_OR_TIMEOUT)
|
||||
|
||||
def _build(self, report: dict, state: dict):
|
||||
return dp.build_drain_proof(
|
||||
impact_report=report, drain_state=state, now=NOW, secret=SECRET
|
||||
)
|
||||
|
||||
def assertAcksFailClosed(self, report: dict, state: dict) -> dp.DrainCheck:
|
||||
"""Failure must propagate through the check, the proof, and the gate."""
|
||||
|
||||
proof = self._build(report, state)
|
||||
check = self._acks_check(proof)
|
||||
self.assertFalse(check.passed)
|
||||
self.assertFalse(proof.clean)
|
||||
self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks)
|
||||
decision = dp.gate_apply_restart(
|
||||
proof=proof.as_dict(),
|
||||
now=NOW,
|
||||
secret=SECRET,
|
||||
expected_impact_fingerprint=dp.impact_fingerprint(report),
|
||||
)
|
||||
self.assertEqual(decision.verdict, dp.GATE_DENY)
|
||||
self.assertFalse(decision.allow)
|
||||
return check
|
||||
|
||||
# --- the reviewer's exact reproduction --------------------------------
|
||||
|
||||
def test_requester_plus_unknown_id_cannot_satisfy_two_live_sessions(self):
|
||||
"""Review 582 B1 verbatim: requester + a nonexistent session.
|
||||
|
||||
``sessions_live_other=2`` with ``ack_state`` naming ``other-0`` and
|
||||
``other-1``; the drain state supplies an acknowledgement from the
|
||||
requesting session itself and from a session that does not exist. The
|
||||
count matches, the identities do not.
|
||||
"""
|
||||
|
||||
report = _identity_report(requester="req", others=("other-0", "other-1"))
|
||||
self.assertEqual(report["counts"]["sessions_live_other"], 2)
|
||||
self.assertEqual(
|
||||
report["ack_state"], {"other-0": "pending", "other-1": "pending"}
|
||||
)
|
||||
state = self._state(acks={"req": "ack", "totally-bogus-session": "ack"})
|
||||
check = self.assertAcksFailClosed(report, state)
|
||||
self.assertIn("other-0", check.detail)
|
||||
self.assertIn("other-1", check.detail)
|
||||
self.assertIn("fail closed", check.detail)
|
||||
|
||||
# --- wrong / unknown / requester identities ---------------------------
|
||||
|
||||
def test_sufficient_count_of_wrong_ids_fails_closed(self):
|
||||
"""Right cardinality, wrong identities: two acks, neither required."""
|
||||
|
||||
report = _identity_report(requester="req", others=("other-0", "other-1"))
|
||||
state = self._state(acks={"ghost-a": "ack", "ghost-b": "ack"})
|
||||
check = self.assertAcksFailClosed(report, state)
|
||||
self.assertIn("do not count", check.detail)
|
||||
|
||||
def test_more_acks_than_required_still_fails_on_wrong_ids(self):
|
||||
"""Coverage cannot be bought with volume: five acks, none required."""
|
||||
|
||||
report = _identity_report(requester="req", others=("other-0", "other-1"))
|
||||
state = self._state(acks={f"ghost-{i}": "acknowledged" for i in range(5)})
|
||||
self.assertAcksFailClosed(report, state)
|
||||
|
||||
def test_partial_identity_match_fails_closed(self):
|
||||
"""One required id acknowledged, the rest padded with unknown ids."""
|
||||
|
||||
report = _identity_report(
|
||||
requester="req", others=("other-0", "other-1", "other-2")
|
||||
)
|
||||
state = self._state(
|
||||
acks={"other-0": "ack", "ghost-1": "ack", "ghost-2": "ack"}
|
||||
)
|
||||
check = self.assertAcksFailClosed(report, state)
|
||||
self.assertIn("other-1", check.detail)
|
||||
self.assertIn("other-2", check.detail)
|
||||
|
||||
def test_requester_ack_never_satisfies_another_sessions_obligation(self):
|
||||
"""The requester is excluded from the required set and stays excluded."""
|
||||
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
requester_rows = [s for s in report["affected_sessions"] if s["is_requester"]]
|
||||
self.assertEqual([s["session_id"] for s in requester_rows], ["req"])
|
||||
self.assertNotIn("req", report["ack_state"])
|
||||
check = self.assertAcksFailClosed(report, self._state(acks={"req": "ack"}))
|
||||
self.assertIn("other-0", check.detail)
|
||||
|
||||
def test_fabricated_ids_do_not_count_toward_coverage(self):
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
for bogus in ("", " ", "other-0 extra", "OTHER-0", "other-01", "0"):
|
||||
with self.subTest(bogus=bogus):
|
||||
self.assertAcksFailClosed(report, self._state(acks={bogus: "ack"}))
|
||||
|
||||
# --- per-session state must be explicitly valid ------------------------
|
||||
|
||||
def test_unproven_per_session_states_fail_closed(self):
|
||||
"""A required id present but not explicitly acknowledged fails closed."""
|
||||
|
||||
report = _identity_report(requester="req", others=("other-0", "other-1"))
|
||||
for value in ("pending", "stale", "unknown", "", None, True, 1, NOW):
|
||||
with self.subTest(value=value):
|
||||
self.assertAcksFailClosed(
|
||||
report,
|
||||
self._state(acks={"other-0": "ack", "other-1": value}),
|
||||
)
|
||||
|
||||
def test_report_ack_state_placeholder_is_never_read_as_an_ack(self):
|
||||
"""``ack_state`` values are the report's own placeholders, not evidence."""
|
||||
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
report["ack_state"] = {"other-0": "ack"}
|
||||
self.assertAcksFailClosed(report, self._state())
|
||||
|
||||
# --- missing / malformed / contradictory identity evidence -------------
|
||||
|
||||
def test_missing_identity_evidence_fails_closed(self):
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
report.pop("ack_state", None)
|
||||
report.pop("affected_sessions", None)
|
||||
check = self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
|
||||
self.assertIn("no session-identity evidence", check.detail)
|
||||
|
||||
def test_malformed_ack_state_fails_closed(self):
|
||||
for malformed in ([], "other-0", 7, None, ("other-0",)):
|
||||
with self.subTest(malformed=malformed):
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
report["ack_state"] = malformed
|
||||
self.assertAcksFailClosed(
|
||||
report, self._state(acks={"other-0": "ack"})
|
||||
)
|
||||
|
||||
def test_non_string_ack_state_key_fails_closed(self):
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
report["ack_state"] = {7: "pending"}
|
||||
self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
|
||||
|
||||
def test_malformed_affected_sessions_fails_closed(self):
|
||||
for malformed in ("sessions", 7, {"session_id": "other-0"}, [None], [7]):
|
||||
with self.subTest(malformed=malformed):
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
report.pop("ack_state", None)
|
||||
report["affected_sessions"] = malformed
|
||||
self.assertAcksFailClosed(
|
||||
report, self._state(acks={"other-0": "ack"})
|
||||
)
|
||||
|
||||
def test_affected_sessions_without_explicit_booleans_fails_closed(self):
|
||||
"""``live``/``is_requester`` must be real booleans, never inferred."""
|
||||
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
report.pop("ack_state", None)
|
||||
for row in report["affected_sessions"]:
|
||||
if row["session_id"] == "other-0":
|
||||
row["is_requester"] = "false"
|
||||
self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
|
||||
|
||||
def test_affected_sessions_missing_live_flag_fails_closed(self):
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
report.pop("ack_state", None)
|
||||
for row in report["affected_sessions"]:
|
||||
row.pop("live", None)
|
||||
self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
|
||||
|
||||
def test_contradictory_ack_state_and_affected_sessions_fails_closed(self):
|
||||
"""Both views present and disagreeing is unresolvable, not a tie-break."""
|
||||
|
||||
report = _identity_report(requester="req", others=("other-0", "other-1"))
|
||||
report["ack_state"] = {"other-0": "pending", "other-9": "pending"}
|
||||
check = self.assertAcksFailClosed(
|
||||
report, self._state(acks={"other-0": "ack", "other-9": "ack"})
|
||||
)
|
||||
self.assertIn("contradicts itself", check.detail)
|
||||
|
||||
def test_identity_count_mismatch_fails_closed(self):
|
||||
"""Identity evidence that cannot be reconciled with the count denies."""
|
||||
|
||||
report = _identity_report(requester="req", others=("other-0", "other-1"))
|
||||
report["counts"] = dict(report["counts"], sessions_live_other=1)
|
||||
check = self.assertAcksFailClosed(
|
||||
report, self._state(acks={"other-0": "ack", "other-1": "ack"})
|
||||
)
|
||||
self.assertIn("cannot be reconciled", check.detail)
|
||||
|
||||
def test_broken_identity_evidence_outranks_timeout_policy(self):
|
||||
"""The sanctioned timeout path cannot paper over an unreadable report."""
|
||||
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
report["ack_state"] = "not-a-mapping"
|
||||
self.assertAcksFailClosed(report, self._state(ack_timeout_policy_applied=True))
|
||||
|
||||
# --- legitimate success is preserved -----------------------------------
|
||||
|
||||
def test_every_required_session_acknowledged_passes(self):
|
||||
report = _identity_report(
|
||||
requester="req", others=("other-0", "other-1", "other-2")
|
||||
)
|
||||
state = self._state(
|
||||
acks={
|
||||
"other-0": "ack",
|
||||
"other-1": "acked",
|
||||
"other-2": "acknowledged",
|
||||
}
|
||||
)
|
||||
proof = self._build(report, state)
|
||||
check = self._acks_check(proof)
|
||||
self.assertTrue(check.passed)
|
||||
self.assertTrue(proof.clean)
|
||||
self.assertEqual(proof.failed_checks, [])
|
||||
self.assertIn("acknowledged by identity", check.detail)
|
||||
decision = dp.gate_apply_restart(
|
||||
proof=proof.as_dict(),
|
||||
now=NOW,
|
||||
secret=SECRET,
|
||||
expected_impact_fingerprint=dp.impact_fingerprint(report),
|
||||
)
|
||||
self.assertEqual(decision.verdict, dp.GATE_ALLOW)
|
||||
self.assertTrue(decision.allow)
|
||||
|
||||
def test_required_session_ack_tolerates_surrounding_whitespace(self):
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
proof = self._build(report, self._state(acks={" other-0 ": " ACK "}))
|
||||
self.assertTrue(self._acks_check(proof).passed)
|
||||
self.assertTrue(proof.clean)
|
||||
|
||||
def test_no_other_live_sessions_still_passes_with_identity_evidence(self):
|
||||
report = _identity_report(requester="req", others=())
|
||||
self.assertEqual(report["counts"]["sessions_live_other"], 0)
|
||||
self.assertEqual(report["ack_state"], {})
|
||||
proof = self._build(report, self._state())
|
||||
check = self._acks_check(proof)
|
||||
self.assertTrue(check.passed)
|
||||
self.assertIn("sessions_live_other=0", check.detail)
|
||||
self.assertTrue(proof.clean)
|
||||
|
||||
def test_explicit_timeout_policy_retains_intended_behavior(self):
|
||||
"""Valid, correctly typed timeout evidence still permits the check."""
|
||||
|
||||
report = _identity_report(requester="req", others=("other-0", "other-1"))
|
||||
proof = self._build(report, self._state(ack_timeout_policy_applied=True))
|
||||
check = self._acks_check(proof)
|
||||
self.assertTrue(check.passed)
|
||||
self.assertIn("timeout policy", check.detail)
|
||||
self.assertTrue(proof.clean)
|
||||
|
||||
def test_timeout_policy_still_strictly_typed_under_identity_binding(self):
|
||||
report = _identity_report(requester="req", others=("other-0",))
|
||||
for value in (None, "true", "True", 1, [], {}):
|
||||
with self.subTest(value=value):
|
||||
self.assertAcksFailClosed(
|
||||
report, self._state(ack_timeout_policy_applied=value)
|
||||
)
|
||||
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user