feat: add canonical state handoff ledger (#494)
Add templates, validators, and documentation for discussion/issue/PR state comments and final-report next-action handoff fields. Wire enforcement into assess_final_report_validator across all workflow task kinds. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,374 @@
|
||||
"""Canonical state handoff ledger for discussions, issues, and PRs (#494)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
HANDOFF_HEADING = "Controller Handoff"
|
||||
|
||||
ISSUE_STATE_FIELDS = (
|
||||
"STATE",
|
||||
"NEXT_ACTOR",
|
||||
"NEXT_ACTION",
|
||||
"NEXT_PROMPT",
|
||||
"STATUS",
|
||||
"RELATED_PRS",
|
||||
"BRANCH",
|
||||
"HEAD_SHA",
|
||||
"VALIDATION",
|
||||
"BLOCKERS",
|
||||
"DUPLICATES_OR_SUPERSEDES",
|
||||
"LAST_UPDATED_BY",
|
||||
)
|
||||
|
||||
PR_STATE_FIELDS = (
|
||||
"STATE",
|
||||
"NEXT_ACTOR",
|
||||
"NEXT_ACTION",
|
||||
"NEXT_PROMPT",
|
||||
"ISSUE",
|
||||
"BASE",
|
||||
"HEAD",
|
||||
"HEAD_SHA",
|
||||
"REVIEW_STATUS",
|
||||
"VALIDATION",
|
||||
"BLOCKERS",
|
||||
"SUPERSEDES",
|
||||
"SUPERSEDED_BY",
|
||||
"MERGE_READY",
|
||||
"LAST_UPDATED_BY",
|
||||
)
|
||||
|
||||
DISCUSSION_STATE_FIELDS = (
|
||||
"STATE",
|
||||
"NEXT_ACTOR",
|
||||
"NEXT_ACTION",
|
||||
"NEXT_PROMPT",
|
||||
"COMMENT_COUNT",
|
||||
"SUBSTANTIVE_COMMENTS",
|
||||
"URGENCY",
|
||||
"BLOCKERS",
|
||||
"LAST_UPDATED_BY",
|
||||
)
|
||||
|
||||
DISCUSSION_SUMMARY_FIELDS = (
|
||||
"DECISION",
|
||||
"ISSUES_TO_CREATE",
|
||||
"NON_GOALS",
|
||||
"UNRESOLVED_QUESTIONS",
|
||||
"NEXT_PROMPT",
|
||||
"LAST_UPDATED_BY",
|
||||
)
|
||||
|
||||
QUEUE_CONTROLLER_FIELDS = (
|
||||
"QUEUE_STATE",
|
||||
"SELECTED_OBJECT",
|
||||
"SELECTED_OBJECT_TYPE",
|
||||
"NEXT_ACTOR",
|
||||
"NEXT_ACTION",
|
||||
"NEXT_PROMPT",
|
||||
"EVIDENCE",
|
||||
"LAST_UPDATED_BY",
|
||||
)
|
||||
|
||||
FINAL_REPORT_NEXT_ACTION_FIELDS = (
|
||||
("Current status", ("current status",)),
|
||||
("Next actor", ("next actor", "next_actor")),
|
||||
("Next action", ("next action", "next_action")),
|
||||
("Next prompt", ("next prompt", "next_prompt")),
|
||||
)
|
||||
|
||||
STATE_COMMENT_KINDS = frozenset({
|
||||
"issue_state",
|
||||
"pr_state",
|
||||
"discussion_state",
|
||||
"discussion_summary",
|
||||
"queue_controller",
|
||||
})
|
||||
|
||||
_FIELDS_BY_KIND = {
|
||||
"issue_state": ISSUE_STATE_FIELDS,
|
||||
"pr_state": PR_STATE_FIELDS,
|
||||
"discussion_state": DISCUSSION_STATE_FIELDS,
|
||||
"discussion_summary": DISCUSSION_SUMMARY_FIELDS,
|
||||
"queue_controller": QUEUE_CONTROLLER_FIELDS,
|
||||
}
|
||||
|
||||
_FIELD_LINE_RE = re.compile(
|
||||
r"^([A-Z][A-Z0-9_]+)\s*:\s*(.*)$",
|
||||
re.MULTILINE,
|
||||
)
|
||||
_HANDOFF_SECTION_RE = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M)
|
||||
_READY_TO_MERGE_RE = re.compile(
|
||||
r"\b(?:ready[_ ]to[_ ]merge|merge[_ ]ready\s*:\s*(?:yes|true))\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_APPROVAL_PROOF_RE = re.compile(
|
||||
r"\b(?:review decision\s*:\s*approve|approved at|approval at current head|"
|
||||
r"formal approval|review_status\s*:\s*approved)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_EXTERNAL_NONE_RE = re.compile(
|
||||
r"^\s*[-*]?\s*external[- ]state mutations\s*:\s*none\s*$",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_LOCK_MUTATION_RE = re.compile(
|
||||
r"\b(?:gitea_lock_issue|issue-locks/|lock file edited)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_ISSUE_DONE_RE = re.compile(
|
||||
r"\b(?:issue (?:done|closed)\b|current status\s*:\s*(?:done|closed)\b|"
|
||||
r"current status\s*:\s*issue complete\b)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_PR_MERGE_PROOF_RE = re.compile(
|
||||
r"\b(?:pr (?:merged|number)|merge result\s*:\s*merged|pr mutations\s*:\s*created)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_DISCUSSION_COMPLETE_RE = re.compile(
|
||||
r"\b(?:discussion complete|discussion ready for issue creation)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_DISCUSSION_SUMMARY_PROOF_RE = re.compile(
|
||||
r"\b(?:discussion summary|issues to create|issues created|linked issues)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
MIN_DISCUSSION_COMMENTS_DEFAULT = 5
|
||||
|
||||
|
||||
def _render_fields(fields: tuple[str, ...], values: dict[str, Any]) -> str:
|
||||
lines = ["```text"]
|
||||
for name in fields:
|
||||
raw = values.get(name, values.get(name.lower(), "none"))
|
||||
text = str(raw).strip() if raw is not None else "none"
|
||||
lines.append(f"{name}: {text or 'none'}")
|
||||
lines.append("```")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def render_issue_state_comment(**values: Any) -> str:
|
||||
"""Render canonical issue state comment body."""
|
||||
return _render_fields(ISSUE_STATE_FIELDS, values)
|
||||
|
||||
|
||||
def render_pr_state_comment(**values: Any) -> str:
|
||||
"""Render canonical PR state comment body."""
|
||||
return _render_fields(PR_STATE_FIELDS, values)
|
||||
|
||||
|
||||
def render_discussion_state_comment(**values: Any) -> str:
|
||||
"""Render canonical discussion state comment body."""
|
||||
return _render_fields(DISCUSSION_STATE_FIELDS, values)
|
||||
|
||||
|
||||
def render_discussion_summary_comment(**values: Any) -> str:
|
||||
"""Render discussion-to-issue conversion summary comment."""
|
||||
return _render_fields(DISCUSSION_SUMMARY_FIELDS, values)
|
||||
|
||||
|
||||
def render_queue_controller_report(**values: Any) -> str:
|
||||
"""Render queue-controller selection report."""
|
||||
return _render_fields(QUEUE_CONTROLLER_FIELDS, values)
|
||||
|
||||
|
||||
def parse_state_comment(text: str) -> dict[str, str]:
|
||||
"""Parse KEY: value lines from a canonical state comment."""
|
||||
parsed: dict[str, str] = {}
|
||||
for match in _FIELD_LINE_RE.finditer(text or ""):
|
||||
parsed[match.group(1)] = match.group(2).strip()
|
||||
return parsed
|
||||
|
||||
|
||||
def _handoff_field_map(report_text: str) -> dict[str, str]:
|
||||
section_match = _HANDOFF_SECTION_RE.search(report_text or "")
|
||||
if not section_match:
|
||||
return {}
|
||||
section = (report_text or "")[section_match.end():]
|
||||
fields: dict[str, str] = {}
|
||||
for line in section.splitlines():
|
||||
stripped = line.strip().lstrip("-*").strip()
|
||||
if ":" not in stripped:
|
||||
continue
|
||||
key, value = stripped.split(":", 1)
|
||||
fields[key.strip().lower()] = value.strip()
|
||||
return fields
|
||||
|
||||
|
||||
def assess_state_comment(text: str, *, kind: str) -> dict[str, Any]:
|
||||
"""Validate a canonical Gitea state comment for *kind*."""
|
||||
normalized = (kind or "").strip().lower()
|
||||
if normalized not in STATE_COMMENT_KINDS:
|
||||
return {
|
||||
"complete": False,
|
||||
"block": True,
|
||||
"missing_fields": [],
|
||||
"reasons": [f"unknown state comment kind '{kind}'"],
|
||||
"safe_next_action": "use a supported state comment kind",
|
||||
}
|
||||
|
||||
required = _FIELDS_BY_KIND[normalized]
|
||||
parsed = parse_state_comment(text)
|
||||
missing = [name for name in required if name not in parsed or not parsed[name].strip()]
|
||||
reasons = [f"state comment missing field: {name}" for name in missing]
|
||||
|
||||
if normalized == "discussion_state":
|
||||
urgency = parsed.get("URGENCY", "").strip().lower()
|
||||
count_raw = parsed.get("COMMENT_COUNT", "").strip()
|
||||
try:
|
||||
count = int(count_raw)
|
||||
except (TypeError, ValueError):
|
||||
count = -1
|
||||
if (
|
||||
urgency not in {"urgent", "trivial"}
|
||||
and count >= 0
|
||||
and count < MIN_DISCUSSION_COMMENTS_DEFAULT
|
||||
):
|
||||
reasons.append(
|
||||
f"discussion has {count} comments; need at least "
|
||||
f"{MIN_DISCUSSION_COMMENTS_DEFAULT} substantive comments "
|
||||
"or URGENCY: urgent|trivial"
|
||||
)
|
||||
|
||||
block = bool(missing or reasons)
|
||||
return {
|
||||
"complete": not block,
|
||||
"block": block,
|
||||
"missing_fields": missing,
|
||||
"parsed_fields": parsed,
|
||||
"reasons": reasons,
|
||||
"safe_next_action": (
|
||||
"fill all required state comment fields before posting"
|
||||
if block
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def assess_final_report_next_action_handoff(report_text: str) -> dict[str, Any]:
|
||||
"""Require current status plus next actor/action/prompt in Controller Handoff."""
|
||||
fields = _handoff_field_map(report_text)
|
||||
if not fields:
|
||||
return {
|
||||
"complete": False,
|
||||
"block": True,
|
||||
"missing_fields": [name for name, _ in FINAL_REPORT_NEXT_ACTION_FIELDS],
|
||||
"reasons": [f"final report has no section titled exactly '{HANDOFF_HEADING}'"],
|
||||
"safe_next_action": f"add a complete {HANDOFF_HEADING} section",
|
||||
}
|
||||
|
||||
missing = []
|
||||
for name, aliases in FINAL_REPORT_NEXT_ACTION_FIELDS:
|
||||
value = ""
|
||||
for alias in aliases:
|
||||
if alias in fields:
|
||||
value = fields[alias]
|
||||
break
|
||||
if not value or value.lower() in {"none", "n/a", "not verified in this session", ""}:
|
||||
missing.append(name)
|
||||
|
||||
reasons = [f"handoff missing next-action field: {name}" for name in missing]
|
||||
block = bool(missing)
|
||||
return {
|
||||
"complete": not block,
|
||||
"block": block,
|
||||
"missing_fields": missing,
|
||||
"reasons": reasons,
|
||||
"safe_next_action": (
|
||||
"add Current status, Next actor, Next action, and Next prompt to Controller Handoff"
|
||||
if block
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def assess_contradictory_state_handoff(report_text: str) -> dict[str, Any]:
|
||||
"""Fail closed on contradictory queue/state claims in final reports."""
|
||||
text = report_text or ""
|
||||
fields = _handoff_field_map(text)
|
||||
reasons: list[str] = []
|
||||
|
||||
review_decision = fields.get("review decision", fields.get("review status", ""))
|
||||
has_approval = (
|
||||
review_decision.lower() in {"approve", "approved"}
|
||||
or bool(_APPROVAL_PROOF_RE.search(text))
|
||||
)
|
||||
if _READY_TO_MERGE_RE.search(text) and not has_approval:
|
||||
reasons.append(
|
||||
"report claims ready to merge without approval-at-current-head proof"
|
||||
)
|
||||
|
||||
lock_val = fields.get("external-state mutations", "")
|
||||
mutation_lines = " ".join(
|
||||
fields.get(key, "")
|
||||
for key in (
|
||||
"issue mutations",
|
||||
"mcp/gitea mutations",
|
||||
"external-state mutations",
|
||||
"claim/lock state",
|
||||
)
|
||||
)
|
||||
if (
|
||||
(_EXTERNAL_NONE_RE.search(text) or lock_val.lower() == "none")
|
||||
and _LOCK_MUTATION_RE.search(mutation_lines)
|
||||
):
|
||||
reasons.append(
|
||||
"report claims External-state mutations: none while lock "
|
||||
"activity is documented in mutation fields"
|
||||
)
|
||||
|
||||
if _ISSUE_DONE_RE.search(text) and not _PR_MERGE_PROOF_RE.search(text):
|
||||
reasons.append(
|
||||
"report claims issue done/complete without PR or merge proof"
|
||||
)
|
||||
|
||||
if _DISCUSSION_COMPLETE_RE.search(text) and not _DISCUSSION_SUMMARY_PROOF_RE.search(text):
|
||||
reasons.append(
|
||||
"report claims discussion complete without summary or issue-creation proof"
|
||||
)
|
||||
|
||||
review_status = fields.get("review status", fields.get("review decision", ""))
|
||||
merge_ready = fields.get("merge ready", "")
|
||||
if (
|
||||
merge_ready.lower() in {"yes", "true", "ready"}
|
||||
and review_status.lower() not in {"approve", "approved"}
|
||||
and not _APPROVAL_PROOF_RE.search(text)
|
||||
):
|
||||
reasons.append(
|
||||
"MERGE_READY or merge-ready claim without approved review status"
|
||||
)
|
||||
|
||||
block = bool(reasons)
|
||||
return {
|
||||
"complete": not block,
|
||||
"block": block,
|
||||
"reasons": reasons,
|
||||
"safe_next_action": (
|
||||
"correct contradictory state claims or add missing proof"
|
||||
if block
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def assess_state_handoff_ledger_report(report_text: str) -> dict[str, Any]:
|
||||
"""Composite #494 assessment for final reports."""
|
||||
next_action = assess_final_report_next_action_handoff(report_text)
|
||||
contradictions = assess_contradictory_state_handoff(report_text)
|
||||
reasons = list(next_action.get("reasons") or []) + list(contradictions.get("reasons") or [])
|
||||
block = bool(next_action.get("block") or contradictions.get("block"))
|
||||
return {
|
||||
"complete": not block,
|
||||
"block": block,
|
||||
"next_action": next_action,
|
||||
"contradictions": contradictions,
|
||||
"reasons": reasons,
|
||||
"safe_next_action": (
|
||||
next_action.get("safe_next_action")
|
||||
if next_action.get("block")
|
||||
else contradictions.get("safe_next_action")
|
||||
if contradictions.get("block")
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
Reference in New Issue
Block a user