From ddc9b97d401a3102f4088abc27cbe1be1795f3a0 Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Tue, 21 Jul 2026 02:33:51 -0500 Subject: [PATCH] feat: enforce self-propagating canonical handoffs through controller closure (Closes #626) Adds the canonical cross-role handoff schema and its fail-closed validator, live-state recovery, role-limited continuation, mandatory durable posting, the merged-awaiting-controller boundary, controller accept/reject continuation, and workflow-failure escalation with duplicate handling. Co-Authored-By: Claude Opus 4.8 (1M context) --- final_report_validator.py | 33 + self_propagating_handoff.py | 907 ++++++++++++++++++ skills/llm-project-workflow/SKILL.md | 11 + .../schemas/create-issue-final-report.md | 4 + .../schemas/pr-queue-cleanup-final-report.md | 4 + .../schemas/reconcile-landed-final-report.md | 6 +- .../schemas/review-merge-final-report.md | 7 +- .../schemas/self-propagating-handoff.md | 114 +++ .../schemas/work-issue-final-report.md | 7 +- tests/test_self_propagating_handoff.py | 613 ++++++++++++ 10 files changed, 1703 insertions(+), 3 deletions(-) create mode 100644 self_propagating_handoff.py create mode 100644 skills/llm-project-workflow/schemas/self-propagating-handoff.md create mode 100644 tests/test_self_propagating_handoff.py diff --git a/final_report_validator.py b/final_report_validator.py index 50ee2e0..904ccf0 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -19,6 +19,10 @@ import reviewer_handoff_consistency import thread_state_ledger_validator from mcp_native_cleanup_proof import assess_mcp_native_cleanup_proof from post_merge_cleanup_proof import assess_post_merge_cleanup_proof +from self_propagating_handoff import ( + HANDOFF_HEADING as SELF_PROPAGATING_HANDOFF_HEADING, + assess_final_report_self_propagating_handoff, +) from review_proofs import ( HANDOFF_HEADING, assess_controller_handoff, @@ -1623,6 +1627,21 @@ def _rule_shared_mcp_native_cleanup_proof(report_text: str) -> list[dict[str, st ) +def _rule_shared_self_propagating_handoff(report_text: str) -> list[dict[str, str]]: + """#626: a report that adopts the handoff protocol must complete it.""" + result = assess_final_report_self_propagating_handoff(report_text) + if not result.get("applicable") or not result.get("block"): + return [] + return _findings_from_reasons( + "shared.self_propagating_handoff", + result.get("reasons") or ["incomplete canonical handoff"], + field=SELF_PROPAGATING_HANDOFF_HEADING, + severity="block", + safe_next_action=result.get("safe_next_action") + or "complete every canonical handoff field before posting", + ) + + _SHARED_ISSUE_LOCK_RULES = ( _rule_shared_issue_lock_external_state, _rule_shared_manual_lock_pr_override, @@ -1647,8 +1666,14 @@ _SHARED_MUTATION_BUDGET_RULES = ( _rule_shared_mutation_budget_accounting, ) +# #626: enforced for every task kind that can continue the workflow chain. +_SHARED_SELF_PROPAGATING_HANDOFF_RULES = ( + _rule_shared_self_propagating_handoff, +) + _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { "review_pr": [ + *_SHARED_SELF_PROPAGATING_HANDOFF_RULES, _rule_shared_controller_handoff, _rule_shared_state_handoff_next_action, _rule_shared_email_disclosure, @@ -1684,6 +1709,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reviewer_stale_head_proof, ], "merge_pr": [ + *_SHARED_SELF_PROPAGATING_HANDOFF_RULES, _rule_shared_controller_handoff, _rule_shared_email_disclosure, *_SHARED_ISSUE_LOCK_RULES, @@ -1695,6 +1721,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reviewer_stale_head_proof, ], "reconcile_already_landed": [ + *_SHARED_SELF_PROPAGATING_HANDOFF_RULES, _rule_reconcile_controller_handoff, _rule_shared_state_handoff_next_action, _rule_shared_email_disclosure, @@ -1715,6 +1742,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_audit_reconciliation_boundary, ], "author_issue": [ + *_SHARED_SELF_PROPAGATING_HANDOFF_RULES, _rule_shared_controller_handoff, _rule_shared_state_handoff_next_action, _rule_shared_email_disclosure, @@ -1725,6 +1753,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reviewer_vague_mutations_none, ], "work_issue": [ + *_SHARED_SELF_PROPAGATING_HANDOFF_RULES, _rule_shared_controller_handoff, _rule_shared_state_handoff_next_action, _rule_shared_email_disclosure, @@ -1739,6 +1768,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_worktree_cleanup_audit_proof, ], "issue_filing": [ + *_SHARED_SELF_PROPAGATING_HANDOFF_RULES, _rule_shared_controller_handoff, _rule_shared_state_handoff_next_action, _rule_shared_email_disclosure, @@ -1748,6 +1778,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { *_SHARED_ISSUE_LOCK_RULES, ], "inventory": [ + *_SHARED_SELF_PROPAGATING_HANDOFF_RULES, _rule_shared_controller_handoff, _rule_shared_state_handoff_next_action, _rule_shared_email_disclosure, @@ -1758,6 +1789,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reconcile_pagination_proof, ], "issue_selection": [ + *_SHARED_SELF_PROPAGATING_HANDOFF_RULES, _rule_shared_controller_handoff, _rule_shared_state_handoff_next_action, _rule_shared_email_disclosure, @@ -1771,6 +1803,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { # Kept intentionally narrow so a closure pre-check does not demand the # full reviewer/author handoff schema. "controller_close": [ + *_SHARED_SELF_PROPAGATING_HANDOFF_RULES, _rule_reviewer_premerge_baseline_proof, ], } diff --git a/self_propagating_handoff.py b/self_propagating_handoff.py new file mode 100644 index 0000000..2a0830b --- /dev/null +++ b/self_propagating_handoff.py @@ -0,0 +1,907 @@ +"""Self-propagating canonical handoffs through final controller closure (#626). + +#494-#507 defined the canonical ledger, next-action comments, comment +validation, the controller acceptance gate, and the Canonical Thread Handoff +(CTH) shape. What none of them enforce is the *chain*: that every actor +consumes exactly one canonical handoff, performs exactly one authorized role, +records the result durably in Gitea, and emits the next complete handoff until +the controller records final closure. + +This module owns that systemic gap: + +* one canonical cross-role handoff schema (:data:`HANDOFF_FIELDS`); +* a fail-closed validator that rejects incomplete handoffs; +* live-state recovery so a receiving actor never trusts an inherited handoff; +* role-limited continuation; +* mandatory durable posting into Gitea; +* the ``merged-awaiting-controller`` boundary and controller accept/reject + continuation; +* workflow-failure escalation into separate durable issues, with duplicate + handling; +* terminal closure that must *not* emit an unnecessary next prompt. + +Everything here is pure assessment: no network calls, no mutation. +""" + +from __future__ import annotations + +import re +from typing import Any, Iterable, Mapping, Sequence + +MARKER = "" +HANDOFF_HEADING = "Canonical Handoff" + +#: Canonical workflow states a handoff may declare. +WORKFLOW_STATES: tuple[str, ...] = ( + "needs-author", + "needs-review", + "approved-awaiting-merge", + "merged-awaiting-controller", + "blocked", + "complete", +) + +TERMINAL_STATES = frozenset({"complete"}) + +#: The single role authorized to act on each workflow state. +NEXT_ACTOR_BY_STATE: dict[str, str] = { + "needs-author": "author", + "needs-review": "reviewer", + "approved-awaiting-merge": "merger", + "merged-awaiting-controller": "controller", + "blocked": "operator", + "complete": "none", +} + +WORKFLOW_ROLES = frozenset( + {"author", "reviewer", "merger", "controller", "operator", "reconciler"} +) + +#: What each receiving role is authorized to do when it consumes a handoff. +ROLE_ALLOWED_ACTIONS: dict[str, tuple[str, ...]] = { + "author": ("implement", "commit", "push", "create_pr", "comment"), + "reviewer": ("review", "approve", "request_changes", "comment"), + "merger": ("verify_approval_parity", "merge", "comment"), + "controller": ("accept", "reject", "reopen", "close_issue", "comment"), + "operator": ("repair_infrastructure", "comment"), + "reconciler": ("close_superseded_pr", "cleanup_branch", "comment"), +} + +ROLE_FORBIDDEN_ACTIONS: dict[str, tuple[str, ...]] = { + "author": ("approve", "request_changes", "merge", "close_issue"), + "reviewer": ("merge", "commit", "push", "create_pr"), + "merger": ("approve", "commit", "push", "create_pr"), + "controller": ("approve", "merge", "commit", "push"), + "operator": ("approve", "merge", "close_issue"), + "reconciler": ("approve", "merge", "commit", "push", "create_pr"), +} + +#: Ordered canonical handoff fields. Every one of them is required; the +#: fields in :data:`NONE_ALLOWED_FIELDS` may legitimately carry ``none``. +HANDOFF_FIELDS: tuple[str, ...] = ( + "REPOSITORY", + "ISSUE", + "PR", + "WORKFLOW_STATE", + "HEAD_SHA", + "BASE_BRANCH", + "BASE_OR_MERGE_SHA", + "ACTING_ROLE", + "ACTING_IDENTITY", + "COMPLETED_ACTIONS", + "VALIDATION_EVIDENCE", + "MUTATION_LEDGER", + "BLOCKERS", + "NEXT_ACTOR", + "NEXT_ACTION", + "PROHIBITED_ACTIONS", + "NEXT_PROMPT", + "WORKFLOW_FAILURE_ISSUES", + "LAST_UPDATED", +) + +NONE_ALLOWED_FIELDS = frozenset( + { + "PR", + "HEAD_SHA", + "BASE_OR_MERGE_SHA", + "BLOCKERS", + "WORKFLOW_FAILURE_ISSUES", + "NEXT_PROMPT", + "NEXT_ACTION", + } +) + +#: States where no PR or head SHA exists yet, so ``none`` is legitimate. +_PRE_PR_STATES = frozenset({"needs-author", "blocked"}) + +_PLACEHOLDERS = frozenset({"", "none", "n/a", "na", "tbd", "todo", "unknown", "?"}) + +#: A next prompt short enough to be a stub cannot be "ready to run". +MIN_NEXT_PROMPT_CHARS = 40 + +_FIELD_LINE_RE = re.compile(r"^([A-Z][A-Z0-9_]*)\s*:\s*(.*)$", re.MULTILINE) +_HEADING_RE = re.compile(r"^##\s*Canonical Handoff\s*$", re.IGNORECASE | re.MULTILINE) +_EXTERNAL_CHAT_RE = re.compile( + r"\b(?:previous chat|prior conversation|earlier conversation|see (?:the )?chat|" + r"chat history|paste (?:this )?(?:from|into) chatgpt|ask the operator to paste)\b", + re.IGNORECASE, +) + +LIVE_DETECTION_KINDS: tuple[str, ...] = ( + "changed_pr_head", + "stale_approval", + "issue_closed", + "issue_reopened", + "pr_merged", + "pr_closed_unmerged", + "stale_lease", + "foreign_lease", + "missing_worktree", + "dirty_worktree", + "namespace_mismatch", + "stale_runtime", + "changed_base", + "conflicting_canonical_comments", +) + +CONTROLLER_DECISIONS = frozenset( + { + "accept", + "request_tests", + "request_proof", + "request_corrections", + "reopen", + "return_to_actor", + } +) + +CONTROLLER_CLOSURE_PROOF_FIELDS = ( + "acceptance_criteria_satisfied", + "cleanup_complete", + "canonical_final_state_posted", + "issue_closed_through_workflow", +) + +WORKFLOW_FAILURE_FIELDS = ( + "classification", + "linked_issue", + "temporary_impact", + "next_valid_actor", + "recovery_prompt", +) + + +def _is_placeholder(value: Any) -> bool: + return str(value or "").strip().lower() in _PLACEHOLDERS + + +def _clean(value: Any) -> str: + text = str(value).strip() if value is not None else "" + return text or "none" + + +# --------------------------------------------------------------------------- +# rendering / parsing +# --------------------------------------------------------------------------- + + +def render_self_propagating_handoff(**values: Any) -> str: + """Render a canonical cross-role handoff block. + + Raises ``ValueError`` for an unknown workflow state so a malformed handoff + can never be produced by the sanctioned renderer. + """ + state = str(values.get("WORKFLOW_STATE", values.get("workflow_state", ""))).strip() + if state not in WORKFLOW_STATES: + raise ValueError( + f"unknown workflow state '{state}'; expected one of {list(WORKFLOW_STATES)}" + ) + lines = [MARKER, f"## {HANDOFF_HEADING}", "", "```text"] + for name in HANDOFF_FIELDS: + raw = values.get(name, values.get(name.lower())) + lines.append(f"{name}: {_clean(raw)}") + lines.append("```") + return "\n".join(lines) + + +def parse_self_propagating_handoff(text: str) -> dict[str, str] | None: + """Parse a canonical handoff block, or ``None`` when absent.""" + body = text or "" + if MARKER not in body and not _HEADING_RE.search(body): + return None + fields = { + match.group(1): match.group(2).strip() + for match in _FIELD_LINE_RE.finditer(body) + } + if not fields: + return None + return fields + + +def handoff_present(text: str) -> bool: + """Whether *text* carries a canonical handoff block at all.""" + return parse_self_propagating_handoff(text) is not None + + +# --------------------------------------------------------------------------- +# handoff validation (AC: a validator rejects incomplete handoffs) +# --------------------------------------------------------------------------- + + +def assess_self_propagating_handoff(text: str) -> dict[str, Any]: + """Fail closed unless *text* carries one complete canonical handoff.""" + fields = parse_self_propagating_handoff(text) + if fields is None: + return { + "valid": False, + "block": True, + "present": False, + "fields": {}, + "missing_fields": list(HANDOFF_FIELDS), + "workflow_state": None, + "next_actor": None, + "terminal": False, + "reasons": ["report or comment carries no canonical handoff block"], + "safe_next_action": ( + "add a canonical handoff block with all " + f"{len(HANDOFF_FIELDS)} fields before posting" + ), + } + + reasons: list[str] = [] + state = (fields.get("WORKFLOW_STATE") or "").strip() + terminal = state in TERMINAL_STATES + + if state not in WORKFLOW_STATES: + reasons.append( + f"unknown WORKFLOW_STATE '{state or 'missing'}'; " + f"expected one of {list(WORKFLOW_STATES)}" + ) + + missing = [name for name in HANDOFF_FIELDS if name not in fields] + reasons.extend(f"handoff missing field: {name}" for name in missing) + + for name in HANDOFF_FIELDS: + if name in missing: + continue + value = fields.get(name, "") + if not _is_placeholder(value): + continue + if name in NONE_ALLOWED_FIELDS: + continue + # A terminated chain names no next actor by design. + if name == "NEXT_ACTOR" and terminal: + continue + reasons.append(f"handoff field {name} must be concrete, got '{value or ''}'") + + if state and state not in _PRE_PR_STATES and state in WORKFLOW_STATES: + for name in ("PR", "HEAD_SHA"): + if name not in missing and _is_placeholder(fields.get(name)): + reasons.append( + f"handoff field {name} must be concrete in state '{state}'" + ) + + if state == "blocked" and _is_placeholder(fields.get("BLOCKERS")): + reasons.append("state 'blocked' requires a concrete BLOCKERS entry") + + declared_actor = (fields.get("NEXT_ACTOR") or "").strip().lower() + expected_actor = NEXT_ACTOR_BY_STATE.get(state) + if expected_actor and declared_actor != expected_actor: + reasons.append( + f"NEXT_ACTOR '{declared_actor or 'missing'}' does not match state " + f"'{state}', which authorizes '{expected_actor}'" + ) + + next_prompt = (fields.get("NEXT_PROMPT") or "").strip() + next_action = (fields.get("NEXT_ACTION") or "").strip() + if terminal: + # A completed workflow terminates; it must not manufacture more work. + if not _is_placeholder(next_prompt): + reasons.append( + "terminal state 'complete' must not carry a NEXT_PROMPT; " + "the chain ends at controller closure" + ) + if not _is_placeholder(next_action): + reasons.append( + "terminal state 'complete' must not carry a NEXT_ACTION" + ) + else: + if _is_placeholder(next_prompt): + reasons.append( + "non-terminal handoff requires a complete ready-to-run NEXT_PROMPT" + ) + elif len(next_prompt) < MIN_NEXT_PROMPT_CHARS: + reasons.append( + "NEXT_PROMPT is too short to be ready-to-run " + f"({len(next_prompt)} < {MIN_NEXT_PROMPT_CHARS} characters)" + ) + if _is_placeholder(next_action): + reasons.append("non-terminal handoff requires a concrete NEXT_ACTION") + + acting_role = (fields.get("ACTING_ROLE") or "").strip().lower() + if acting_role and acting_role not in WORKFLOW_ROLES: + reasons.append( + f"unknown ACTING_ROLE '{acting_role}'; expected one of " + f"{sorted(WORKFLOW_ROLES)}" + ) + + block = bool(reasons) + return { + "valid": not block, + "block": block, + "present": True, + "fields": fields, + "missing_fields": missing, + "workflow_state": state or None, + "next_actor": declared_actor or None, + "terminal": terminal, + "reasons": reasons, + "safe_next_action": ( + "complete every canonical handoff field before posting" + if block + else "proceed" + ), + } + + +def assess_thread_recoverability(text: str) -> dict[str, Any]: + """The next actor must recover from the thread alone — never outside chat.""" + assessment = assess_self_propagating_handoff(text) + if assessment["block"]: + return { + "recoverable": False, + "block": True, + "reasons": assessment["reasons"], + "safe_next_action": assessment["safe_next_action"], + } + + fields = assessment["fields"] + reasons: list[str] = [] + if assessment["terminal"]: + return { + "recoverable": True, + "block": False, + "reasons": [], + "safe_next_action": "proceed", + } + + prompt = fields.get("NEXT_PROMPT", "") + repository = fields.get("REPOSITORY", "").strip() + issue = fields.get("ISSUE", "").strip().lstrip("#") + + if repository and repository.lower() not in prompt.lower(): + reasons.append("NEXT_PROMPT must name the repository it applies to") + if issue and issue not in prompt: + reasons.append(f"NEXT_PROMPT must name issue {issue}") + if _EXTERNAL_CHAT_RE.search(prompt): + reasons.append( + "NEXT_PROMPT must not depend on outside chat history; the issue or " + "PR thread, workflow docs, and live repository state must suffice" + ) + + block = bool(reasons) + return { + "recoverable": not block, + "block": block, + "reasons": reasons, + "safe_next_action": ( + "rewrite NEXT_PROMPT so it is self-contained" if block else "proceed" + ), + } + + +# --------------------------------------------------------------------------- +# live-state recovery (AC: head changes invalidate stale review/merge handoffs) +# --------------------------------------------------------------------------- + + +def _detection(kind: str, detail: str) -> dict[str, str]: + return {"kind": kind, "detail": detail} + + +def assess_handoff_live_state( + *, + handoff: str | Mapping[str, str], + live: Mapping[str, Any], +) -> dict[str, Any]: + """Re-derive workflow truth from live state instead of trusting *handoff*. + + *live* carries observed facts; absent keys are simply not checked, but any + fact that contradicts the inherited handoff fails closed. + """ + if isinstance(handoff, Mapping): + fields = dict(handoff) + else: + parsed = parse_self_propagating_handoff(handoff or "") + if parsed is None: + return { + "block": True, + "detections": [_detection("missing_handoff", "no canonical handoff")], + "kinds": ["missing_handoff"], + "reasons": ["no canonical handoff to reconcile against live state"], + "recovered_state": None, + "safe_next_action": "post a canonical handoff before continuing", + } + fields = parsed + + state = (fields.get("WORKFLOW_STATE") or "").strip() + next_actor = (fields.get("NEXT_ACTOR") or "").strip().lower() + detections: list[dict[str, str]] = [] + recovered_state: str | None = None + + handoff_head = (fields.get("HEAD_SHA") or "").strip() + live_head = str(live.get("pr_head_sha") or "").strip() + head_changed = bool( + live_head and handoff_head and not _is_placeholder(handoff_head) + and live_head != handoff_head + ) + if head_changed: + detections.append( + _detection( + "changed_pr_head", + f"handoff pinned {handoff_head}, live head is {live_head}", + ) + ) + if next_actor in {"reviewer", "merger"}: + recovered_state = "needs-review" + + approved_head = str(live.get("approved_head_sha") or "").strip() + if approved_head and live_head and approved_head != live_head: + detections.append( + _detection( + "stale_approval", + f"approval recorded at {approved_head}, live head is {live_head}", + ) + ) + if next_actor == "merger": + recovered_state = "needs-review" + + issue_state = str(live.get("issue_state") or "").strip().lower() + if issue_state == "closed" and state not in TERMINAL_STATES: + detections.append( + _detection("issue_closed", "linked issue is closed but handoff is not complete") + ) + if issue_state == "open" and state in TERMINAL_STATES: + detections.append( + _detection("issue_reopened", "handoff claims complete but the issue is open") + ) + recovered_state = "needs-author" + + pr_state = str(live.get("pr_state") or "").strip().lower() + if pr_state == "merged" and state in { + "needs-author", + "needs-review", + "approved-awaiting-merge", + }: + detections.append( + _detection("pr_merged", "PR is already merged; controller boundary applies") + ) + recovered_state = "merged-awaiting-controller" + if pr_state == "closed" and state not in TERMINAL_STATES: + detections.append( + _detection("pr_closed_unmerged", "PR is closed without merge") + ) + + lease = live.get("lease") or {} + if isinstance(lease, Mapping) and lease: + lease_status = str(lease.get("status") or "").strip().lower() + if lease_status and lease_status != "active": + detections.append( + _detection("stale_lease", f"lease status is '{lease_status}'") + ) + lease_session = str(lease.get("session_id") or "").strip() + actor_session = str(live.get("actor_session_id") or "").strip() + if lease_session and actor_session and lease_session != actor_session: + detections.append( + _detection( + "foreign_lease", + "lease is owned by another session; never adopt it implicitly", + ) + ) + + worktree = live.get("worktree") or {} + if isinstance(worktree, Mapping) and worktree: + if worktree.get("present") is False: + detections.append(_detection("missing_worktree", "bound worktree is absent")) + if worktree.get("dirty") is True: + detections.append( + _detection("dirty_worktree", "bound worktree carries uncommitted changes") + ) + + namespace_role = str(live.get("namespace_role") or "").strip().lower() + if namespace_role and next_actor and next_actor != "none": + if namespace_role != next_actor: + detections.append( + _detection( + "namespace_mismatch", + f"live namespace role '{namespace_role}' cannot act as '{next_actor}'", + ) + ) + + if live.get("runtime_stale") is True: + detections.append( + _detection("stale_runtime", "serving runtime is stale; reconnect required") + ) + + handoff_base = (fields.get("BASE_BRANCH") or "").strip() + live_base = str(live.get("base_branch") or "").strip() + if handoff_base and live_base and not _is_placeholder(handoff_base): + if handoff_base != live_base: + detections.append( + _detection( + "changed_base", + f"handoff base '{handoff_base}' but live base '{live_base}'", + ) + ) + + if live.get("conflicting_canonical_comments") is True: + detections.append( + _detection( + "conflicting_canonical_comments", + "thread carries contradictory canonical comments", + ) + ) + + kinds = [item["kind"] for item in detections] + reasons = [f"{item['kind']}: {item['detail']}" for item in detections] + block = bool(detections) + return { + "block": block, + "detections": detections, + "kinds": kinds, + "reasons": reasons, + "recovered_state": recovered_state, + "safe_next_action": ( + "post a corrected canonical handoff for the recovered live state " + "before acting" + if block + else "proceed" + ), + } + + +# --------------------------------------------------------------------------- +# role-limited continuation +# --------------------------------------------------------------------------- + + +def assess_role_continuation( + *, + handoff: str | Mapping[str, str], + actor_role: str, +) -> dict[str, Any]: + """Only the role the current state authorizes may continue the chain.""" + if isinstance(handoff, Mapping): + fields = dict(handoff) + else: + fields = parse_self_propagating_handoff(handoff or "") or {} + + role = (actor_role or "").strip().lower() + state = (fields.get("WORKFLOW_STATE") or "").strip() + expected = NEXT_ACTOR_BY_STATE.get(state) + reasons: list[str] = [] + + if not fields: + reasons.append("no canonical handoff to continue from") + if role not in WORKFLOW_ROLES: + reasons.append(f"unknown actor role '{actor_role}'") + if expected is None and fields: + reasons.append(f"unknown workflow state '{state}'") + elif expected == "none": + reasons.append( + "workflow state 'complete' is terminal; no further role may continue" + ) + elif expected and role != expected: + reasons.append( + f"state '{state}' authorizes '{expected}', not '{role}'" + ) + + block = bool(reasons) + return { + "allowed": not block, + "block": block, + "expected_actor": expected, + "actor_role": role, + "allowed_actions": () if block else ROLE_ALLOWED_ACTIONS.get(role, ()), + "forbidden_actions": ROLE_FORBIDDEN_ACTIONS.get(role, ()), + "reasons": reasons, + "safe_next_action": ( + f"hand off to '{expected}'" if block and expected else + "stop; the workflow is complete" if expected == "none" else + "proceed" + ), + } + + +# --------------------------------------------------------------------------- +# durable posting (AC: a chat-only report is never sufficient) +# --------------------------------------------------------------------------- + + +def assess_durable_state_update( + *, + handoff_text: str, + posted_comment_id: Any = None, + canonical_state_posted: bool = False, +) -> dict[str, Any]: + """A successful actor session must leave the handoff in Gitea, not chat.""" + reasons: list[str] = [] + assessment = assess_self_propagating_handoff(handoff_text) + if assessment["block"]: + reasons.extend(assessment["reasons"]) + if not posted_comment_id: + reasons.append( + "canonical handoff was not posted to Gitea; a chat-only report is " + "not durable workflow state" + ) + if not canonical_state_posted: + reasons.append( + "canonical issue/PR state and thread ledger were not updated" + ) + + block = bool(reasons) + return { + "durable": not block, + "block": block, + "posted_comment_id": posted_comment_id, + "reasons": reasons, + "safe_next_action": ( + "post the canonical handoff and state update to Gitea before " + "ending the session" + if block + else "proceed" + ), + } + + +# --------------------------------------------------------------------------- +# merge -> controller boundary and controller continuation +# --------------------------------------------------------------------------- + + +def assess_merge_completion_transition( + *, + merge_succeeded: bool, + controller_auto_accept: bool = False, +) -> dict[str, Any]: + """A merged PR is not accepted work until the controller says so.""" + if not merge_succeeded: + return { + "next_state": "approved-awaiting-merge", + "next_actor": "merger", + "next_prompt_required": True, + "reasons": ["merge did not succeed; the merger retains the work item"], + } + if controller_auto_accept: + return { + "next_state": "complete", + "next_actor": "none", + "next_prompt_required": False, + "reasons": ["configured workflow authorizes automatic acceptance on merge"], + } + return { + "next_state": "merged-awaiting-controller", + "next_actor": "controller", + "next_prompt_required": True, + "reasons": [ + "merge succeeded; acceptance requires the authorized controller" + ], + } + + +def assess_controller_decision( + *, + decision: str, + closure_proof: Mapping[str, Any] | None = None, + return_to: str | None = None, +) -> dict[str, Any]: + """Controller acceptance or rejection produces the next or final state.""" + normalized = (decision or "").strip().lower() + if normalized not in CONTROLLER_DECISIONS: + return { + "block": True, + "next_state": None, + "next_actor": None, + "next_prompt_required": True, + "reasons": [ + f"unknown controller decision '{decision}'; expected one of " + f"{sorted(CONTROLLER_DECISIONS)}" + ], + "safe_next_action": "record a supported controller decision", + } + + if normalized == "accept": + proof = dict(closure_proof or {}) + missing = [ + name + for name in CONTROLLER_CLOSURE_PROOF_FIELDS + if proof.get(name) is not True + ] + if missing: + return { + "block": True, + "next_state": "merged-awaiting-controller", + "next_actor": "controller", + "next_prompt_required": True, + "reasons": [ + "controller acceptance missing closure proof: " + ", ".join(missing) + ], + "safe_next_action": ( + "satisfy and record every closure proof field before closing" + ), + } + return { + "block": False, + "next_state": "complete", + "next_actor": "none", + "next_prompt_required": False, + "reasons": ["controller accepted; workflow chain terminates"], + "safe_next_action": "post the final canonical state and stop", + } + + if normalized == "return_to_actor": + target = (return_to or "").strip().lower() + state_by_actor = { + "author": "needs-author", + "reviewer": "needs-review", + "merger": "approved-awaiting-merge", + } + if target not in state_by_actor: + return { + "block": True, + "next_state": None, + "next_actor": None, + "next_prompt_required": True, + "reasons": [ + f"return_to_actor requires a target in {sorted(state_by_actor)}" + ], + "safe_next_action": "name the actor the work returns to", + } + return { + "block": False, + "next_state": state_by_actor[target], + "next_actor": target, + "next_prompt_required": True, + "reasons": [f"controller returned the work item to '{target}'"], + "safe_next_action": f"post a complete handoff for '{target}'", + } + + # request_tests / request_proof / request_corrections / reopen + return { + "block": False, + "next_state": "needs-author", + "next_actor": "author", + "next_prompt_required": True, + "reasons": [f"controller decision '{normalized}' returns the work to the author"], + "safe_next_action": "post a complete author handoff describing what is required", + } + + +# --------------------------------------------------------------------------- +# workflow-failure escalation +# --------------------------------------------------------------------------- + + +def assess_workflow_failure_escalation( + *, + failures: Sequence[Mapping[str, Any]] | None, + active_issue_number: int | str | None, + existing_failure_issues: Iterable[Mapping[str, Any]] | None = None, +) -> dict[str, Any]: + """Tooling defects hit while working an issue become separate durable work.""" + entries = list(failures or []) + known = { + str((item.get("signature") or "")).strip().lower(): item.get("number") + for item in (existing_failure_issues or []) + if str((item.get("signature") or "")).strip() + } + active = str(active_issue_number or "").strip().lstrip("#") + + reasons: list[str] = [] + reused: list[dict[str, Any]] = [] + seen_signatures: dict[str, str] = {} + + for index, failure in enumerate(entries): + label = str(failure.get("signature") or f"failure[{index}]") + missing = [ + name + for name in WORKFLOW_FAILURE_FIELDS + if _is_placeholder(failure.get(name)) + ] + if missing: + reasons.append( + f"{label}: workflow failure missing " + ", ".join(missing) + ) + + linked = str(failure.get("linked_issue") or "").strip().lstrip("#") + if linked and active and linked == active: + reasons.append( + f"{label}: workflow defects must not be folded into the active " + f"work item #{active}; file a separate durable issue" + ) + + signature = str(failure.get("signature") or "").strip().lower() + if not signature: + continue + if signature in known: + expected = str(known[signature] or "").strip().lstrip("#") + if linked and expected and linked != expected: + reasons.append( + f"{label}: duplicate workflow-failure issue #{linked}; " + f"reuse the existing issue #{expected}" + ) + else: + reused.append({"signature": signature, "issue": expected}) + if signature in seen_signatures: + reasons.append( + f"{label}: duplicate workflow-failure signature reported twice " + "in one session" + ) + else: + seen_signatures[signature] = linked + + block = bool(reasons) + return { + "escalated": not block, + "block": block, + "failure_count": len(entries), + "reused_issues": reused, + "reasons": reasons, + "safe_next_action": ( + "file or reference one durable issue per distinct workflow failure" + if block + else "proceed" + ), + } + + +# --------------------------------------------------------------------------- +# final-report integration +# --------------------------------------------------------------------------- + + +def assess_final_report_self_propagating_handoff(report_text: str) -> dict[str, Any]: + """#626 gate for final reports. + + Applicability mirrors the #495 canonical-state gate: once a report adopts + the protocol — by carrying the marker, the ``Canonical Handoff`` heading, + or a ``WORKFLOW_STATE`` line — the full schema is enforced. Reports that + predate the protocol are untouched here; the workflow schemas require the + block going forward. + """ + text = report_text or "" + applicable = ( + MARKER in text + or bool(_HEADING_RE.search(text)) + or bool(re.search(r"^WORKFLOW_STATE\s*:", text, re.MULTILINE)) + ) + if not applicable: + return { + "applicable": False, + "valid": True, + "block": False, + "reasons": [], + "safe_next_action": "proceed", + } + + assessment = assess_self_propagating_handoff(text) + recoverability = assess_thread_recoverability(text) + reasons = list(assessment["reasons"]) + if not assessment["block"]: + reasons.extend(recoverability.get("reasons") or []) + block = bool(assessment["block"] or recoverability.get("block")) + return { + "applicable": True, + "valid": not block, + "block": block, + "workflow_state": assessment.get("workflow_state"), + "next_actor": assessment.get("next_actor"), + "terminal": assessment.get("terminal"), + "reasons": reasons, + "safe_next_action": ( + assessment["safe_next_action"] + if assessment["block"] + else recoverability.get("safe_next_action", "proceed") + ), + } diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index 5720069..9f14d7a 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -224,6 +224,17 @@ format canonical field set per issue #182; mode-specific schemas in for the loaded workflow mode — not the legacy compact block alone. `review_proofs.assess_controller_handoff()` validates presence. +## Canonical self-propagating handoff + +Every workflow mode also carries the cross-role handoff block defined in +[`schemas/self-propagating-handoff.md`](schemas/self-propagating-handoff.md) +(#626). Each actor consumes exactly one canonical handoff, performs exactly one +authorized role, posts the result to the Gitea issue or PR thread, and emits the +next complete handoff — until the controller records final closure. The block +must be posted to Gitea, not returned in chat alone, and the next prompt is not +an optional prose section. `self_propagating_handoff.py` implements the schema; +`final_report_validator.py` enforces it as `shared.self_propagating_handoff`. + ## Prompt templates Ready-to-copy task prompts live in [`templates/`](templates/): diff --git a/skills/llm-project-workflow/schemas/create-issue-final-report.md b/skills/llm-project-workflow/schemas/create-issue-final-report.md index f35c233..083cdcc 100644 --- a/skills/llm-project-workflow/schemas/create-issue-final-report.md +++ b/skills/llm-project-workflow/schemas/create-issue-final-report.md @@ -44,3 +44,7 @@ mutations occurred). ``` Identity format: `username / profile` (not personal email unless required — #305). + +The report must also carry the canonical self-propagating handoff block +(`schemas/self-propagating-handoff.md`, #626) and that block must be posted to +the Gitea issue thread. diff --git a/skills/llm-project-workflow/schemas/pr-queue-cleanup-final-report.md b/skills/llm-project-workflow/schemas/pr-queue-cleanup-final-report.md index c7abe8c..0ad8b80 100644 --- a/skills/llm-project-workflow/schemas/pr-queue-cleanup-final-report.md +++ b/skills/llm-project-workflow/schemas/pr-queue-cleanup-final-report.md @@ -29,3 +29,7 @@ use `none` where nothing occurred. Validated by * Read-only diagnostics: * Blockers: * Safe next action: (fresh run for the next PR) + +The report must also carry the canonical self-propagating handoff block +(`schemas/self-propagating-handoff.md`, #626) and that block must be posted to +the Gitea PR thread. diff --git a/skills/llm-project-workflow/schemas/reconcile-landed-final-report.md b/skills/llm-project-workflow/schemas/reconcile-landed-final-report.md index d7bf77b..daf4f2e 100644 --- a/skills/llm-project-workflow/schemas/reconcile-landed-final-report.md +++ b/skills/llm-project-workflow/schemas/reconcile-landed-final-report.md @@ -50,4 +50,8 @@ occurred). Identity format: `username / profile` (not personal email unless required — #305). -`git fetch` belongs under `Git ref mutations`, not read-only diagnostics (#297). \ No newline at end of file +`git fetch` belongs under `Git ref mutations`, not read-only diagnostics (#297). + +The report must also carry the canonical self-propagating handoff block +(`schemas/self-propagating-handoff.md`, #626) and that block must be posted to +the Gitea issue or PR thread. \ No newline at end of file diff --git a/skills/llm-project-workflow/schemas/review-merge-final-report.md b/skills/llm-project-workflow/schemas/review-merge-final-report.md index 6ef00bc..a688432 100644 --- a/skills/llm-project-workflow/schemas/review-merge-final-report.md +++ b/skills/llm-project-workflow/schemas/review-merge-final-report.md @@ -116,4 +116,9 @@ or structured MCP metadata — not narrative alone: When a claim relies on prior-session blocker state or MCP metadata only, label the proof source explicitly (`command`, `MCP metadata`, `prior blocker`, -`not checked`). Do not use `live proof` without that classification. \ No newline at end of file +`not checked`). Do not use `live proof` without that classification. + +The report must also carry the canonical self-propagating handoff block +(`schemas/self-propagating-handoff.md`, #626) and that block must be posted to +the Gitea PR thread. A reviewer hands off to `merger`; a merger transitions to +`merged-awaiting-controller` rather than declaring the work accepted. \ No newline at end of file diff --git a/skills/llm-project-workflow/schemas/self-propagating-handoff.md b/skills/llm-project-workflow/schemas/self-propagating-handoff.md new file mode 100644 index 0000000..f0252e2 --- /dev/null +++ b/skills/llm-project-workflow/schemas/self-propagating-handoff.md @@ -0,0 +1,114 @@ +# Canonical self-propagating handoff schema (#626) + +**Applies to:** every workflow actor — author, reviewer, merger, controller, +operator, reconciler. + +`#494`–`#507` defined the ledger, the canonical state comments, and the +Canonical Thread Handoff shape. This schema owns the *chain*: each actor +consumes exactly one canonical handoff, performs exactly one authorized role, +records the result durably in Gitea, and emits the next complete handoff — +until the controller records final closure. + +Implemented and enforced by `self_propagating_handoff.py`; wired into +`final_report_validator.py` as rule `shared.self_propagating_handoff`. + +## The block + +Post this block into the Gitea issue or PR thread, and include it verbatim in +the final report. It is not an optional prose section. + +```md + +## Canonical Handoff + +```text +REPOSITORY: / +ISSUE: +PR: +WORKFLOW_STATE: +HEAD_SHA: +BASE_BRANCH: +BASE_OR_MERGE_SHA: +ACTING_ROLE: +ACTING_IDENTITY: +COMPLETED_ACTIONS: +VALIDATION_EVIDENCE: +MUTATION_LEDGER: +BLOCKERS: +NEXT_ACTOR: +NEXT_ACTION: +PROHIBITED_ACTIONS: +NEXT_PROMPT: +WORKFLOW_FAILURE_ISSUES: +LAST_UPDATED: +``` +``` + +## Workflow states and the single authorized actor + +| `WORKFLOW_STATE` | `NEXT_ACTOR` | +| --------------------------- | ------------ | +| `needs-author` | `author` | +| `needs-review` | `reviewer` | +| `approved-awaiting-merge` | `merger` | +| `merged-awaiting-controller`| `controller` | +| `blocked` | `operator` | +| `complete` | `none` | + +A merged PR is **not** accepted work: merge transitions to +`merged-awaiting-controller` unless the configured workflow explicitly +authorizes automatic acceptance. + +## Fail-closed rules + +* Every field is required. Only `PR`, `HEAD_SHA`, `BASE_OR_MERGE_SHA`, + `BLOCKERS`, `WORKFLOW_FAILURE_ISSUES`, `NEXT_ACTION`, and `NEXT_PROMPT` may + carry `none`, and `PR`/`HEAD_SHA` only in `needs-author` or `blocked`. +* `NEXT_ACTOR` must equal the actor the declared state authorizes. +* `blocked` requires a concrete `BLOCKERS` entry. +* A non-terminal handoff requires a concrete `NEXT_ACTION` and a + `NEXT_PROMPT` long enough to be ready to run. +* `complete` must carry no `NEXT_ACTION` and no `NEXT_PROMPT`: a finished + workflow terminates instead of manufacturing more work. +* `NEXT_PROMPT` must name the repository and the issue, and must not depend on + outside chat history. The issue or PR thread, workflow documentation, and + live repository state must be sufficient to recover the task. +* The handoff must be posted to Gitea. A chat-only report is not durable + workflow state. + +## Live-state recovery before acting + +The receiving actor re-derives truth from live state instead of trusting the +inherited handoff. `assess_handoff_live_state` detects and fails closed on: +`changed_pr_head`, `stale_approval`, `issue_closed`, `issue_reopened`, +`pr_merged`, `pr_closed_unmerged`, `stale_lease`, `foreign_lease`, +`missing_worktree`, `dirty_worktree`, `namespace_mismatch`, `stale_runtime`, +`changed_base`, and `conflicting_canonical_comments`. + +A changed head invalidates any inherited review or merge handoff; the chain +recovers to `needs-review`. + +## Controller closure + +`accept` is only honored with all four closure proofs recorded: +`acceptance_criteria_satisfied`, `cleanup_complete`, +`canonical_final_state_posted`, `issue_closed_through_workflow`. Otherwise the +work item stays at `merged-awaiting-controller`. + +`request_tests`, `request_proof`, `request_corrections`, and `reopen` return +the work to the author; `return_to_actor` returns it to a named earlier actor. + +## Workflow-failure escalation + +Tooling or workflow defects found while working an item never get folded into +the active feature issue. Each distinct failure carries `classification`, +`linked_issue`, `temporary_impact`, `next_valid_actor`, and `recovery_prompt`. +A failure whose signature already has a durable issue must reuse that issue +instead of filing a duplicate. + +## Applicability + +Enforcement is applicability-gated exactly like the #495 canonical-state gate: +once a report carries the `sph:v1` marker, the `Canonical Handoff` heading, or +a `WORKFLOW_STATE:` line, the full schema is enforced and incomplete handoffs +are rejected. Reports written before the protocol existed are unaffected. diff --git a/skills/llm-project-workflow/schemas/work-issue-final-report.md b/skills/llm-project-workflow/schemas/work-issue-final-report.md index d55f8a9..9f4d059 100644 --- a/skills/llm-project-workflow/schemas/work-issue-final-report.md +++ b/skills/llm-project-workflow/schemas/work-issue-final-report.md @@ -70,4 +70,9 @@ selected issue, and mutation ledger categories (#319, #320). `Read-only diagnostics` (#297). Forbidden claims without proof (#330): `next eligible issue`, `issue claimed`, -`validation passed`, `PR created`, `worktree clean`, `all gates passed`, etc. \ No newline at end of file +`validation passed`, `PR created`, `worktree clean`, `all gates passed`, etc. + +The report must also carry the canonical self-propagating handoff block +(`schemas/self-propagating-handoff.md`, #626) and that block must be posted to +the Gitea issue or PR thread. The next prompt is not an optional prose +section. \ No newline at end of file diff --git a/tests/test_self_propagating_handoff.py b/tests/test_self_propagating_handoff.py new file mode 100644 index 0000000..fb95703 --- /dev/null +++ b/tests/test_self_propagating_handoff.py @@ -0,0 +1,613 @@ +"""Tests for self-propagating canonical handoffs (#626).""" + +from __future__ import annotations + +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from final_report_validator import assess_final_report_validator # noqa: E402 +from self_propagating_handoff import ( # noqa: E402 + HANDOFF_FIELDS, + NEXT_ACTOR_BY_STATE, + WORKFLOW_STATES, + assess_controller_decision, + assess_durable_state_update, + assess_final_report_self_propagating_handoff, + assess_handoff_live_state, + assess_merge_completion_transition, + assess_role_continuation, + assess_self_propagating_handoff, + assess_thread_recoverability, + assess_workflow_failure_escalation, + parse_self_propagating_handoff, + render_self_propagating_handoff, +) + +REPO = "Scaled-Tech-Consulting/Gitea-Tools" + +AUTHOR_PROMPT = ( + "Review PR #900 on Scaled-Tech-Consulting/Gitea-Tools for issue 626 at head " + "aaaa111. Validate the branch, then submit an independent review verdict." +) +REVIEWER_PROMPT = ( + "Merge PR #900 on Scaled-Tech-Consulting/Gitea-Tools for issue 626 once the " + "approval at head aaaa111 still applies to the live head." +) +MERGER_PROMPT = ( + "Accept or reject the merged work for issue 626 on " + "Scaled-Tech-Consulting/Gitea-Tools; verify acceptance criteria then close." +) +CONTROLLER_PROMPT = ( + "Address the controller's requested changes for issue 626 on " + "Scaled-Tech-Consulting/Gitea-Tools, then hand back to an independent reviewer." +) + + +def build_handoff(**overrides): + """Render a valid author -> reviewer handoff, with overrides applied.""" + values = { + "REPOSITORY": REPO, + "ISSUE": "626", + "PR": "900", + "WORKFLOW_STATE": "needs-review", + "HEAD_SHA": "aaaa111", + "BASE_BRANCH": "master", + "BASE_OR_MERGE_SHA": "bbbb222", + "ACTING_ROLE": "author", + "ACTING_IDENTITY": "jcwalker3 (prgs-author)", + "COMPLETED_ACTIONS": "implemented AC1-AC9; opened PR #900", + "VALIDATION_EVIDENCE": "pytest tests/test_self_propagating_handoff.py: 20 passed", + "MUTATION_LEDGER": "branch pushed; PR #900 opened; comment 13547 posted", + "BLOCKERS": "none", + "NEXT_ACTOR": "reviewer", + "NEXT_ACTION": "independently review PR #900 at head aaaa111", + "PROHIBITED_ACTIONS": "merge, self-approve, force-push", + "NEXT_PROMPT": AUTHOR_PROMPT, + "WORKFLOW_FAILURE_ISSUES": "none", + "LAST_UPDATED": "2026-07-21T03:55:00Z", + } + values.update(overrides) + return render_self_propagating_handoff(**values) + + +class RenderAndParseTests(unittest.TestCase): + def test_render_emits_every_canonical_field(self): + body = build_handoff() + parsed = parse_self_propagating_handoff(body) + self.assertIsNotNone(parsed) + for name in HANDOFF_FIELDS: + self.assertIn(name, parsed) + + def test_render_rejects_unknown_workflow_state(self): + with self.assertRaises(ValueError): + build_handoff(WORKFLOW_STATE="almost-done") + + def test_every_state_maps_to_exactly_one_actor(self): + self.assertEqual(set(WORKFLOW_STATES), set(NEXT_ACTOR_BY_STATE)) + + def test_absent_block_parses_as_none(self): + self.assertIsNone(parse_self_propagating_handoff("no handoff here")) + + +class AuthorToReviewerTests(unittest.TestCase): + """Scenario 1: author -> reviewer.""" + + def test_valid_author_handoff_passes(self): + result = assess_self_propagating_handoff(build_handoff()) + self.assertTrue(result["valid"], result["reasons"]) + self.assertEqual(result["next_actor"], "reviewer") + self.assertFalse(result["terminal"]) + + def test_reviewer_may_continue_author_handoff(self): + result = assess_role_continuation( + handoff=build_handoff(), actor_role="reviewer" + ) + self.assertTrue(result["allowed"], result["reasons"]) + self.assertIn("review", result["allowed_actions"]) + + def test_author_may_not_continue_its_own_handoff(self): + result = assess_role_continuation( + handoff=build_handoff(), actor_role="author" + ) + self.assertTrue(result["block"]) + self.assertEqual(result["expected_actor"], "reviewer") + + def test_next_actor_must_match_declared_state(self): + result = assess_self_propagating_handoff( + build_handoff(NEXT_ACTOR="merger") + ) + self.assertTrue(result["block"]) + self.assertTrue( + any("does not match state" in reason for reason in result["reasons"]) + ) + + +class ReviewerToMergerTests(unittest.TestCase): + """Scenario 2: reviewer -> merger.""" + + def build(self, **overrides): + values = { + "WORKFLOW_STATE": "approved-awaiting-merge", + "ACTING_ROLE": "reviewer", + "ACTING_IDENTITY": "reviewer-bot (prgs-reviewer)", + "COMPLETED_ACTIONS": "review 500 APPROVED at aaaa111", + "NEXT_ACTOR": "merger", + "NEXT_ACTION": "merge PR #900 at approved head aaaa111", + "PROHIBITED_ACTIONS": "re-review, commit, push", + "NEXT_PROMPT": REVIEWER_PROMPT, + } + values.update(overrides) + return build_handoff(**values) + + def test_reviewer_handoff_is_valid(self): + result = assess_self_propagating_handoff(self.build()) + self.assertTrue(result["valid"], result["reasons"]) + self.assertEqual(result["next_actor"], "merger") + + def test_merger_may_continue(self): + result = assess_role_continuation(handoff=self.build(), actor_role="merger") + self.assertTrue(result["allowed"], result["reasons"]) + self.assertIn("merge", result["allowed_actions"]) + + +class MergerToControllerTests(unittest.TestCase): + """Scenario 3: merger -> controller.""" + + def test_merge_success_stops_at_controller_boundary(self): + result = assess_merge_completion_transition(merge_succeeded=True) + self.assertEqual(result["next_state"], "merged-awaiting-controller") + self.assertEqual(result["next_actor"], "controller") + self.assertTrue(result["next_prompt_required"]) + + def test_configured_auto_accept_may_complete(self): + result = assess_merge_completion_transition( + merge_succeeded=True, controller_auto_accept=True + ) + self.assertEqual(result["next_state"], "complete") + self.assertFalse(result["next_prompt_required"]) + + def test_failed_merge_keeps_the_work_item_with_the_merger(self): + result = assess_merge_completion_transition(merge_succeeded=False) + self.assertEqual(result["next_state"], "approved-awaiting-merge") + + def test_merger_handoff_names_the_controller(self): + body = build_handoff( + WORKFLOW_STATE="merged-awaiting-controller", + ACTING_ROLE="merger", + ACTING_IDENTITY="merger-bot (prgs-merger)", + COMPLETED_ACTIONS="merged PR #900 as cccc333", + BASE_OR_MERGE_SHA="cccc333", + NEXT_ACTOR="controller", + NEXT_ACTION="verify acceptance criteria and close issue 626", + PROHIBITED_ACTIONS="reopen the PR, re-merge", + NEXT_PROMPT=MERGER_PROMPT, + ) + result = assess_self_propagating_handoff(body) + self.assertTrue(result["valid"], result["reasons"]) + self.assertEqual(result["next_actor"], "controller") + + +class ControllerBackToAuthorTests(unittest.TestCase): + """Scenario 4: controller -> author.""" + + def test_request_corrections_returns_to_author(self): + result = assess_controller_decision(decision="request_corrections") + self.assertFalse(result["block"]) + self.assertEqual(result["next_state"], "needs-author") + self.assertTrue(result["next_prompt_required"]) + + def test_return_to_actor_requires_a_named_target(self): + result = assess_controller_decision(decision="return_to_actor") + self.assertTrue(result["block"]) + + def test_return_to_reviewer_is_supported(self): + result = assess_controller_decision( + decision="return_to_actor", return_to="reviewer" + ) + self.assertEqual(result["next_state"], "needs-review") + + def test_unknown_decision_fails_closed(self): + result = assess_controller_decision(decision="looks-fine") + self.assertTrue(result["block"]) + + def test_controller_handoff_back_to_author_validates(self): + body = build_handoff( + WORKFLOW_STATE="needs-author", + ACTING_ROLE="controller", + ACTING_IDENTITY="controller (operator)", + COMPLETED_ACTIONS="reviewed merged work; requested corrections", + NEXT_ACTOR="author", + NEXT_ACTION="address controller corrections on issue 626", + PROHIBITED_ACTIONS="close the issue, merge", + NEXT_PROMPT=CONTROLLER_PROMPT, + ) + result = assess_self_propagating_handoff(body) + self.assertTrue(result["valid"], result["reasons"]) + + +class StaleHeadRejectionTests(unittest.TestCase): + """Scenario 5: stale-head rejection.""" + + def test_changed_head_invalidates_a_merge_handoff(self): + body = build_handoff( + WORKFLOW_STATE="approved-awaiting-merge", + ACTING_ROLE="reviewer", + NEXT_ACTOR="merger", + NEXT_ACTION="merge PR #900 at approved head aaaa111", + NEXT_PROMPT=REVIEWER_PROMPT, + ) + result = assess_handoff_live_state( + handoff=body, + live={"pr_head_sha": "dddd444", "pr_state": "open"}, + ) + self.assertTrue(result["block"]) + self.assertIn("changed_pr_head", result["kinds"]) + self.assertEqual(result["recovered_state"], "needs-review") + + def test_stale_approval_blocks_the_merger(self): + body = build_handoff( + WORKFLOW_STATE="approved-awaiting-merge", + NEXT_ACTOR="merger", + NEXT_ACTION="merge PR #900", + HEAD_SHA="dddd444", + NEXT_PROMPT=REVIEWER_PROMPT, + ) + result = assess_handoff_live_state( + handoff=body, + live={"pr_head_sha": "dddd444", "approved_head_sha": "aaaa111"}, + ) + self.assertTrue(result["block"]) + self.assertIn("stale_approval", result["kinds"]) + + def test_unchanged_head_is_not_blocked(self): + result = assess_handoff_live_state( + handoff=build_handoff(), + live={ + "pr_head_sha": "aaaa111", + "pr_state": "open", + "issue_state": "open", + "base_branch": "master", + "namespace_role": "reviewer", + }, + ) + self.assertFalse(result["block"], result["reasons"]) + + def test_merged_pr_recovers_to_the_controller_boundary(self): + result = assess_handoff_live_state( + handoff=build_handoff(), + live={"pr_head_sha": "aaaa111", "pr_state": "merged"}, + ) + self.assertIn("pr_merged", result["kinds"]) + self.assertEqual(result["recovered_state"], "merged-awaiting-controller") + + def test_reopened_issue_invalidates_a_complete_handoff(self): + body = build_handoff( + WORKFLOW_STATE="complete", + ACTING_ROLE="controller", + NEXT_ACTOR="none", + NEXT_ACTION="none", + NEXT_PROMPT="none", + ) + result = assess_handoff_live_state( + handoff=body, live={"issue_state": "open"} + ) + self.assertIn("issue_reopened", result["kinds"]) + self.assertEqual(result["recovered_state"], "needs-author") + + def test_foreign_lease_and_worktree_faults_are_detected(self): + result = assess_handoff_live_state( + handoff=build_handoff(), + live={ + "pr_head_sha": "aaaa111", + "lease": {"status": "expired", "session_id": "other-session"}, + "actor_session_id": "my-session", + "worktree": {"present": False, "dirty": True}, + "namespace_role": "author", + "runtime_stale": True, + "base_branch": "dev", + "conflicting_canonical_comments": True, + }, + ) + for kind in ( + "stale_lease", + "foreign_lease", + "missing_worktree", + "dirty_worktree", + "namespace_mismatch", + "stale_runtime", + "changed_base", + "conflicting_canonical_comments", + ): + self.assertIn(kind, result["kinds"]) + + +class BlockedInfrastructurePathTests(unittest.TestCase): + """Scenario 6: blocked infrastructure path.""" + + def build(self, **overrides): + values = { + "WORKFLOW_STATE": "blocked", + "PR": "none", + "HEAD_SHA": "none", + "ACTING_ROLE": "author", + "COMPLETED_ACTIONS": "attempted native publish; MCP mutation rejected", + "BLOCKERS": "gitea_create_pr rejected: namespace unreachable", + "NEXT_ACTOR": "operator", + "NEXT_ACTION": "restore the author MCP namespace", + "PROHIBITED_ACTIONS": "raw git push, curl, force-push", + "NEXT_PROMPT": ( + "Repair the author MCP namespace for " + "Scaled-Tech-Consulting/Gitea-Tools so issue 626 can publish " + "natively, then hand back to the author." + ), + "WORKFLOW_FAILURE_ISSUES": "#640", + } + values.update(overrides) + return build_handoff(**values) + + def test_blocked_handoff_without_pr_is_valid(self): + result = assess_self_propagating_handoff(self.build()) + self.assertTrue(result["valid"], result["reasons"]) + self.assertEqual(result["next_actor"], "operator") + + def test_blocked_requires_a_concrete_blocker(self): + result = assess_self_propagating_handoff(self.build(BLOCKERS="none")) + self.assertTrue(result["block"]) + self.assertTrue( + any("BLOCKERS" in reason for reason in result["reasons"]) + ) + + def test_operator_is_the_only_authorized_continuation(self): + self.assertTrue( + assess_role_continuation(handoff=self.build(), actor_role="operator")[ + "allowed" + ] + ) + self.assertTrue( + assess_role_continuation(handoff=self.build(), actor_role="merger")["block"] + ) + + +class FinalClosureTests(unittest.TestCase): + """Scenario 7: final successful closure.""" + + def build(self, **overrides): + values = { + "WORKFLOW_STATE": "complete", + "ACTING_ROLE": "controller", + "ACTING_IDENTITY": "controller (operator)", + "COMPLETED_ACTIONS": "verified acceptance criteria; closed issue 626", + "BASE_OR_MERGE_SHA": "cccc333", + "NEXT_ACTOR": "none", + "NEXT_ACTION": "none", + "PROHIBITED_ACTIONS": "reopen without new evidence", + "NEXT_PROMPT": "none", + } + values.update(overrides) + return build_handoff(**values) + + def test_terminal_handoff_is_valid_without_a_next_prompt(self): + result = assess_self_propagating_handoff(self.build()) + self.assertTrue(result["valid"], result["reasons"]) + self.assertTrue(result["terminal"]) + + def test_terminal_handoff_must_not_manufacture_more_work(self): + result = assess_self_propagating_handoff( + self.build(NEXT_PROMPT=CONTROLLER_PROMPT) + ) + self.assertTrue(result["block"]) + self.assertTrue( + any("must not carry a NEXT_PROMPT" in r for r in result["reasons"]) + ) + + def test_no_role_may_continue_a_complete_workflow(self): + result = assess_role_continuation(handoff=self.build(), actor_role="author") + self.assertTrue(result["block"]) + + def test_controller_acceptance_requires_full_closure_proof(self): + partial = assess_controller_decision( + decision="accept", + closure_proof={"acceptance_criteria_satisfied": True}, + ) + self.assertTrue(partial["block"]) + self.assertEqual(partial["next_state"], "merged-awaiting-controller") + + full = assess_controller_decision( + decision="accept", + closure_proof={ + "acceptance_criteria_satisfied": True, + "cleanup_complete": True, + "canonical_final_state_posted": True, + "issue_closed_through_workflow": True, + }, + ) + self.assertFalse(full["block"]) + self.assertEqual(full["next_state"], "complete") + self.assertFalse(full["next_prompt_required"]) + + +class IncompleteHandoffRejectionTests(unittest.TestCase): + """Scenario 8: incomplete handoff rejection.""" + + def test_missing_block_is_rejected(self): + result = assess_self_propagating_handoff("Work is done, ping the reviewer.") + self.assertTrue(result["block"]) + self.assertFalse(result["present"]) + + def test_missing_field_is_rejected(self): + body = build_handoff() + body = "\n".join( + line for line in body.splitlines() if not line.startswith("MUTATION_LEDGER:") + ) + result = assess_self_propagating_handoff(body) + self.assertTrue(result["block"]) + self.assertIn("MUTATION_LEDGER", result["missing_fields"]) + + def test_placeholder_field_is_rejected(self): + result = assess_self_propagating_handoff( + build_handoff(VALIDATION_EVIDENCE="TBD") + ) + self.assertTrue(result["block"]) + + def test_stub_next_prompt_is_rejected(self): + result = assess_self_propagating_handoff(build_handoff(NEXT_PROMPT="review it")) + self.assertTrue(result["block"]) + self.assertTrue( + any("ready-to-run" in reason for reason in result["reasons"]) + ) + + def test_prompt_depending_on_outside_chat_is_rejected(self): + prompt = ( + "Continue issue 626 on Scaled-Tech-Consulting/Gitea-Tools using the " + "previous chat for the missing details." + ) + result = assess_thread_recoverability(build_handoff(NEXT_PROMPT=prompt)) + self.assertTrue(result["block"]) + + def test_prompt_must_name_repository_and_issue(self): + prompt = ( + "Please review the pull request at the current head and submit an " + "independent verdict when validation passes." + ) + result = assess_thread_recoverability(build_handoff(NEXT_PROMPT=prompt)) + self.assertTrue(result["block"]) + + def test_self_contained_prompt_is_recoverable(self): + self.assertFalse(assess_thread_recoverability(build_handoff())["block"]) + + def test_chat_only_report_is_not_durable(self): + result = assess_durable_state_update( + handoff_text=build_handoff(), + posted_comment_id=None, + canonical_state_posted=False, + ) + self.assertTrue(result["block"]) + self.assertEqual(len(result["reasons"]), 2) + + def test_posted_handoff_is_durable(self): + result = assess_durable_state_update( + handoff_text=build_handoff(), + posted_comment_id=13550, + canonical_state_posted=True, + ) + self.assertTrue(result["durable"], result["reasons"]) + + +class WorkflowFailureEscalationTests(unittest.TestCase): + """Scenario 9: duplicate workflow-failure issue handling.""" + + def failure(self, **overrides): + values = { + "signature": "lease-cleanup-internal-error", + "classification": "mcp-tool-defect", + "linked_issue": "718", + "temporary_impact": "lease cleanup unavailable this session", + "next_valid_actor": "operator", + "recovery_prompt": "restart the namespace and re-run lease cleanup", + } + values.update(overrides) + return values + + def test_complete_failure_record_passes(self): + result = assess_workflow_failure_escalation( + failures=[self.failure()], active_issue_number=626 + ) + self.assertTrue(result["escalated"], result["reasons"]) + + def test_incomplete_failure_record_fails_closed(self): + result = assess_workflow_failure_escalation( + failures=[self.failure(recovery_prompt="")], active_issue_number=626 + ) + self.assertTrue(result["block"]) + + def test_folding_into_the_active_issue_is_rejected(self): + result = assess_workflow_failure_escalation( + failures=[self.failure(linked_issue="626")], active_issue_number=626 + ) + self.assertTrue(result["block"]) + self.assertTrue( + any("folded into the active work item" in r for r in result["reasons"]) + ) + + def test_known_signature_reuses_the_existing_issue(self): + result = assess_workflow_failure_escalation( + failures=[self.failure()], + active_issue_number=626, + existing_failure_issues=[ + {"signature": "lease-cleanup-internal-error", "number": 718} + ], + ) + self.assertTrue(result["escalated"], result["reasons"]) + self.assertEqual(result["reused_issues"], [ + {"signature": "lease-cleanup-internal-error", "issue": "718"} + ]) + + def test_duplicate_issue_for_known_signature_is_rejected(self): + result = assess_workflow_failure_escalation( + failures=[self.failure(linked_issue="799")], + active_issue_number=626, + existing_failure_issues=[ + {"signature": "lease-cleanup-internal-error", "number": 718} + ], + ) + self.assertTrue(result["block"]) + self.assertTrue( + any("reuse the existing issue #718" in r for r in result["reasons"]) + ) + + def test_same_signature_twice_in_one_session_is_rejected(self): + result = assess_workflow_failure_escalation( + failures=[self.failure(), self.failure()], active_issue_number=626 + ) + self.assertTrue(result["block"]) + + def test_no_failures_is_not_an_error(self): + result = assess_workflow_failure_escalation( + failures=[], active_issue_number=626 + ) + self.assertTrue(result["escalated"]) + + +class FinalReportIntegrationTests(unittest.TestCase): + def test_report_without_the_protocol_is_not_applicable(self): + result = assess_final_report_self_propagating_handoff("## Controller Handoff\n") + self.assertFalse(result["applicable"]) + self.assertFalse(result["block"]) + + def test_report_with_a_complete_handoff_passes(self): + result = assess_final_report_self_propagating_handoff(build_handoff()) + self.assertTrue(result["applicable"]) + self.assertFalse(result["block"], result["reasons"]) + + def test_report_with_an_incomplete_handoff_blocks(self): + result = assess_final_report_self_propagating_handoff( + build_handoff(NEXT_ACTION="") + ) + self.assertTrue(result["block"]) + + def test_validator_blocks_an_incomplete_handoff_in_a_work_issue_report(self): + report = build_handoff(MUTATION_LEDGER="TBD") + result = assess_final_report_validator(report, "work_issue") + self.assertTrue(result["blocked"]) + self.assertTrue( + any( + finding["rule_id"] == "shared.self_propagating_handoff" + for finding in result["findings"] + ) + ) + + def test_validator_ignores_reports_that_predate_the_protocol(self): + result = assess_final_report_validator("plain legacy report", "work_issue") + self.assertFalse( + any( + finding["rule_id"] == "shared.self_propagating_handoff" + for finding in result["findings"] + ) + ) + + +if __name__ == "__main__": + unittest.main()