Compare commits

...
Author SHA1 Message Date
sysadminandClaude Opus 4.8 a002864a06 fix(mcp): add sanctioned gitea_edit_issue and a doc/registry drift guard (Closes #781)
The gitea-workflow documentation named a gitea_edit_issue tool that no
namespace had ever registered. There was no sanctioned MCP path to change an
issue title or body at all: the only edit tool, gitea_edit_pr, PATCHes the
pull-request endpoint. An authorized body correction on issue #780 therefore
had to be recorded as a discussion comment instead.

Add edit_issue.py as the authoritative rule and gitea_edit_issue as the tool
built on it:

- Only the fields the caller names are sent, so labels, state, assignee, and
  milestone cannot be overwritten from a stale read.
- A pull-request number is refused. Gitea serves pull requests from the same
  /issues/{n} collection, so without that check the issue path would quietly
  become a second, ungated PR edit path. gitea_edit_pr stays PR-only.
- Structurally invalid requests raise before any credential or network work;
  a request that would change nothing is reported as an explicit no-op with a
  next action rather than a silent success.
- Read-after-write proves the applied title/body and proves that state,
  labels, assignees, and milestone did not move. Transport failures on the
  pre-read, the PATCH, and the read-back are each reported, redacted, with
  the correct performed/verified state.

Gates match every other issue mutation: profile permission via the shared
capability map (resolver task edit_issue, gitea.issue.comment), preflight
purity, branches worktree validation, anti-stomp inventory membership, and
audited mutation.

Fix the drift that hid this. docs/mcp-tool-inventory.md is now the canonical
registered-tool list, and mcp_tool_inventory.py compares it to the live
registry in both directions, plus checks that every tool named under skills/
is registered. The new guard immediately found a second instance of the same
defect: gitea_record_pre_review_command had lost its @mcp.tool() decorator
while the canonical review workflow still instructed reviewers to call it, so
that registration is restored.

Validation:
PASSED: venv/bin/python -m pytest tests/test_issue_781_edit_issue_tool.py -s -q
  — 50 passed
FAILED: venv/bin/python -m pytest -s -q — 4095 passed, 11 failed, 6 skipped.
  The identical 11 tests fail on clean master 8e149e6 with no changes applied
  (254 passed, 11 failed across those five files), so they are pre-existing
  and proven by a baseline run rather than asserted.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-21 16:52:23 -04:00
sysadmin 8e149e6cfa Merge pull request 'fix(workflow): retire status:pr-open on every terminal PR transition (Closes #780)' (#782) from fix/issue-780-terminal-pr-open-label-cleanup into master 2026-07-21 15:14:21 -05:00
sysadminandClaude Opus 4.8 ed0e8c82de fix(workflow): retire status:pr-open on every terminal PR transition (Closes #780)
status:pr-open was applied by gitea_create_pr and never removed again. Every
terminal path finished without touching it, so a repository audit found 40
closed issues still advertising an open PR.

Add terminal_pr_label_cleanup.py as the single authoritative rule and route
every sanctioned terminal path through it, so the paths cannot drift:

- merge (gitea_merge_pr)
- close without merge (gitea_edit_pr)
- supersession/abandonment (gitea_reconcile_superseded_by_merged_pr)
- already-landed reconciliation (gitea_reconcile_already_landed_pr)
- controller closure (gitea_close_issue)
- retry/recovery (new gitea_cleanup_terminal_pr_labels)

The rule removes only status:pr-open, preserves every other label, allows an
empty resulting set, is a no-op when the label is absent (so retries are
safe), and confirms the outcome by read-after-write rather than assumption.

Controller closure runs the cleanup before the state change and fails closed
if it cannot be completed and verified; closing first would bake in the stale
label with no later step to catch it. Post-merge cleanup never blocks the
merge, which already happened, and reports failures with a safe next action.

Also:
- gitea_assess_terminal_label_hygiene: read-only terminal validation that
  reports residual status:pr-open, exempting issues with a genuinely open PR.
- _put_issue_label_names now accepts Gitea's empty response body when the
  requested set is empty, so clearing the last label works.
- test_audit's close_issue fixture keys on the request instead of call order,
  since closing now also reads labels for the cleanup and its read-back.

Docs: label-taxonomy terminal-transition section, runbook pointer, and the
review-merge / reconcile-landed final-report terminal-label requirements.

Suite: 4045 passed, 11 failed, 6 skipped. The same 11 failures reproduce on
clean master df31674 (4010 passed, 11 failed) and are pre-existing.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-21 15:41:24 -04:00
sysadmin df3167488c Merge pull request 'feat: enforce self-propagating canonical handoffs through controller closure (Closes #626)' (#779) from feat/issue-626-self-propagating-handoffs into master 2026-07-21 12:53:50 -05:00
jcwalker3andClaude Opus 4.8 ddc9b97d40 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) <[email protected]>
2026-07-21 02:33:51 -05:00
sysadmin 1d11cbab0f Merge pull request 'fix(allocator): pre-rank exclusions and candidates_json transport (Closes #776)' (#777) from fix/issue-776-allocator-pre-rank-exclusions into master 2026-07-20 22:33:05 -05:00
sysadminandClaude Opus 4.8 d17f055e86 fix(allocator): pre-rank exclusions and candidates_json transport (#776)
Expose exclude_issue_numbers on gitea_allocate_next_work, remove excluded
numbers before ranking, normalize decoded-list and JSON-string
candidates_json fail-closed, and return candidate-set fingerprints for
dry-run/apply CAS. Same-owner leases on excluded issues surface a
structured resume/release blocker.

Closes #776

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-20 22:40:39 -04:00
sysadmin 52ded0ea71 Merge pull request 'fix: mutation budget counts server-side changes only (Closes #617)' (#775) from fix/issue-617-mutation-budget-classifier into master 2026-07-20 20:28:32 -05:00
jcwalker3andClaude Opus 4.8 296601647d fix: mutation budget counts server-side changes only (Closes #617)
Auto-mode classifier now distinguishes local validator rejection,
capability-gate rejection, transport failure before API, and successful
server-side mutation. Pre-API validator failures no longer consume
server-side mutation budget; the final report separately accounts for
local failed attempts, blocked API attempts, and successful server-side
mutations.

Recovered from preserved unpublished commit b46f0f9 via native MCP
unpublished-claim recovery (#772) and author-worktree lock binding (#618),
reconciled onto current master.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-20 18:58:40 -05:00
sysadmin 702ceb2480 Merge pull request 'fix(mcp): recover clean unpublished author work after the owning session exits (Closes #772)' (#774) from fix/issue-772-unpublished-claim-recovery into master 2026-07-20 15:52:07 -05:00
sysadmin ccfaa0ec0c Merge pull request 'fix: exclude foreign-claimed work from allocation (Closes #765)' (#773) from fix/issue-765-allocator-foreign-lease into master 2026-07-20 14:24:58 -05:00
jcwalker3andClaude Opus 4.8 ad13d872df fix: exclude foreign-claimed work from allocation instead of blockading the queue (Closes #765)
Recovered preserved candidate d06198b onto current master 0c2f45a via clean
cherry-pick. Active foreign leases are excluded before allocator ranking;
controller_instance_id ownership is persisted; dashboard matches allocator
exclusion. Stable patch-id bacafc5f… preserved.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-20 13:52:39 -05:00
30 changed files with 6803 additions and 76 deletions
+476 -10
View File
@@ -18,10 +18,12 @@ after they exist as normal issues; this module never assigns incidents.
from __future__ import annotations
import hashlib
import json
import os
import uuid
from dataclasses import dataclass, field
from typing import Any, Sequence
from typing import Any, Mapping, Sequence
from control_plane_db import (
ControlPlaneDB,
@@ -40,6 +42,22 @@ OUTCOME_NEEDS_CONTROLLER = "needs_controller"
OUTCOME_NO_SAFE = "no_safe_work"
OUTCOME_ROLE_INELIGIBLE = "role_ineligible"
OUTCOME_PREVIEW = "preview" # dry-run only (apply=false)
# #765: ownership could not be established for every remaining candidate.
OUTCOME_OWNERSHIP_DEFECT = "allocator_ownership_defect"
# #776: excluded issue still carries a live same-owner lease — resume or release.
OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE = "blocked_by_excluded_own_lease"
# #776: dry-run/apply candidate-set fingerprint mismatch (CAS drift).
OUTCOME_CANDIDATE_SET_DRIFT = "candidate_set_drift"
# #765 skip reason code for work already claimed by a different controller.
SKIP_CLAIMED_BY_OTHER_SESSION = "claimed_by_other_session"
# #776: controller-supplied pre-rank exclusion.
SKIP_EXCLUDED_BY_CONTROLLER = "excluded_by_controller"
# Ownership verdicts for a live claim on a candidate (#765).
OWNERSHIP_OWN = "own"
OWNERSHIP_FOREIGN = "foreign"
OWNERSHIP_UNKNOWN = "unknown"
# Human-readable statement of how a winner is chosen (#758 AC10). Reported
# alongside allocator results so the flat status:ready tier and its
@@ -144,9 +162,76 @@ class SkipRecord:
kind: str
number: int
reason: str
reason_code: str | None = None
def as_dict(self) -> dict[str, Any]:
return {"kind": self.kind, "number": self.number, "reason": self.reason}
return {
"kind": self.kind,
"number": self.number,
"reason": self.reason,
"reason_code": self.reason_code,
}
CONTROLLER_INSTANCE_ENV = "GITEA_CONTROLLER_INSTANCE_ID"
def resolve_controller_instance_id(
env: Mapping[str, str] | None = None,
) -> str | None:
"""Return this controller's stable identity, or ``None`` if undeclared.
Deliberately has no derived fallback. The obvious candidates are unsafe:
``session_id`` is regenerated per invocation, and the MCP process pid is
shared by every controller attached to the same daemon — two independent
controllers really do report the same pid and profile. Guessing from either
would let one controller adopt another's lease, which is the failure #765
exists to prevent. When this returns ``None``, live claims are treated as
unidentified: they are excluded from selection and reported as ownership
defects rather than adopted.
"""
source = env if env is not None else os.environ
return (source.get(CONTROLLER_INSTANCE_ENV) or "").strip() or None
def classify_claim_ownership(
claim: dict[str, Any] | None,
*,
session_id: str | None,
controller_instance_id: str | None,
) -> str | None:
"""Classify a live claim as own / foreign / unknown ownership (#765).
Returns ``None`` when the candidate carries no live claim.
Session ids are regenerated per allocator invocation, so they only prove
ownership positively (an exact match is certainly this session). The
durable signal is ``controller_instance_id``. When either side lacks one,
ownership is *unknown*: the allocator must not assume that a lease sharing
the same profile belongs to this controller, so unknown is treated as
not-ours for selection purposes and reported as an ownership defect.
"""
if not claim:
return None
claim_session = str(claim.get("session_id") or "").strip()
claim_instance = str(claim.get("controller_instance_id") or "").strip()
own_session = str(session_id or "").strip()
own_instance = str(controller_instance_id or "").strip()
if claim_session and own_session and claim_session == own_session:
return OWNERSHIP_OWN
if claim_instance and own_instance:
return (
OWNERSHIP_OWN if claim_instance == own_instance else OWNERSHIP_FOREIGN
)
if not claim_instance and not own_instance:
# Neither side declares a controller identity. The session ids differ
# (an exact match returned OWN above), so this is simply someone
# else's lease: foreign, and we wait rather than adopt.
return OWNERSHIP_FOREIGN
# Exactly one side is identified, so the two cannot be compared: this may
# or may not be our own task under a different session id. Never guess.
return OWNERSHIP_UNKNOWN
def normalize_role(role: str | None, *, profile_name: str | None = None) -> str:
@@ -203,8 +288,16 @@ def classify_skip(
*,
role: str,
terminal_pr: int | None,
claim_ownership: str | None = None,
) -> str | None:
"""Return skip reason, or None if candidate is selectable for *role*."""
"""Return skip reason, or None if candidate is selectable for *role*.
*claim_ownership* (#765) is the verdict from
:func:`classify_claim_ownership` for this candidate's live claim. Foreign
and unknown claims are excluded so one session's in-progress task can never
blockade the queue for a different controller; ``own`` stays selectable so
a controller can resume its own work.
"""
if c.state in ("merged", "closed"):
return f"{c.kind}#{c.number} is {c.state}; never assign"
if c.blocked or "status:blocked" in c.labels:
@@ -214,6 +307,16 @@ def classify_skip(
c.dependency_reason
or f"{c.kind}#{c.number} has unmet dependencies"
)
if claim_ownership in (OWNERSHIP_FOREIGN, OWNERSHIP_UNKNOWN):
detail = (
"owned by another controller instance"
if claim_ownership == OWNERSHIP_FOREIGN
else "owner could not be identified; never adopt on a guess"
)
return (
f"{c.kind}#{c.number} {SKIP_CLAIMED_BY_OTHER_SESSION}: "
f"active lease {detail}"
)
if c.already_claimed_elsewhere:
return f"{c.kind}#{c.number} already claimed elsewhere"
if c.kind == "pr" and not (c.head_sha or "").strip():
@@ -274,6 +377,113 @@ def sort_candidates(candidates: Sequence[WorkCandidate]) -> list[WorkCandidate]:
)
def _require_strict_int(value: Any, *, field: str) -> int:
"""Parse an issue number; reject bools and non-integers (#776)."""
if isinstance(value, bool) or not isinstance(value, int):
raise ValueError(
f"{field} must be an integer (booleans and non-integers rejected; "
f"got {type(value).__name__})"
)
return int(value)
def normalize_exclude_issue_numbers(
exclude_issue_numbers: Any = None,
) -> list[int]:
"""Normalize controller-supplied exclusions to a sorted unique int list (#776).
``None`` / omitted → empty list (existing behavior). Accepts a list/tuple of
integers. Rejects scalars, bools-as-ints, nested structures, and strings.
"""
if exclude_issue_numbers is None:
return []
if isinstance(exclude_issue_numbers, (str, bytes)) or not isinstance(
exclude_issue_numbers, (list, tuple)
):
raise ValueError(
"exclude_issue_numbers must be a list of integers "
f"(got {type(exclude_issue_numbers).__name__})"
)
out: list[int] = []
seen: set[int] = set()
for idx, raw in enumerate(exclude_issue_numbers):
num = _require_strict_int(raw, field=f"exclude_issue_numbers[{idx}]")
if num not in seen:
seen.add(num)
out.append(num)
return sorted(out)
def candidate_set_fingerprint(
candidates: Sequence[WorkCandidate],
*,
exclude_issue_numbers: Sequence[int] | None = None,
) -> str:
"""Stable CAS fingerprint of normalized candidate set + exclusions (#776 AC4)."""
payload = {
"candidates": sorted(
({"kind": c.kind, "number": int(c.number)} for c in candidates),
key=lambda x: (x["kind"], x["number"]),
),
"exclude_issue_numbers": list(
normalize_exclude_issue_numbers(exclude_issue_numbers)
),
}
blob = json.dumps(payload, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(blob.encode("utf-8")).hexdigest()
def normalize_candidates_payload(raw: Any) -> list[WorkCandidate]:
"""Decode MCP ``candidates_json`` from list or JSON string (#776 AC3).
Accepts:
* an already-decoded ``list`` of candidate dicts (native MCP transport);
* a valid JSON string that decodes to such a list (backward compatible).
Rejects malformed JSON, scalars, non-list containers, invalid records,
booleans-as-integers, and unsupported types with fail-closed ``ValueError``.
"""
if raw is None:
raise ValueError("candidates_json is empty")
if isinstance(raw, (bytes, bytearray)):
try:
raw = raw.decode("utf-8")
except Exception as exc: # noqa: BLE001
raise ValueError(
f"candidates_json bytes are not valid utf-8: {exc}"
) from exc
if isinstance(raw, str):
text = raw.strip()
if not text:
raise ValueError("candidates_json string is empty")
try:
decoded = json.loads(text)
except json.JSONDecodeError as exc:
raise ValueError(
f"malformed candidates_json JSON: {exc.msg} at pos {exc.pos}"
) from exc
raw = decoded
if not isinstance(raw, list):
raise ValueError(
"candidates_json must be a JSON list (or already-decoded list); "
f"got {type(raw).__name__}"
)
candidates: list[WorkCandidate] = []
for idx, item in enumerate(raw):
if not isinstance(item, dict):
raise ValueError(
f"invalid candidate record at index {idx}: expected object, "
f"got {type(item).__name__}"
)
try:
candidates.append(candidate_from_dict(item))
except (KeyError, TypeError, ValueError, InvalidWorkKindError) as exc:
raise ValueError(
f"invalid candidate record at index {idx}: {exc}"
) from exc
return candidates
def allocate_next_work(
db: ControlPlaneDB,
*,
@@ -287,12 +497,22 @@ def allocate_next_work(
profile_name: str | None = None,
username: str | None = None,
lease_ttl_seconds: int | None = None,
controller_instance_id: str | None = None,
claims: Mapping[tuple[str, int], dict[str, Any]] | None = None,
exclude_issue_numbers: Sequence[int] | None = None,
expected_candidate_set_fingerprint: str | None = None,
) -> dict[str, Any]:
"""Select and optionally reserve the next work unit via control-plane DB.
*apply=False* (default): dry-run selection only — no lease/assignment.
*apply=True*: atomic ``assign_and_lease`` for the selected candidate.
*exclude_issue_numbers* (#776): numbers removed before ranking. Omitted /
empty preserves prior behavior.
*expected_candidate_set_fingerprint* (#776 AC4): when set on apply, rejects
material candidate-set drift vs a prior dry-run.
Never uses file locks or comment-only leases as the assignment source.
"""
if db is None:
@@ -328,6 +548,7 @@ def allocate_next_work(
role=role_norm,
profile=profile_name,
pid=os.getpid(),
controller_instance_id=controller_instance_id,
)
except Exception as exc: # noqa: BLE001 — surface structured
return {
@@ -369,30 +590,238 @@ def allocate_next_work(
}
terminal_pr = int(terminal["terminal_pr"]) if terminal else None
# #765: live claims exclude work owned by a *different* controller before
# ranking, so one session's in-progress task cannot blockade the queue.
# #776 AC5: load live claims in this call path immediately before selection
# (and before apply reserve) so ownership is never stale within the
# allocation attempt. Test callers may inject *claims* explicitly.
if claims is None:
try:
claims = db.list_active_claims(remote=remote, org=org, repo=repo)
except Exception as exc: # noqa: BLE001
return {
"success": False,
"outcome": OUTCOME_NO_SAFE,
"reasons": [
f"active claim lookup failed: {exc} (fail closed, #765)"
],
"skipped": [],
"assignment": None,
"substrate": "control_plane_db",
}
try:
exclude_nums = normalize_exclude_issue_numbers(exclude_issue_numbers)
except ValueError as exc:
return {
"success": False,
"outcome": OUTCOME_NO_SAFE,
"apply": bool(apply),
"reasons": [
f"invalid exclude_issue_numbers: {exc} (fail closed, #776)"
],
"skipped": [],
"assignment": None,
"substrate": "control_plane_db",
}
exclude_set = set(exclude_nums)
cas_fp = candidate_set_fingerprint(
candidates, exclude_issue_numbers=exclude_nums
)
expected_fp = (expected_candidate_set_fingerprint or "").strip() or None
if expected_fp and expected_fp != cas_fp:
return {
"success": False,
"outcome": OUTCOME_CANDIDATE_SET_DRIFT,
"apply": bool(apply),
"reasons": [
"candidate-set fingerprint drift: apply rejected rather than "
"silently leasing a different candidate (#776 AC4)"
],
"candidate_set_fingerprint": cas_fp,
"expected_candidate_set_fingerprint": expected_fp,
"exclude_issue_numbers": list(exclude_nums),
"skipped": [],
"assignment": None,
"substrate": "control_plane_db",
"file_lock_only": False,
"comment_lease_only": False,
}
skipped: list[SkipRecord] = []
ordered = sort_candidates(list(candidates))
claims_excluded: list[dict[str, Any]] = []
ownership_defects: list[dict[str, Any]] = []
controller_excluded: list[dict[str, Any]] = []
# #776 AC2: remove excluded numbers *before* ranking / selection / lease.
rankable: list[WorkCandidate] = []
for c in candidates:
if int(c.number) in exclude_set:
reason = (
f"{c.kind}#{c.number} {SKIP_EXCLUDED_BY_CONTROLLER}: "
"controller pre-rank exclusion"
)
skipped.append(
SkipRecord(
c.kind,
c.number,
reason,
SKIP_EXCLUDED_BY_CONTROLLER,
)
)
controller_excluded.append(
{
"kind": c.kind,
"number": c.number,
"reason_code": SKIP_EXCLUDED_BY_CONTROLLER,
}
)
# #776 AC5: same-owner live lease on an excluded issue is a
# structured resume/release blocker, never a silent strand.
claim = claims.get((c.kind, int(c.number))) if claims else None
ownership = classify_claim_ownership(
claim,
session_id=session_id,
controller_instance_id=controller_instance_id,
)
if ownership == OWNERSHIP_OWN and claim:
return {
"success": True,
"outcome": OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE,
"apply": bool(apply),
"role": role_norm,
"profile_name": profile_name,
"username": username,
"session_id": session_id,
"remote": remote,
"org": org,
"repo": repo,
"selected": None,
"expected_role_next": None,
"reasons": [
f"{c.kind}#{c.number} is excluded_by_controller but "
"carries a live same-owner lease; resume or release "
"that lease before allocating other work (#776 AC5)"
],
"skipped": [s.as_dict() for s in skipped],
"terminal_pr": terminal_pr,
"assignment": None,
"substrate": "control_plane_db",
"file_lock_only": False,
"comment_lease_only": False,
"controller_instance_id": controller_instance_id,
"claims_excluded": list(claims_excluded),
"ownership_defects": list(ownership_defects),
"controller_excluded": list(controller_excluded),
"exclude_issue_numbers": list(exclude_nums),
"candidate_set_fingerprint": cas_fp,
"blocked_lease": {
"kind": c.kind,
"number": c.number,
"lease_id": claim.get("lease_id"),
"owner_session_id": claim.get("session_id"),
"owner_controller_instance_id": claim.get(
"controller_instance_id"
),
"expires_at": claim.get("expires_at"),
"safe_next_action": (
"resume the same-owner lease or release it, then "
"re-run allocation without stranding the excluded "
"issue"
),
},
}
continue
rankable.append(c)
ordered = sort_candidates(rankable)
selected: WorkCandidate | None = None
for c in ordered:
reason = classify_skip(c, role=role_norm, terminal_pr=terminal_pr)
claim = claims.get((c.kind, int(c.number))) if claims else None
ownership = classify_claim_ownership(
claim,
session_id=session_id,
controller_instance_id=controller_instance_id,
)
reason = classify_skip(
c,
role=role_norm,
terminal_pr=terminal_pr,
claim_ownership=ownership,
)
if reason:
skipped.append(SkipRecord(c.kind, c.number, reason))
is_claim_skip = SKIP_CLAIMED_BY_OTHER_SESSION in reason
skipped.append(
SkipRecord(
c.kind,
c.number,
reason,
SKIP_CLAIMED_BY_OTHER_SESSION if is_claim_skip else None,
)
)
if is_claim_skip and claim:
record = {
"kind": c.kind,
"number": c.number,
"ownership": ownership,
"lease_id": claim.get("lease_id"),
"owner_session_id": claim.get("session_id"),
"owner_controller_instance_id": claim.get(
"controller_instance_id"
),
"expires_at": claim.get("expires_at"),
}
claims_excluded.append(record)
if ownership == OWNERSHIP_UNKNOWN:
ownership_defects.append(record)
continue
selected = c
break
if selected is None:
# If terminal lock blocks all review work, surface that explicitly.
owner_session_id: str | None = None
if terminal_pr is not None and role_norm in (ROLE_REVIEWER, ROLE_MERGER):
outcome = OUTCOME_BLOCKED_TERMINAL
reasons = [
f"no safe work for role '{role_norm}': active terminal-review "
f"lock on PR #{terminal_pr} (resolve terminal path first, #332/#600)"
]
elif ownership_defects:
# #765: every remaining candidate is claimed and at least one owner
# could not be identified. Report the defect; never adopt.
outcome = OUTCOME_OWNERSHIP_DEFECT
reasons = [
f"no safe assignable work for role '{role_norm}': "
f"{len(ownership_defects)} candidate(s) carry an active lease "
"whose controller ownership could not be established. Record a "
"controller_instance_id on those sessions; the allocator will "
"not assume a shared profile means shared ownership (#765)."
]
elif claims_excluded:
outcome = OUTCOME_WAIT
reasons = [
f"no unclaimed work for role '{role_norm}': "
f"{len(claims_excluded)} candidate(s) are actively claimed by "
"another controller. Waiting; their leases are not adopted (#765)."
]
# Preserve the pre-#765 wait contract: name the blocking owner.
owner_session_id = claims_excluded[0].get("owner_session_id")
elif controller_excluded and not ordered:
# #776 AC7: every candidate was controller-excluded → wait / no lease.
outcome = OUTCOME_WAIT
reasons = [
f"no assignable work for role '{role_norm}': all "
f"{len(controller_excluded)} candidate(s) were removed by "
f"{SKIP_EXCLUDED_BY_CONTROLLER} before ranking; no assignment "
"or lease created (#776 AC7)"
]
else:
outcome = OUTCOME_NO_SAFE
reasons = [
f"no safe assignable work for role '{role_norm}' "
f"among {len(ordered)} candidates"
f"among {len(ordered)} rankable candidates "
f"({len(controller_excluded)} controller-excluded)"
]
return {
"success": True,
@@ -414,6 +843,13 @@ def allocate_next_work(
"substrate": "control_plane_db",
"file_lock_only": False,
"comment_lease_only": False,
"controller_instance_id": controller_instance_id,
"claims_excluded": list(claims_excluded),
"ownership_defects": list(ownership_defects),
"controller_excluded": list(controller_excluded),
"exclude_issue_numbers": list(exclude_nums),
"candidate_set_fingerprint": cas_fp,
"owner_session_id": owner_session_id,
"downstream_note": (
"#612 incident bridge remains downstream of #600; "
"allocator never assigns raw monitoring incidents"
@@ -460,6 +896,12 @@ def allocate_next_work(
"substrate": "control_plane_db",
"file_lock_only": False,
"comment_lease_only": False,
"controller_instance_id": controller_instance_id,
"claims_excluded": list(claims_excluded),
"ownership_defects": list(ownership_defects),
"controller_excluded": list(controller_excluded),
"exclude_issue_numbers": list(exclude_nums),
"candidate_set_fingerprint": cas_fp,
"downstream_note": (
"#612 incident bridge remains downstream of #600; "
"allocator never assigns raw monitoring incidents"
@@ -583,6 +1025,12 @@ def allocate_next_work(
"substrate": "control_plane_db",
"file_lock_only": False,
"comment_lease_only": False,
"controller_instance_id": controller_instance_id,
"claims_excluded": list(claims_excluded),
"ownership_defects": list(ownership_defects),
"controller_excluded": list(controller_excluded),
"exclude_issue_numbers": list(exclude_nums),
"candidate_set_fingerprint": cas_fp,
"downstream_note": (
"#612 incident bridge remains downstream of #600; "
"allocator never assigns raw monitoring incidents"
@@ -614,14 +1062,32 @@ def _next_command(role: str, c: WorkCandidate) -> str:
def candidate_from_dict(data: dict[str, Any]) -> WorkCandidate:
"""Build a WorkCandidate from a plain dict (tests / MCP inventory)."""
"""Build a WorkCandidate from a plain dict (tests / MCP inventory).
#776: reject booleans-as-integers and non-int numbers fail-closed.
"""
if "number" not in data:
raise KeyError("number")
number = _require_strict_int(data["number"], field="number")
priority_raw = data.get("priority") or 0
if isinstance(priority_raw, bool) or not isinstance(priority_raw, (int, float)):
# Allow numeric strings only for priority? Keep strict for bools.
if isinstance(priority_raw, str) and priority_raw.strip().lstrip("-").isdigit():
priority = int(priority_raw)
else:
raise ValueError(
f"priority must be numeric (booleans rejected; "
f"got {type(priority_raw).__name__})"
)
else:
priority = int(priority_raw)
return WorkCandidate(
kind=str(data.get("kind") or "issue"),
number=int(data["number"]),
number=number,
state=str(data.get("state") or "open"),
labels=tuple(data.get("labels") or ()),
title=str(data.get("title") or ""),
priority=int(data.get("priority") or 0),
priority=priority,
head_sha=data.get("head_sha"),
request_changes_current_head=bool(data.get("request_changes_current_head")),
approval_on_current_head=bool(data.get("approval_on_current_head")),
+1
View File
@@ -77,6 +77,7 @@ MUTATION_TASKS = frozenset({
"create_issue",
"comment_issue",
"close_issue",
"edit_issue",
"mark_issue",
"lock_issue",
"set_issue_labels",
+112 -13
View File
@@ -302,6 +302,7 @@ class ControlPlaneDB:
conn.executescript(_SCHEMA_SQL)
self._migrate_incident_links_null_scope(conn)
self._migrate_lease_lifecycle_columns(conn)
self._migrate_session_ownership_columns(conn)
conn.execute(
"INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)",
("schema_version", str(SCHEMA_VERSION)),
@@ -492,32 +493,62 @@ class ControlPlaneDB:
namespace: str | None = None,
pid: int | None = None,
status: str = "active",
controller_instance_id: str | None = None,
) -> dict[str, Any]:
"""Register/refresh a session row.
*controller_instance_id* (#765) is the stable identity of the
controller that owns this session. Session ids are regenerated per
invocation, so they cannot express "my own in-progress task"; the
controller instance can. It is never overwritten with ``None``, so a
heartbeat from a caller that does not supply one cannot erase
ownership.
"""
now = _ts()
instance = (controller_instance_id or "").strip() or None
with self._tx() as conn:
existing = conn.execute(
"SELECT session_id FROM sessions WHERE session_id = ?",
(session_id,),
).fetchone()
if existing:
conn.execute(
"""
UPDATE sessions
SET role = ?, profile = ?, namespace = ?, pid = ?,
last_heartbeat_at = ?, status = ?
WHERE session_id = ?
""",
(role, profile, namespace, pid, now, status, session_id),
)
if instance is None:
conn.execute(
"""
UPDATE sessions
SET role = ?, profile = ?, namespace = ?, pid = ?,
last_heartbeat_at = ?, status = ?
WHERE session_id = ?
""",
(role, profile, namespace, pid, now, status, session_id),
)
else:
conn.execute(
"""
UPDATE sessions
SET role = ?, profile = ?, namespace = ?, pid = ?,
last_heartbeat_at = ?, status = ?,
controller_instance_id = ?
WHERE session_id = ?
""",
(
role, profile, namespace, pid, now, status,
instance, session_id,
),
)
else:
conn.execute(
"""
INSERT INTO sessions(
session_id, role, profile, namespace, pid,
started_at, last_heartbeat_at, status
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
started_at, last_heartbeat_at, status,
controller_instance_id
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(session_id, role, profile, namespace, pid, now, now, status),
(
session_id, role, profile, namespace, pid, now, now,
status, instance,
),
)
row = conn.execute(
"SELECT * FROM sessions WHERE session_id = ?",
@@ -1215,6 +1246,73 @@ class ControlPlaneDB:
if name not in cols:
conn.execute(f"ALTER TABLE leases ADD COLUMN {name} {decl}")
_SESSION_OWNERSHIP_COLUMNS: tuple[tuple[str, str], ...] = (
("controller_instance_id", "TEXT"),
)
def _migrate_session_ownership_columns(self, conn: sqlite3.Connection) -> None:
"""Add the stable controller identity to sessions (#765).
Pre-existing rows migrate with ``NULL``. A NULL instance is treated as
*unknown ownership* by the allocator and is never silently adopted.
"""
cols = {
row[1]
for row in conn.execute("PRAGMA table_info(sessions)").fetchall()
}
if not cols:
return
for name, decl in self._SESSION_OWNERSHIP_COLUMNS:
if name not in cols:
conn.execute(f"ALTER TABLE sessions ADD COLUMN {name} {decl}")
def list_active_claims(
self,
*,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
role: str | None = None,
limit: int = 500,
) -> dict[tuple[str, int], dict[str, Any]]:
"""Map ``(work_kind, work_number)`` to its live claim (#765).
Only ``active`` leases count as claims; released/expired rows never
withhold work. Callers compare the returned ``controller_instance_id``
against their own to decide own-task vs foreign-task.
"""
claims: dict[tuple[str, int], dict[str, Any]] = {}
for row in self.list_leases(
remote=remote,
org=org,
repo=repo,
role=role,
statuses=("active",),
limit=limit,
):
kind = str(row.get("work_kind") or "").strip().lower()
number = row.get("work_number")
if not kind or number is None:
continue
key = (kind, int(number))
claim = {
"lease_id": row.get("lease_id"),
"session_id": row.get("session_id"),
"controller_instance_id": row.get("session_controller_instance_id"),
"role": row.get("role"),
"profile": row.get("session_profile"),
"expires_at": row.get("expires_at"),
"work_kind": kind,
"work_number": int(number),
}
# Keep the longest-lived claim when duplicates exist.
previous = claims.get(key)
if previous is None or str(claim["expires_at"] or "") > str(
previous["expires_at"] or ""
):
claims[key] = claim
return claims
def _lease_columns(self, conn: sqlite3.Connection) -> set[str]:
return {
row[1]
@@ -1256,7 +1354,8 @@ class ControlPlaneDB:
w.number AS work_number, w.state AS work_state,
w.current_head_sha AS work_head_sha,
s.pid AS session_pid, s.profile AS session_profile,
s.status AS session_status
s.status AS session_status,
s.controller_instance_id AS session_controller_instance_id
FROM leases l
JOIN work_items w ON w.work_item_id = l.work_item_id
LEFT JOIN sessions s ON s.session_id = l.session_id
+42
View File
@@ -131,6 +131,44 @@ Suggested lifecycle:
The helper module `issue_workflow_labels.py` is the source of truth for the
canonical label specs and status transition replacement behavior.
## Terminal PR transitions retire `status:pr-open` (#780)
`status:pr-open` states that a linked PR is *currently open*. The moment that
stops being true the label must go, whatever ended the PR:
| Terminal reason | Raised by |
|---|---|
| `merged` | `gitea_merge_pr` |
| `closed_without_merge` | `gitea_edit_pr` closing the PR |
| `superseded` | `gitea_reconcile_superseded_by_merged_pr` |
| `already_landed` | `gitea_reconcile_already_landed_pr` |
| `controller_closure` | `gitea_close_issue` |
| `abandoned` | abandonment handling |
| `retry_recovery` | `gitea_cleanup_terminal_pr_labels` after a partial failure |
All of these route through one rule in `terminal_pr_label_cleanup.py`, so the
paths cannot drift apart. The rule guarantees:
- only `status:pr-open` is removed — every other label is preserved verbatim;
- an empty resulting label set is valid (it was the issue's only label);
- an issue that no longer carries the label is a no-op, so retries are safe;
- the result is confirmed by a read-after-write re-read, not assumed.
Controller closure runs the cleanup **before** changing issue state and fails
closed if it cannot be completed and verified — closing first would bake in the
stale label with no later step to catch it. Post-merge cleanup never blocks the
merge: the transition already happened, so failures are reported with a
`safe_next_action` instead.
Use `gitea_assess_terminal_label_hygiene` as terminal validation before
declaring a transition or cleanup batch complete. It enumerates issues plus the
live open PRs and reports any issue still carrying `status:pr-open` without an
open PR to justify it. Issues with a genuinely open PR are exempt, not
residual.
Recovery from a partial failure is `gitea_cleanup_terminal_pr_labels` with
`terminal_reason='retry_recovery'`.
## Discussion Issues
Discussion issues must be labeled `type:discussion`.
@@ -157,6 +195,10 @@ If a discussion produces implementation work, either:
be applied to the locked issue, then applies it after the PR is created.
- `gitea_set_issue_labels` accepts an explicit `worktree_path` so author
sessions can satisfy the branches-only mutation guard while changing labels.
- `gitea_cleanup_terminal_pr_labels` retires `status:pr-open` after a terminal
PR transition; it is idempotent, so it is also the retry/recovery path.
- `gitea_assess_terminal_label_hygiene` is the read-only terminal validation
for residual `status:pr-open`.
## Existing Non-Workflow Labels
+3 -1
View File
@@ -706,7 +706,9 @@ do **not** improvise shell wrappers or fall back to direct API / temp scripts.
`fix/...` / `docs/...`); `cd` into that worktree; implement narrowly; add or
update tests if behavior changes; run the full suite; commit with an
issue-linked message; open a PR to `master`; move the issue to
`status:pr-open`. **Do not** review or merge your own PR. Include an
`status:pr-open` (every terminal transition later retires that label
automatically — see [`label-taxonomy.md`](label-taxonomy.md)). **Do not**
review or merge your own PR. Include an
`LLM Handoff Metadata` block (with `LLM-Agent-SHA`) in the PR body — see
[`llm-agent-sha.md`](llm-agent-sha.md).
- **Prompt:** `Use an author profile to implement issue #N and open a PR to
+169
View File
@@ -0,0 +1,169 @@
# Registered MCP tool inventory
This is the canonical list of tools the Gitea-Tools MCP server registers. It
exists because documentation and the registered inventory drifted: the workflow
documented a `gitea_edit_issue` tool that no namespace had ever registered, so a
mutation could be planned against a tool that did not exist and only fail at
execution time (#781).
## The rule
**Documentation must never name a tool an actor cannot reach.**
Two guards enforce it, both in `tests/test_issue_781_edit_issue_tool.py`:
1. The list below must equal the registered tool set exactly — sorted, no
duplicates, nothing missing in either direction. Adding a tool without
documenting it fails, and documenting a tool without registering it fails.
2. Every backticked `gitea_*` / `mcp_*` identifier in `skills/**/*.md` must be a
registered tool. Module and script names that share the prefix are listed
explicitly in `mcp_tool_inventory.NON_TOOL_IDENTIFIERS` rather than being
waved through by a looser pattern.
## Updating this file
When you add or remove an `@mcp.tool()`, regenerate the block below:
```bash
PYTEST_CURRENT_TEST=1 venv/bin/python -c "
import mcp_server, mcp_tool_inventory
print(mcp_tool_inventory.render_inventory_block(
mcp_server.mcp._tool_manager._tools))
"
```
Replace everything between the markers with that output. Do not hand-edit
individual entries — the generator and the guard share one ordering rule.
## Registered tools
Namespaces (`gitea-tools`, `gitea-reviewer`, `gitea-merger`, `gitea-reconciler`)
register the same tool set; what differs per namespace is the execution profile
that gates each call, not which tools exist.
<!-- BEGIN REGISTERED TOOL INVENTORY -->
- `gitea_abandon_workflow_lease`
- `gitea_acquire_conflict_fix_lease`
- `gitea_acquire_merger_pr_lease`
- `gitea_acquire_reviewer_pr_lease`
- `gitea_activate_profile`
- `gitea_adopt_merger_pr_lease`
- `gitea_adopt_workflow_lease`
- `gitea_allocate_next_work`
- `gitea_assess_already_landed_reconciliation`
- `gitea_assess_conflict_fix_classification`
- `gitea_assess_conflict_fix_push`
- `gitea_assess_gitea_operation_path`
- `gitea_assess_master_parity`
- `gitea_assess_mcp_namespace_health`
- `gitea_assess_pr_sync_status`
- `gitea_assess_review_merge_state_machine`
- `gitea_assess_reviewer_pr_lease`
- `gitea_assess_terminal_label_hygiene`
- `gitea_assess_work_issue_duplicate`
- `gitea_assess_worktree_cleanup_integrity`
- `gitea_audit_config`
- `gitea_audit_stable_branch_contamination`
- `gitea_audit_worktree_cleanup`
- `gitea_authorize_reconciliation_cleanup_phase`
- `gitea_authorize_review_correction`
- `gitea_capability_stop_terminal_report`
- `gitea_capture_branches_worktree_snapshot`
- `gitea_check_pr_eligibility`
- `gitea_cleanup_merged_pr_branch`
- `gitea_cleanup_obsolete_reviewer_comment_lease`
- `gitea_cleanup_post_merge_moot_lease`
- `gitea_cleanup_stale_claims`
- `gitea_cleanup_stale_review_decision_lock`
- `gitea_cleanup_terminal_pr_labels`
- `gitea_close_issue`
- `gitea_commit_files`
- `gitea_consume_irrecoverable_decision_lock_provenance`
- `gitea_create_issue`
- `gitea_create_issue_comment`
- `gitea_create_label`
- `gitea_create_pr`
- `gitea_delete_branch`
- `gitea_diagnose_review_decision_lock`
- `gitea_diagnose_reviewer_pr_lease_handoff`
- `gitea_diagnose_terminal`
- `gitea_dry_run_pr_review`
- `gitea_edit_issue`
- `gitea_edit_pr`
- `gitea_expire_workflow_leases`
- `gitea_get_authenticated_user`
- `gitea_get_current_user`
- `gitea_get_file`
- `gitea_get_pr_review_feedback`
- `gitea_get_profile`
- `gitea_get_runtime_context`
- `gitea_get_shell_health`
- `gitea_heartbeat_reviewer_pr_lease`
- `gitea_inspect_workflow_lease`
- `gitea_issue_irrecoverable_provenance_authorization`
- `gitea_list_issue_comments`
- `gitea_list_issues`
- `gitea_list_labels`
- `gitea_list_profiles`
- `gitea_list_prs`
- `gitea_list_workflow_leases`
- `gitea_load_review_workflow`
- `gitea_lock_issue`
- `gitea_mark_final_review_decision`
- `gitea_mark_issue`
- `gitea_merge_pr`
- `gitea_mirror_refs`
- `gitea_observability_link_issue`
- `gitea_observability_list_projects`
- `gitea_observability_reconcile_incident`
- `gitea_post_heartbeat`
- `gitea_quarantine_contaminated_review`
- `gitea_reclaim_expired_workflow_lease`
- `gitea_reconcile_already_landed_pr`
- `gitea_reconcile_issue_claims`
- `gitea_reconcile_merged_cleanups`
- `gitea_reconcile_superseded_by_merged_pr`
- `gitea_record_irrecoverable_decision_lock_provenance`
- `gitea_record_pre_review_command`
- `gitea_record_shell_spawn_outcome`
- `gitea_record_stable_branch_push_attempt`
- `gitea_release_merger_pr_lease`
- `gitea_release_reviewer_pr_lease`
- `gitea_release_workflow_lease`
- `gitea_resolve_task_capability`
- `gitea_resume_review_draft`
- `gitea_review_pr`
- `gitea_route_task_session`
- `gitea_save_review_draft`
- `gitea_scan_already_landed_open_prs`
- `gitea_sentry_get_issue_events`
- `gitea_sentry_link_gitea_issue`
- `gitea_sentry_list_issues`
- `gitea_sentry_reconcile_issue`
- `gitea_sentry_watchdog`
- `gitea_set_issue_labels`
- `gitea_submit_pr_review`
- `gitea_update_pr_branch_by_merge`
- `gitea_validate_review_final_report`
- `gitea_view_issue`
- `gitea_view_pr`
- `gitea_whoami`
- `gitea_workflow_dashboard`
- `mcp_check_workflow_skill_preflight`
- `mcp_get_control_plane_guide`
- `mcp_get_skill_guide`
- `mcp_list_project_skills`
<!-- END REGISTERED TOOL INVENTORY -->
## Issue-content editing
`gitea_edit_issue` is the only path that changes an issue's title or body. It
PATCHes the issue endpoint, refuses a pull-request number, sends only the fields
the caller named, and proves the result by read-after-write — including that
state, labels, assignees, and milestone did not move.
`gitea_edit_pr` remains pull-request-only. The two paths never merge: a single
tool that accepted either kind would make the narrower capability reachable
through the wider one.
+271
View File
@@ -0,0 +1,271 @@
"""Authoritative rule for editing an issue's title and body (#781).
The workflow documented a ``gitea_edit_issue`` tool that was never registered,
so an authorized body correction on an issue had no sanctioned path at all: the
only edit tool, ``gitea_edit_pr``, PATCHes the pull-request endpoint and cannot
target an issue. This module is the rule that path is built on, kept separate
from the pull-request edit path by construction.
- :func:`validate_edit_request` rejects structurally invalid requests before any
credential, network, or profile work happens. A request that names no field,
or names one with the wrong type, is a pure input error.
- :func:`assess_issue_target` refuses a pull request. Gitea serves pull requests
from the same ``/issues/{n}`` collection, so without this check the issue edit
path would quietly become a second, ungated PR edit path.
- :func:`plan_issue_edit` decides the exact PATCH payload from the pre-image. It
only ever sends fields the caller named, and it reports a request that would
change nothing as an explicit no-op rather than a silent success.
- :func:`verify_issue_edit` is the read-after-write check. It proves the applied
title/body match what was requested *and* that every field the caller did not
name — state, labels, assignees, milestone — is unchanged.
This module performs no I/O — callers own the Gitea API calls.
"""
from __future__ import annotations
from typing import Any, Mapping
import issue_workflow_labels
#: Fields this tool is allowed to change. Anything else must be untouched.
EDITABLE_FIELDS: tuple[str, ...] = ("title", "body")
#: Fields the caller never names and which must survive an edit verbatim.
PRESERVED_FIELDS: tuple[str, ...] = ("state", "labels", "assignees", "milestone")
def validate_edit_request(
title: str | None = None,
body: str | None = None,
) -> dict[str, str]:
"""Return the requested field map, failing closed on an invalid request.
Raises ``ValueError`` when no field is named, when a named field is not a
string, or when a title is blank. An empty *body* is legitimate — clearing
an issue description is a real edit — but an empty title is not, because
Gitea has no issue without one.
"""
requested: dict[str, str] = {}
if title is not None:
if not isinstance(title, str):
raise ValueError(
f"Invalid title type {type(title).__name__}: title must be a "
"string (fail closed)."
)
if not title.strip():
raise ValueError(
"Invalid title: an issue title cannot be blank. Pass the exact "
"replacement title, or omit title= to leave it unchanged "
"(fail closed)."
)
requested["title"] = title
if body is not None:
if not isinstance(body, str):
raise ValueError(
f"Invalid body type {type(body).__name__}: body must be a "
"string (fail closed)."
)
requested["body"] = body
if not requested:
raise ValueError(
"At least one field to edit (title, body) must be provided. "
"gitea_edit_issue never edits state, labels, assignees, or "
"milestone (fail closed)."
)
return requested
def assess_issue_target(
issue: Mapping[str, Any],
*,
issue_number: int,
) -> dict[str, Any]:
"""Confirm the fetched object is an issue and not a pull request.
Gitea serves pull requests from ``/issues/{n}`` as well, so a PR number
reaches this path unchallenged. Issue and pull-request edits stay separate
capabilities, so a PR target is refused here rather than silently PATCHed.
"""
is_pull_request = bool(issue.get("pull_request"))
return {
"is_issue": not is_pull_request,
"is_pull_request": is_pull_request,
"reasons": (
[
f"#{issue_number} is a pull request, not an issue; "
"gitea_edit_issue never edits pull requests"
]
if is_pull_request
else []
),
"safe_next_action": (
f"Use gitea_edit_pr for pull request #{issue_number}."
if is_pull_request
else ""
),
}
def preserved_snapshot(issue: Mapping[str, Any]) -> dict[str, Any]:
"""Capture the fields an edit must leave alone, in a comparable shape."""
return {
"state": issue.get("state"),
"labels": issue_workflow_labels.label_names(issue),
"assignees": _assignee_names(issue),
"milestone": _milestone_key(issue),
}
def _assignee_names(issue: Mapping[str, Any]) -> list[str]:
names: list[str] = []
for entry in issue.get("assignees") or []:
if isinstance(entry, Mapping):
login = entry.get("login") or entry.get("username")
else:
login = entry
if login:
names.append(str(login))
return names
def _milestone_key(issue: Mapping[str, Any]) -> str | None:
milestone = issue.get("milestone")
if not milestone:
return None
if isinstance(milestone, Mapping):
key = milestone.get("title") or milestone.get("id")
return None if key is None else str(key)
return str(milestone)
def plan_issue_edit(
current: Mapping[str, Any],
*,
title: str | None = None,
body: str | None = None,
issue_number: int | None = None,
) -> dict[str, Any]:
"""Plan the PATCH payload for an issue edit against its pre-image.
Only fields the caller named are ever put in the payload, so unspecified
fields cannot be overwritten with a stale read. A request whose named fields
already hold the requested values is reported as a no-op with an actionable
reason instead of being sent and reported as a success.
"""
requested = validate_edit_request(title=title, body=body)
number = issue_number if issue_number is not None else current.get("number")
changes: dict[str, dict[str, Any]] = {}
unchanged: list[str] = []
for field, value in requested.items():
before = current.get(field)
if field == "body":
before = before or ""
if before == value:
unchanged.append(field)
else:
changes[field] = {"before": before, "after": value}
no_op = not changes
payload = {field: requested[field] for field in changes}
return {
"issue_number": number,
"requested_fields": sorted(requested),
"requested": dict(requested),
"payload": payload,
"changes": changes,
"unchanged_fields": sorted(unchanged),
"no_op": no_op,
"preserved_before": preserved_snapshot(current),
"reasons": (
[
"requested "
+ ", ".join(sorted(unchanged))
+ " already match the issue's current content; no edit was sent"
]
if no_op
else []
),
"safe_next_action": (
(
f"Re-read issue #{number} and call gitea_edit_issue only with "
"content that differs, or drop the call if the issue is already "
"correct."
)
if no_op
else ""
),
}
def verify_issue_edit(
observed: Mapping[str, Any],
*,
plan: Mapping[str, Any],
) -> dict[str, Any]:
"""Read-after-write proof for an applied issue edit.
Fails closed on two distinct defects: an edited field whose stored value is
not what was requested, and an untouched field that moved anyway.
"""
requested = dict(plan.get("requested") or {})
number = plan.get("issue_number")
applied: dict[str, Any] = {}
mismatches: list[dict[str, Any]] = []
for field, expected in requested.items():
actual = observed.get(field)
if field == "body":
actual = actual or ""
applied[field] = actual
if actual != expected:
mismatches.append(
{"field": field, "expected": expected, "observed": actual}
)
before = dict(plan.get("preserved_before") or {})
after = preserved_snapshot(observed)
preserved_changed: list[dict[str, Any]] = [
{"field": field, "before": before.get(field), "after": after.get(field)}
for field in PRESERVED_FIELDS
if before.get(field) != after.get(field)
]
reasons: list[str] = []
for entry in mismatches:
reasons.append(
f"{entry['field']} was not applied: requested "
f"{entry['expected']!r} but the issue stores {entry['observed']!r}"
)
for entry in preserved_changed:
reasons.append(
f"{entry['field']} changed during the edit: {entry['before']!r} "
f"became {entry['after']!r}; gitea_edit_issue must leave it alone"
)
verified = not reasons
return {
"verified": verified,
"applied": applied,
"mismatches": mismatches,
"preserved_before": before,
"preserved_after": after,
"preserved_changed": preserved_changed,
"preserved_intact": not preserved_changed,
"reasons": reasons,
"safe_next_action": (
""
if verified
else (
f"Re-read issue #{number} with gitea_view_issue and reconcile it "
"before treating the edit as applied. Do not retry blindly — the "
"stored content does not match what was requested."
)
),
}
+105
View File
@@ -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,
@@ -728,6 +732,65 @@ def _rule_reviewer_stale_head_proof(report_text: str) -> list[dict[str, str]]:
)
_MUTATION_ACCOUNTING_PATTERNS = {
"local_failed_attempts": re.compile(
r"local\s+failed\s+attempts\s*:\s*(\d+)", re.IGNORECASE
),
"blocked_api_attempts": re.compile(
r"blocked\s+api\s+attempts\s*:\s*(\d+)", re.IGNORECASE
),
"successful_server_mutations": re.compile(
r"successful\s+server(?:[-\s]side)?\s+mutations\s*:\s*(\d+)", re.IGNORECASE
),
}
_READBACK_VERIFIED_PATTERN = re.compile(
r"read[-\s]?after[-\s]?write\s+verified\s*:\s*(yes|true)", re.IGNORECASE
)
def _rule_shared_mutation_budget_accounting(
report_text: str,
*,
mutation_attempt_ledger: list[dict] | None = None,
) -> list[dict[str, str]]:
"""#617: mutation budget counts server-side changes only.
No-op unless the session supplies an attempt ledger. When it does, the
report's three attempt categories must match the ledger exactly, so a
pre-API validator rejection can never be reported as a Gitea mutation and
a real mutation can never be hidden.
"""
if mutation_attempt_ledger is None:
return []
from mutation_budget_classifier import assess_final_report_mutation_accounting
text = report_text or ""
claimed: dict[str, Any] = {}
for field, pattern in _MUTATION_ACCOUNTING_PATTERNS.items():
match = pattern.search(text)
if match:
claimed[field] = int(match.group(1))
if _READBACK_VERIFIED_PATTERN.search(text):
claimed["readback_verified"] = True
result = assess_final_report_mutation_accounting(claimed, mutation_attempt_ledger)
if result.get("valid"):
return []
return _findings_from_reasons(
"shared.mutation_budget_accounting",
result.get("reasons") or [],
field="Mutation accounting",
severity="block",
safe_next_action=(
"report 'Local failed attempts:', 'Blocked API attempts:', and "
"'Successful server-side mutations:' with counts matching the "
"attempt ledger; pre-API rejections are not Gitea mutations"
),
)
def _rule_conflict_fix_classification_proof(report_text: str) -> list[dict[str, str]]:
from conflict_fix_classification import (
assess_conflict_fix_classification_final_report,
@@ -1564,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,
@@ -1584,13 +1662,24 @@ _SHARED_CANONICAL_COMMENT_RULES = (
_rule_shared_canonical_comment_post_claim,
)
_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,
*_SHARED_TWO_COMMENT_RULES,
*_SHARED_CANONICAL_COMMENT_RULES,
*_SHARED_MUTATION_BUDGET_RULES,
*_SHARED_ISSUE_LOCK_RULES,
_rule_reviewer_legacy_workspace_mutations,
_rule_reviewer_vague_mutations_none,
@@ -1620,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,
@@ -1631,11 +1721,13 @@ _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,
*_SHARED_TWO_COMMENT_RULES,
*_SHARED_CANONICAL_COMMENT_RULES,
*_SHARED_MUTATION_BUDGET_RULES,
*_SHARED_ISSUE_LOCK_RULES,
*_SHARED_CLEANUP_PROOF_RULES,
_rule_reconcile_stale_author_fields,
@@ -1650,20 +1742,24 @@ _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,
*_SHARED_TWO_COMMENT_RULES,
*_SHARED_CANONICAL_COMMENT_RULES,
*_SHARED_MUTATION_BUDGET_RULES,
*_SHARED_ISSUE_LOCK_RULES,
_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,
*_SHARED_TWO_COMMENT_RULES,
*_SHARED_CANONICAL_COMMENT_RULES,
*_SHARED_MUTATION_BUDGET_RULES,
*_SHARED_ISSUE_LOCK_RULES,
_rule_shared_issue_acceptance_gate,
_rule_reviewer_vague_mutations_none,
@@ -1672,28 +1768,34 @@ _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,
*_SHARED_TWO_COMMENT_RULES,
*_SHARED_CANONICAL_COMMENT_RULES,
*_SHARED_MUTATION_BUDGET_RULES,
*_SHARED_ISSUE_LOCK_RULES,
],
"inventory": [
*_SHARED_SELF_PROPAGATING_HANDOFF_RULES,
_rule_shared_controller_handoff,
_rule_shared_state_handoff_next_action,
_rule_shared_email_disclosure,
*_SHARED_TWO_COMMENT_RULES,
*_SHARED_CANONICAL_COMMENT_RULES,
*_SHARED_MUTATION_BUDGET_RULES,
*_SHARED_ISSUE_LOCK_RULES,
_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,
*_SHARED_TWO_COMMENT_RULES,
*_SHARED_CANONICAL_COMMENT_RULES,
*_SHARED_MUTATION_BUDGET_RULES,
*_SHARED_ISSUE_LOCK_RULES,
],
# Controller issue closure (#529): a closure report must not bury an
@@ -1701,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,
],
}
@@ -1766,6 +1869,7 @@ def assess_final_report_validator(
session_pr_opened: bool = False,
validation_session: dict | None = None,
reconciler_close_lock: dict | None = None,
mutation_attempt_ledger: list[dict] | None = None,
) -> dict[str, Any]:
"""Validate final-report text against task-specific proof rules (#327).
@@ -1829,6 +1933,7 @@ def assess_final_report_validator(
"session_pr_opened": session_pr_opened,
"validation_session": validation_session,
"reconciler_close_lock": reconciler_close_lock,
"mutation_attempt_ledger": mutation_attempt_ledger,
}
for rule in _RULES_BY_TASK.get(normalized_kind, ()):
+751 -44
View File
@@ -2122,6 +2122,8 @@ def _seed_session_context(
)
import issue_work_duplicate_gate # noqa: E402
import issue_workflow_labels # noqa: E402
import terminal_pr_label_cleanup # noqa: E402 # #780 status:pr-open terminal rule
import edit_issue # noqa: E402 # #781 issue title/body edit rule
import reviewer_pr_lease # noqa: E402
import post_merge_moot_lease_gate # noqa: E402 # #745 reconciler cleanup gate
import merger_lease_adoption # noqa: E402
@@ -2698,6 +2700,12 @@ def _put_issue_label_names(
auth,
{"labels": ids},
)
# Clearing every label is a legitimate full-set replacement (#780 AC4): the
# last label may be the one being retired. Gitea answers that PUT with an
# empty body, which api_request surfaces as None — treat it as the empty
# label set rather than a protocol violation.
if res is None and not names:
res = []
if not isinstance(res, list):
raise RuntimeError(
"Post-mutation label verification failed: expected a list of labels "
@@ -2824,7 +2832,191 @@ def release_in_progress_label(issue_numbers: list[int], remote: str, host: str |
return results
def cleanup_in_progress_for_pr(pr_payload: dict, remote: str, host: str | None, org: str | None, repo: str | None) -> dict:
def _clear_pr_open_label_for_issue(
*,
base: str,
auth: str,
issue_number: int,
terminal_reason: str,
resolve_label_ids,
audit_kwargs: dict,
) -> dict:
"""Apply the authoritative #780 cleanup rule to exactly one issue.
Plans through :mod:`terminal_pr_label_cleanup`, mutates via the verified
full-set replacement, then re-reads the issue so the caller reports a
read-after-write proof rather than an assumption.
"""
try:
current = _issue_label_names(base, auth, issue_number)
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"issue_number": issue_number,
"status": "error",
"performed": False,
"verified": False,
"reasons": [f"could not read current labels: {_redact(str(exc))}"],
}
plan = terminal_pr_label_cleanup.plan_pr_open_cleanup(
current, terminal_reason=terminal_reason
)
if not plan["cleanup_required"]:
# Idempotent by construction: nothing to remove, nothing to verify
# beyond the read we already did (#780 AC5).
return {
"issue_number": issue_number,
"status": "not present",
"performed": False,
"verified": True,
"idempotent_noop": True,
"labels_after": plan["labels_after"],
"reasons": [],
}
try:
# Resolved only once a mutation is actually needed, so the no-op path
# stays free of extra API traffic.
label_ids_by_name = resolve_label_ids()
with _audited(
"set_issue_labels",
issue_number=issue_number,
request_metadata={
"source": f"terminal_pr_label_cleanup:{plan['terminal_reason']}",
"removed": plan["removed"],
"preserved": plan["preserved"],
},
**audit_kwargs,
):
_put_issue_label_names(
base=base,
auth=auth,
issue_number=issue_number,
names=plan["labels_after"],
label_ids_by_name=label_ids_by_name,
)
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"issue_number": issue_number,
"status": "error",
"performed": False,
"verified": False,
"labels_before": plan["labels_before"],
"reasons": [f"label replacement failed: {_redact(str(exc))}"],
"safe_next_action": (
"Re-run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason='{terminal_pr_label_cleanup.RETRY_RECOVERY}' "
f"for issue #{issue_number}."
),
}
try:
observed = _issue_label_names(base, auth, issue_number)
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"issue_number": issue_number,
"status": "unverified",
"performed": True,
"verified": False,
"labels_before": plan["labels_before"],
"reasons": [f"read-after-write check failed: {_redact(str(exc))}"],
"safe_next_action": (
"Re-run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason='{terminal_pr_label_cleanup.RETRY_RECOVERY}' "
f"for issue #{issue_number} to confirm the final label set."
),
}
verification = terminal_pr_label_cleanup.verify_pr_open_cleanup(
observed, plan=plan
)
return {
"issue_number": issue_number,
"status": "removed" if verification["verified"] else "verification_failed",
"performed": True,
"verified": verification["verified"],
"labels_before": plan["labels_before"],
"labels_after": verification["observed_labels"],
"empty_label_set": verification["empty_label_set"],
"verification": verification,
"reasons": verification["reasons"],
"safe_next_action": verification["safe_next_action"],
}
def clear_pr_open_label(
issue_numbers: list[int],
remote: str,
host: str | None,
org: str | None,
repo: str | None,
*,
terminal_reason: str,
) -> dict:
"""Authoritative ``status:pr-open`` cleanup for terminal PR states (#780).
Every sanctioned terminal path routes here merge, close-without-merge,
supersession/abandonment, already-landed reconciliation, controller
closure, and retry/recovery so the rule cannot drift between them. Only
``status:pr-open`` is ever removed; all other labels are preserved, and an
empty resulting set is a valid outcome.
"""
reason = terminal_pr_label_cleanup.canonical_terminal_reason(terminal_reason)
numbers: list[int] = []
for raw in issue_numbers or []:
try:
num = int(raw)
except (TypeError, ValueError):
continue
if num not in numbers:
numbers.append(num)
if not numbers:
return terminal_pr_label_cleanup.summarize_cleanup_results(
[], terminal_reason=reason
)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
audit_kwargs = {"host": h, "remote": remote, "org": o, "repo": r}
# Memoized, lazily resolved label inventory: an issue that no longer
# carries the label needs no ids at all, so a clean repository costs one
# read per issue and nothing else.
cached: dict[str, dict[str, int]] = {}
def resolve_label_ids() -> dict[str, int]:
if "map" not in cached:
cached["map"] = _repo_label_id_map(base, auth)
return cached["map"]
results = [
_clear_pr_open_label_for_issue(
base=base,
auth=auth,
issue_number=num,
terminal_reason=reason,
resolve_label_ids=resolve_label_ids,
audit_kwargs=audit_kwargs,
)
for num in numbers
]
return terminal_pr_label_cleanup.summarize_cleanup_results(
results, terminal_reason=reason
)
def cleanup_in_progress_for_pr(
pr_payload: dict,
remote: str,
host: str | None,
org: str | None,
repo: str | None,
*,
terminal_reason: str = terminal_pr_label_cleanup.MERGED,
) -> dict:
body = pr_payload.get("body") or ""
title = pr_payload.get("title") or ""
branch = pr_payload.get("head", {}).get("ref") or ""
@@ -2833,10 +3025,38 @@ def cleanup_in_progress_for_pr(pr_payload: dict, remote: str, host: str | None,
issues = extract_linked_issue_numbers(text, branch)
if not issues:
return {"cleanup_status": "no linked issue found"}
return {
"cleanup_status": "no linked issue found",
"pr_open_label_cleanup": terminal_pr_label_cleanup.summarize_cleanup_results(
[], terminal_reason=terminal_reason
),
}
results = release_in_progress_label(issues, remote, host, org, repo)
return {"cleanup_status": results}
# #780: the PR has reached a terminal state, so status:pr-open must go too.
# Never raises — a failed cleanup is reported, never silently dropped, and
# never undoes the terminal transition that already happened.
try:
pr_open_cleanup = clear_pr_open_label(
issues, remote, host, org, repo, terminal_reason=terminal_reason
)
except Exception as exc: # noqa: BLE001 — redact before surfacing
pr_open_cleanup = {
"label": terminal_pr_label_cleanup.PR_OPEN_LABEL,
"clean": False,
"checked": issues,
"removed": [],
"already_absent": [],
"failed": issues,
"results": [],
"reasons": [f"terminal label cleanup failed: {_redact(str(exc))}"],
"safe_next_action": (
"Re-run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason='{terminal_pr_label_cleanup.RETRY_RECOVERY}' "
"for the linked issues."
),
}
return {"cleanup_status": results, "pr_open_label_cleanup": pr_open_cleanup}
# ── Helpers ───────────────────────────────────────────────────────────────────
@@ -8357,9 +8577,18 @@ def gitea_edit_pr(
data = api_request("PATCH", url, auth, payload)
cleanup_status = None
pr_open_label_cleanup = None
if state == "closed":
cleanup = cleanup_in_progress_for_pr(data, remote, host, org, repo)
cleanup = cleanup_in_progress_for_pr(
data,
remote,
host,
org,
repo,
terminal_reason=terminal_pr_label_cleanup.CLOSED_WITHOUT_MERGE,
)
cleanup_status = cleanup.get("cleanup_status")
pr_open_label_cleanup = cleanup.get("pr_open_label_cleanup")
if isinstance(cleanup_status, dict):
for issue_num, st in cleanup_status.items():
if st == "released":
@@ -8376,6 +8605,7 @@ def gitea_edit_pr(
"body": data.get("body", ""),
"state": data["state"],
"cleanup_status": cleanup_status,
"pr_open_label_cleanup": pr_open_label_cleanup,
}
return _with_optional_url(result, data.get("html_url"))
@@ -9111,12 +9341,33 @@ def gitea_merge_pr(
result["cleanup_status"] = "skipped (merge read-back failed)"
else:
try:
cleanup = cleanup_in_progress_for_pr(merged, remote, host, org, repo)
cleanup = cleanup_in_progress_for_pr(
merged,
remote,
host,
org,
repo,
terminal_reason=terminal_pr_label_cleanup.MERGED,
)
result["cleanup_status"] = cleanup.get("cleanup_status")
result["pr_open_label_cleanup"] = cleanup.get("pr_open_label_cleanup")
except Exception as cleanup_exc: # noqa: BLE001 — redact before surfacing
result["cleanup_status"] = (
f"skipped (cleanup error: {_redact(str(cleanup_exc))})"
)
result["pr_open_label_cleanup"] = {
"label": terminal_pr_label_cleanup.PR_OPEN_LABEL,
"clean": False,
"reasons": [
f"cleanup error: {_redact(str(cleanup_exc))}"
],
"safe_next_action": (
"Re-run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason="
f"'{terminal_pr_label_cleanup.RETRY_RECOVERY}' for the "
"linked issues."
),
}
except Exception as exc: # noqa: BLE001 — redact before surfacing
reasons.append(f"merge failed: {_redact(str(exc))}")
return result
@@ -11104,6 +11355,20 @@ def gitea_reconcile_already_landed_pr(
result["issue_closed"] = True
result["performed"] = True
# #780: once the PR is closed as already-landed, the linked issue is no
# longer awaiting an open PR — retire status:pr-open even when the issue
# itself stays open. Reported, never fatal: the reconciliation already
# happened and must not be reversed by a label failure.
if linked_issue and (result.get("pr_closed") or result.get("issue_closed")):
result["pr_open_label_cleanup"] = clear_pr_open_label(
[linked_issue],
remote,
host,
org,
repo,
terminal_reason=terminal_pr_label_cleanup.ALREADY_LANDED,
)
return result
@@ -11359,6 +11624,20 @@ def gitea_reconcile_superseded_by_merged_pr(
result["issue_closed"] = True
result["performed"] = True
# #780: supersession/abandonment is terminal for the target PR, so the
# target issue must not keep advertising an open PR.
if target_issue_number and (
result.get("pr_closed") or result.get("issue_closed")
):
result["pr_open_label_cleanup"] = clear_pr_open_label(
[target_issue_number],
remote,
host,
org,
repo,
terminal_reason=terminal_pr_label_cleanup.SUPERSEDED,
)
return result
@mcp.tool()
@@ -11417,6 +11696,33 @@ def gitea_close_issue(
"reasons": closure_gate["reasons"],
"safe_next_action": closure_gate["safe_next_action"],
}
# #780: controller closure is a terminal transition, so status:pr-open must
# be retired. Run it *before* the state change and fail closed if it cannot
# be completed — closing first would leave the exact stale-label state this
# gate exists to prevent, with no remaining step to catch it.
pr_open_cleanup = clear_pr_open_label(
[issue_number],
remote,
host,
org,
repo,
terminal_reason=terminal_pr_label_cleanup.CONTROLLER_CLOSURE,
)
if not pr_open_cleanup["clean"]:
return {
"success": False,
"blocked": True,
"performed": False,
"message": (
f"Issue #{issue_number} close blocked (#780): the terminal "
f"'{terminal_pr_label_cleanup.PR_OPEN_LABEL}' cleanup could not "
"be completed and verified."
),
"pr_open_label_cleanup": pr_open_cleanup,
"reasons": pr_open_cleanup["reasons"],
"safe_next_action": pr_open_cleanup["safe_next_action"],
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -11426,13 +11732,370 @@ def gitea_close_issue(
cleanup_result = release_in_progress_label([issue_number], remote, host, org, repo)
# Terminal validation (#780): re-read the closed issue and report any
# residual status:pr-open rather than assuming the earlier cleanup held.
try:
closed_labels = _issue_label_names(
repo_api_url(h, o, r), auth, issue_number
)
terminal_validation = terminal_pr_label_cleanup.detect_residual_pr_open(
[{"number": issue_number, "state": "closed", "labels": closed_labels}]
)
except Exception as exc: # noqa: BLE001 — redact before surfacing
terminal_validation = {
"label": terminal_pr_label_cleanup.PR_OPEN_LABEL,
"clean": False,
"checked_count": 0,
"residual_count": None,
"residual_issues": [],
"reasons": [
f"terminal label validation read-back failed: {_redact(str(exc))}"
],
"safe_next_action": (
"Re-check the issue labels with gitea_assess_terminal_label_hygiene."
),
}
return {
"success": True,
"message": f"Issue #{issue_number} closed.",
"cleanup_status": cleanup_result
"cleanup_status": cleanup_result,
"pr_open_label_cleanup": pr_open_cleanup,
"terminal_label_validation": terminal_validation,
}
@mcp.tool()
def gitea_edit_issue(
issue_number: int,
title: str | None = None,
body: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Edit an issue's title, body, or both through the sanctioned path (#781).
This is the only MCP path that changes issue content. It PATCHes the issue
endpoint and never the pull-request endpoint: ``gitea_edit_pr`` stays
pull-request-only, and a pull-request number passed here is refused rather
than edited, even though Gitea serves pull requests from ``/issues/{n}``.
Only the fields named in the call are sent, so labels, state, assignee, and
milestone cannot be overwritten by a stale read. The edit is proven by
read-after-write: the tool re-reads the issue and fails closed if the stored
title/body differ from what was requested, or if any unspecified field moved.
Args:
issue_number: The issue number to edit.
title: New issue title. Omit to leave the title unchanged.
body: New issue body. Omit to leave the body unchanged; pass ''
deliberately to clear it.
remote: Known instance 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Optional branches worktree to verify before mutation.
Returns:
dict with 'success', 'performed', the 'applied' content, and a
'read_after_write' proof carrying the preserved-field comparison. A
no-op, a pull-request target, a permission block, or a failed
verification all return 'performed'/'success' False with 'reasons' and
a 'safe_next_action'. Invalid input raises ValueError before any
credential or network work.
"""
# Validate BEFORE auth/profile/API setup: a malformed or empty request is a
# pure input error and must not depend on credentials or configuration.
edit_issue.validate_edit_request(title=title, body=body)
blocked = _profile_permission_block(
task_capability_map.required_permission("edit_issue"),
issue_number=issue_number,
remote=remote,
host=host,
org=org,
repo=repo,
org_explicit=org is not None,
repo_explicit=repo is not None,
)
if blocked:
return blocked
verify_preflight_purity(
remote,
worktree_path=worktree_path,
task="edit_issue",
org=org,
repo=repo,
)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
try:
current = api_request("GET", url, auth)
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"success": False,
"performed": False,
"issue_number": issue_number,
"reasons": [
f"could not read issue #{issue_number} before editing: "
f"{_redact(str(exc))}"
],
"safe_next_action": (
f"Confirm issue #{issue_number} exists in {o}/{r} with "
"gitea_view_issue, then retry the edit."
),
}
target = edit_issue.assess_issue_target(current, issue_number=issue_number)
if not target["is_issue"]:
return {
"success": False,
"performed": False,
"blocked": True,
"issue_number": issue_number,
"target": target,
"reasons": target["reasons"],
"safe_next_action": target["safe_next_action"],
}
plan = edit_issue.plan_issue_edit(
current, title=title, body=body, issue_number=issue_number
)
if plan["no_op"]:
return {
"success": False,
"performed": False,
"no_op": True,
"issue_number": issue_number,
"requested_fields": plan["requested_fields"],
"unchanged_fields": plan["unchanged_fields"],
"reasons": plan["reasons"],
"safe_next_action": plan["safe_next_action"],
}
try:
with _audited("edit_issue", host=h, remote=remote, org=o, repo=r,
issue_number=issue_number,
request_metadata={"fields": plan["requested_fields"]}):
api_request("PATCH", url, auth, plan["payload"])
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"success": False,
"performed": False,
"issue_number": issue_number,
"requested_fields": plan["requested_fields"],
"reasons": [f"issue edit failed: {_redact(str(exc))}"],
"safe_next_action": (
f"Re-read issue #{issue_number} with gitea_view_issue to confirm "
"nothing was applied, then retry the edit."
),
}
# Read-after-write (#781): the PATCH response is not proof. Re-read the
# issue so an unapplied edit, or a preserved field that moved, fails closed.
try:
observed = api_request("GET", url, auth)
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"success": False,
"performed": True,
"verified": False,
"issue_number": issue_number,
"requested_fields": plan["requested_fields"],
"reasons": [
f"issue edit read-back failed: {_redact(str(exc))}"
],
"safe_next_action": (
f"Re-read issue #{issue_number} with gitea_view_issue and confirm "
"the applied content before relying on this edit."
),
}
verification = edit_issue.verify_issue_edit(observed, plan=plan)
result = {
"success": verification["verified"],
"performed": True,
"verified": verification["verified"],
"issue_number": issue_number,
"requested_fields": plan["requested_fields"],
"changed_fields": sorted(plan["changes"]),
"applied": verification["applied"],
"read_after_write": verification,
"preserved_intact": verification["preserved_intact"],
"reasons": verification["reasons"],
"safe_next_action": verification["safe_next_action"],
}
return _with_optional_url(result, observed.get("html_url"))
@mcp.tool()
def gitea_cleanup_terminal_pr_labels(
issue_numbers: list[int],
terminal_reason: str = terminal_pr_label_cleanup.RETRY_RECOVERY,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Retire ``status:pr-open`` after a terminal PR transition (#780).
This is the sanctioned recovery path when a terminal transition completed
but its label cleanup did not for example a merge whose read-back failed,
or a workflow interrupted between closing a PR and updating its issue.
The rule is the same one every terminal path uses: remove only
``status:pr-open``, preserve every other label, allow an empty resulting
set, and verify by read-after-write. Calling it on an issue that no longer
carries the label is a harmless no-op, so retries are always safe.
Args:
issue_numbers: Issues whose terminal label state should be repaired.
terminal_reason: Why the PR is terminal one of merged,
closed_without_merge, superseded, already_landed,
controller_closure, abandoned, retry_recovery.
remote: Known instance 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Optional branches worktree to verify before mutation.
Returns:
dict with 'clean', per-issue 'results' carrying read-after-write proof,
'removed', 'already_absent', 'failed', and 'safe_next_action'.
"""
blocked = _profile_permission_block(
task_capability_map.required_permission("cleanup_terminal_pr_labels"),
remote=remote,
host=host,
org=org,
repo=repo,
org_explicit=org is not None,
repo_explicit=repo is not None,
)
if blocked:
return blocked
verify_preflight_purity(
remote,
worktree_path=worktree_path,
task="cleanup_terminal_pr_labels",
org=org,
repo=repo,
)
try:
reason = terminal_pr_label_cleanup.canonical_terminal_reason(terminal_reason)
except ValueError as exc:
return {
"success": False,
"clean": False,
"performed": False,
"reasons": [str(exc)],
"safe_next_action": (
"Re-call with terminal_reason set to one of: "
+ ", ".join(terminal_pr_label_cleanup.TERMINAL_REASONS)
),
}
summary = clear_pr_open_label(
issue_numbers, remote, host, org, repo, terminal_reason=reason
)
summary["success"] = summary["clean"]
summary["performed"] = bool(summary["removed"])
return summary
@mcp.tool()
def gitea_assess_terminal_label_hygiene(
state: str = "all",
limit: int = 500,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Read-only terminal validation for residual ``status:pr-open`` (#780).
Enumerates issues and the live open pull requests, then reports any issue
still carrying ``status:pr-open`` without an open PR to justify it. Use it
before declaring a terminal transition or a cleanup batch complete.
Args:
state: Issue state filter 'open', 'closed', or 'all' (default).
limit: Max issues to enumerate across all pages.
remote: Known instance 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
Returns:
dict with 'clean', 'residual_count', 'residual_issues',
'open_pr_count', and 'safe_next_action'.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"clean": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
try:
issues = api_get_all(
f"{base}/issues?state={state}&type=issues", auth, limit=limit
) or []
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"success": False,
"clean": False,
"reasons": [f"could not enumerate issues: {_redact(str(exc))}"],
"safe_next_action": (
"Retry once the Gitea API is reachable; an incomplete "
"enumeration cannot prove terminal label hygiene."
),
}
try:
open_pulls = _list_open_pulls(h, o, r, auth)
except Exception as exc: # noqa: BLE001 — redact before surfacing
return {
"success": False,
"clean": False,
"reasons": [f"could not enumerate open pull requests: {_redact(str(exc))}"],
"safe_next_action": (
"Retry once the Gitea API is reachable; without the open-PR "
"inventory a legitimate status:pr-open cannot be distinguished "
"from a stale one."
),
}
open_pr_issues: list[int] = []
for pull in open_pulls:
head_obj = pull.get("head") or {}
branch = head_obj.get("ref") if isinstance(head_obj, dict) else None
text = f"{pull.get('title') or ''}\n{pull.get('body') or ''}"
open_pr_issues.extend(extract_linked_issue_numbers(text, branch))
assessment = terminal_pr_label_cleanup.detect_residual_pr_open(
issues, open_pr_issue_numbers=open_pr_issues
)
assessment["success"] = True
assessment["issue_state_filter"] = state
assessment["open_pr_count"] = len(open_pulls)
return assessment
@mcp.tool()
def gitea_list_issues(
state: str = "open",
@@ -15509,6 +16172,12 @@ def gitea_assess_master_parity(
if parity["stale"] and enforced:
out["report"] = master_parity_gate.parity_report(parity)
return out
# #781: documented in the canonical review workflow as a tool reviewers call
# before workflow load, but the registration decorator had been lost, so the
# documented inventory named something no namespace could reach.
@mcp.tool()
def gitea_record_pre_review_command(
command: str,
cwd: str | None = None,
@@ -19202,7 +19871,9 @@ def gitea_allocate_next_work(
repo: str | None = None,
include_issues: bool = True,
include_prs: bool = True,
candidates_json: str | None = None,
candidates_json: Any = None,
exclude_issue_numbers: list[int] | None = None,
expected_candidate_set_fingerprint: str | None = None,
limit: int = 50,
) -> dict:
"""Controller-owned next-work allocator using the #613 control-plane DB (#600).
@@ -19216,10 +19887,18 @@ def gitea_allocate_next_work(
leases as the coordination source).
Outcomes include: ``assigned_work``, ``preview``, ``wait``,
``blocked_by_terminal_path``, ``no_safe_work``, ``role_ineligible``.
``blocked_by_terminal_path``, ``no_safe_work``, ``role_ineligible``,
``blocked_by_excluded_own_lease``, ``candidate_set_drift``.
*candidates_json* may inject a JSON list of candidate dicts (tests /
controller overrides). When omitted, open issues/PRs are loaded from Gitea.
*candidates_json* may inject a candidate list as an already-decoded list
(native MCP transport) or a JSON string (backward compatible). When
omitted, open issues/PRs are loaded from Gitea.
*exclude_issue_numbers* (#776): typed list of issue/PR numbers removed
before ranking. Omitted/empty retains prior behavior.
*expected_candidate_set_fingerprint* (#776 AC4): optional CAS pin from a
prior dry-run; apply fails closed on material candidate-set drift.
#612 remains downstream: raw monitoring incidents are never candidates.
"""
@@ -19273,15 +19952,12 @@ def gitea_allocate_next_work(
inv_reasons: list[str] = []
candidates: list[Any] = []
if candidates_json:
if candidates_json is not None and candidates_json != "":
try:
raw = json.loads(candidates_json)
if not isinstance(raw, list):
raise ValueError("candidates_json must be a JSON list")
for item in raw:
if not isinstance(item, dict):
continue
candidates.append(allocator_service.candidate_from_dict(item))
# #776 AC3: accept decoded list or JSON string; reject malformed.
candidates = allocator_service.normalize_candidates_payload(
candidates_json
)
except Exception as exc: # noqa: BLE001
return {
"success": False,
@@ -19324,23 +20000,40 @@ def gitea_allocate_next_work(
sid = (session_id or "").strip() or (
f"{profile_name or 'session'}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
)
result = allocator_service.allocate_next_work(
db,
session_id=sid,
role=role_in,
remote=remote if remote in REMOTES else remote,
org=o,
repo=r,
candidates=candidates,
apply=bool(apply),
profile_name=profile_name,
username=username,
)
try:
result = allocator_service.allocate_next_work(
db,
session_id=sid,
role=role_in,
remote=remote if remote in REMOTES else remote,
org=o,
repo=r,
candidates=candidates,
apply=bool(apply),
profile_name=profile_name,
username=username,
controller_instance_id=allocator_service.resolve_controller_instance_id(),
exclude_issue_numbers=exclude_issue_numbers,
expected_candidate_set_fingerprint=expected_candidate_set_fingerprint,
)
except ValueError as exc:
return {
"success": False,
"outcome": allocator_service.OUTCOME_NO_SAFE,
"apply": bool(apply),
"reasons": [
f"invalid allocator input: {_redact(str(exc))} (fail closed, #776)"
],
"assignment": None,
"substrate": "control_plane_db",
}
if inv_reasons:
result.setdefault("inventory_warnings", inv_reasons)
result["candidate_count"] = len(candidates)
result["inventory_source"] = (
"candidates_json" if candidates_json else "gitea_live"
"candidates_json"
if candidates_json is not None and candidates_json != ""
else "gitea_live"
)
result["inventory_complete"] = True
result["selection_policy"] = allocator_service.SELECTION_POLICY
@@ -19434,7 +20127,7 @@ def gitea_workflow_dashboard(
include_issues: bool = True,
include_prs: bool = True,
include_non_active_leases: bool = True,
candidates_json: str | None = None,
candidates_json: Any = None,
limit: int = 100,
) -> dict:
"""Read-only MCP workflow dashboard: queues, leases, blockers, next safe action (#605).
@@ -19443,9 +20136,10 @@ def gitea_workflow_dashboard(
still requires ``gitea_allocate_next_work``. Never presents blocked or
terminal-locked items as safe next work.
*candidates_json* may inject a JSON list of candidate dicts (tests /
offline fixtures). When omitted, open issues/PRs are loaded from Gitea.
Incomplete live inventory fails closed (no safe suggestions).
*candidates_json* may inject a candidate list as an already-decoded list
or JSON string (#776 transport parity). When omitted, open issues/PRs are
loaded from Gitea. Incomplete live inventory fails closed (no safe
suggestions).
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
@@ -19470,15 +20164,11 @@ def gitea_workflow_dashboard(
inv_reasons: list[str] = []
candidates: list[Any] = []
inventory_complete = True
if candidates_json:
if candidates_json is not None and candidates_json != "":
try:
raw = json.loads(candidates_json)
if not isinstance(raw, list):
raise ValueError("candidates_json must be a JSON list")
for item in raw:
if not isinstance(item, dict):
continue
candidates.append(allocator_service.candidate_from_dict(item))
candidates = allocator_service.normalize_candidates_payload(
candidates_json
)
except Exception as exc: # noqa: BLE001
return {
"success": False,
@@ -19502,6 +20192,7 @@ def gitea_workflow_dashboard(
leases: list[dict] = []
terminal_pr: int | None = None
terminal_lock: dict | None = None
claims: dict = {}
db, db_errs = _control_plane_db_or_error()
if db is None:
inv_reasons.extend(db_errs or ["control-plane DB unavailable"])
@@ -19540,6 +20231,18 @@ def gitea_workflow_dashboard(
inv_reasons.append(
f"terminal lock lookup failed: {_redact(str(exc))}"
)
try:
# #765: live claims so the dashboard never advertises another
# controller's active task as safe next work.
claims = db.list_active_claims(
remote=remote if remote in REMOTES else remote,
org=o,
repo=r,
)
except Exception as exc: # noqa: BLE001
inv_reasons.append(
f"active claim lookup failed: {_redact(str(exc))}"
)
snap = workflow_dashboard.build_workflow_dashboard(
candidates=candidates,
@@ -19551,10 +20254,14 @@ def gitea_workflow_dashboard(
terminal_lock=terminal_lock,
inventory_complete=inventory_complete,
inventory_reasons=inv_reasons,
claims=claims,
controller_instance_id=allocator_service.resolve_controller_instance_id(),
)
payload = snap.as_dict()
payload["inventory_source"] = (
"candidates_json" if candidates_json else "gitea_live"
"candidates_json"
if candidates_json is not None and candidates_json != ""
else "gitea_live"
)
payload["limit_applies_to"] = "lease_listing_only"
return payload
+195
View File
@@ -0,0 +1,195 @@
"""Documented-vs-registered MCP tool inventory guard (#781).
The workflow documentation named a ``gitea_edit_issue`` tool that no namespace
had ever registered. Nothing compared the two lists, so an actor could plan a
mutation against a tool that did not exist and only discover it at execution
time — after the work was already scoped around it.
This module is that comparison, in two directions:
- :func:`assess_inventory_drift` compares the canonical inventory documented in
``docs/mcp-tool-inventory.md`` against the tools actually registered on the
MCP server. Either list drifting fails the guard, so a new tool must be
documented and a removed tool must be undocumented in the same change.
- :func:`assess_doc_references` catches the original defect directly: any tool
name a workflow/skill document tells an actor to call must be registered.
Module and script names legitimately appear in the same prose (``gitea_auth``,
``offline_mcp_runner``), so :data:`NON_TOOL_IDENTIFIERS` names the known
non-tool identifiers explicitly rather than loosening the pattern.
This module performs no I/O — callers own reading the files.
"""
from __future__ import annotations
import re
from typing import Any, Iterable, Mapping
#: Canonical documented inventory, relative to the repository root.
INVENTORY_DOC_PATH = "docs/mcp-tool-inventory.md"
#: Delimiters around the generated inventory list in the doc.
INVENTORY_BEGIN_MARKER = "<!-- BEGIN REGISTERED TOOL INVENTORY -->"
INVENTORY_END_MARKER = "<!-- END REGISTERED TOOL INVENTORY -->"
#: Backticked identifiers that look like tool names but are modules/scripts.
#: Every entry is a real file in this repository, not an MCP tool.
NON_TOOL_IDENTIFIERS: frozenset[str] = frozenset(
{
"gitea_auth",
"gitea_config",
"gitea_mcp_server",
"mcp_server",
"offline_mcp_helper",
"offline_mcp_runner",
}
)
#: Prefixes that mark an identifier as a candidate MCP tool name.
TOOL_NAME_PREFIXES: tuple[str, ...] = ("gitea_", "mcp_")
_INVENTORY_ENTRY = re.compile(r"^-\s+`([A-Za-z_][A-Za-z0-9_]*)`")
_BACKTICKED = re.compile(r"`([A-Za-z_][A-Za-z0-9_]*)`")
def parse_documented_inventory(text: str) -> list[str]:
"""Return the tool names listed between the inventory markers.
Raises ``ValueError`` when the markers are missing or out of order, so a
mangled document fails the guard instead of silently documenting nothing.
"""
start = text.find(INVENTORY_BEGIN_MARKER)
end = text.find(INVENTORY_END_MARKER)
if start == -1 or end == -1 or end < start:
raise ValueError(
f"{INVENTORY_DOC_PATH} must contain "
f"'{INVENTORY_BEGIN_MARKER}' followed by "
f"'{INVENTORY_END_MARKER}' (fail closed)."
)
block = text[start + len(INVENTORY_BEGIN_MARKER) : end]
names: list[str] = []
for line in block.splitlines():
match = _INVENTORY_ENTRY.match(line.strip())
if match:
names.append(match.group(1))
return names
def looks_like_tool_name(identifier: str) -> bool:
"""Return whether a backticked identifier is a candidate tool name."""
if identifier in NON_TOOL_IDENTIFIERS:
return False
return identifier.startswith(TOOL_NAME_PREFIXES)
def extract_tool_references(text: str) -> set[str]:
"""Return candidate tool names a document tells an actor to call."""
return {
name
for name in _BACKTICKED.findall(text)
if looks_like_tool_name(name)
}
def assess_inventory_drift(
documented: Iterable[str],
registered: Iterable[str],
) -> dict[str, Any]:
"""Compare the documented inventory against the registered tool set."""
documented_list = list(documented)
documented_set = set(documented_list)
registered_set = set(registered)
duplicates = sorted(
{name for name in documented_list if documented_list.count(name) > 1}
)
documented_not_registered = sorted(documented_set - registered_set)
registered_not_documented = sorted(registered_set - documented_set)
unsorted = documented_list != sorted(documented_list)
reasons: list[str] = []
if documented_not_registered:
reasons.append(
"documented but not registered: "
+ ", ".join(documented_not_registered)
)
if registered_not_documented:
reasons.append(
"registered but not documented: "
+ ", ".join(registered_not_documented)
)
if duplicates:
reasons.append("listed more than once: " + ", ".join(duplicates))
if unsorted:
reasons.append("inventory entries are not in sorted order")
in_sync = not reasons
return {
"in_sync": in_sync,
"documented_count": len(documented_set),
"registered_count": len(registered_set),
"documented_not_registered": documented_not_registered,
"registered_not_documented": registered_not_documented,
"duplicates": duplicates,
"sorted": not unsorted,
"reasons": reasons,
"safe_next_action": (
""
if in_sync
else (
f"Update {INVENTORY_DOC_PATH} so the block between the "
"inventory markers lists exactly the registered tools, sorted, "
"one '- `tool_name`' entry per line."
)
),
}
def assess_doc_references(
references: Mapping[str, Iterable[str]],
registered: Iterable[str],
) -> dict[str, Any]:
"""Verify every tool a document names is actually registered.
*references* maps a document path to the candidate tool names it mentions.
"""
registered_set = set(registered)
unregistered: list[dict[str, Any]] = []
checked = 0
for path, names in sorted(references.items()):
for name in sorted(set(names)):
checked += 1
if name not in registered_set:
unregistered.append({"document": path, "tool": name})
clean = not unregistered
reasons = [
f"{entry['document']} documents '{entry['tool']}', "
"which no namespace registers"
for entry in unregistered
]
return {
"clean": clean,
"checked_count": checked,
"unregistered": unregistered,
"reasons": reasons,
"safe_next_action": (
""
if clean
else (
"Either register the named tool with @mcp.tool() or correct the "
"document. Documentation must never name a tool an actor cannot "
"reach. If the identifier is a module or script rather than a "
"tool, add it to mcp_tool_inventory.NON_TOOL_IDENTIFIERS."
)
),
}
def render_inventory_block(registered: Iterable[str]) -> str:
"""Render the marker-delimited inventory block for the documentation."""
lines = [INVENTORY_BEGIN_MARKER, ""]
lines.extend(f"- `{name}`" for name in sorted(set(registered)))
lines.extend(["", INVENTORY_END_MARKER])
return "\n".join(lines)
+286
View File
@@ -0,0 +1,286 @@
"""Mutation-budget classification for auto-mode attempts (#617).
Mutation budget must count only *server-side* Gitea state changes. A tool call
that fails closed before the Gitea API is reached changed nothing on the
server, so it must not consume the budget that protects against repeated real
mutations.
The classifier separates four outcome classes plus an explicit ambiguous class:
``local_validator_rejection``
A canonical-content validator (for example the ``[THREAD STATE LEDGER]`` or
``## Canonical Issue State`` blocks) rejected the payload before any API
call. No server-side state exists.
``capability_gate_rejection``
A profile/permission gate refused the operation before any API call.
``transport_failure_before_api``
The request never reached the Gitea API (transport/EOF/connection error).
``server_side_mutation``
The API succeeded and returned proof of durable state (comment id, review
id, merge commit, label result, or an issue/PR state change).
``ambiguous_requires_readback``
The API *was* reached but the result carries no usable proof either way.
This fails closed: the attempt is treated as budget-consuming until a
read-after-write check proves otherwise, so #617 never weakens the guard
that prevents repeated real mutations.
Only ``server_side_mutation`` consumes budget outright. Every attempt — failed
or not — is still recorded in the local attempt ledger so a final report can
show local failed attempts, blocked API attempts, and successful server-side
mutations separately.
"""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Any
LOCAL_VALIDATOR_REJECTION = "local_validator_rejection"
CAPABILITY_GATE_REJECTION = "capability_gate_rejection"
TRANSPORT_FAILURE_BEFORE_API = "transport_failure_before_api"
SERVER_SIDE_MUTATION = "server_side_mutation"
AMBIGUOUS_REQUIRES_READBACK = "ambiguous_requires_readback"
CLASSIFICATIONS = (
LOCAL_VALIDATOR_REJECTION,
CAPABILITY_GATE_REJECTION,
TRANSPORT_FAILURE_BEFORE_API,
SERVER_SIDE_MUTATION,
AMBIGUOUS_REQUIRES_READBACK,
)
#: Result fields that prove durable server-side state was created.
MUTATION_PROOF_FIELDS = (
"comment_id",
"review_id",
"merge_commit_sha",
"label_result",
"state_change",
"created_pr_number",
)
#: Classes that never consume server-side mutation budget.
PRE_API_CLASSIFICATIONS = (
LOCAL_VALIDATOR_REJECTION,
CAPABILITY_GATE_REJECTION,
TRANSPORT_FAILURE_BEFORE_API,
)
FINAL_REPORT_REQUIRED_FIELDS = (
"local_failed_attempts",
"blocked_api_attempts",
"successful_server_mutations",
)
def _clean(value: Any) -> str:
return (value or "").strip() if isinstance(value, str) else str(value or "").strip()
def _proof_fields_present(result: dict) -> list[str]:
"""Return the mutation-proof fields carrying a usable value."""
present: list[str] = []
for field in MUTATION_PROOF_FIELDS:
value = result.get(field)
if value is None or value is False:
continue
if isinstance(value, str) and not value.strip():
continue
present.append(field)
return present
def _decision(
classification: str,
*,
budget_consumed: bool,
requires_readback: bool,
reasons: list[str],
proof_fields: list[str],
api_called: bool | None,
) -> dict:
return {
"classification": classification,
"budget_consumed": budget_consumed,
"requires_readback": requires_readback,
"pre_api": classification in PRE_API_CLASSIFICATIONS,
"api_called": api_called,
"proof_fields": proof_fields,
"reasons": reasons,
}
def classify_mutation_attempt(result: dict | None) -> dict:
"""Classify one mutation attempt and decide whether it consumes budget.
``result`` is the raw dict a Gitea MCP tool returned. The caller does not
pre-interpret it: classification is driven by the explicit ``api_called``
signal plus the proof fields the tool reports.
"""
data = dict(result or {})
success = bool(data.get("success"))
proof_fields = _proof_fields_present(data)
api_called = data.get("api_called")
# An unambiguous success carrying durable proof is a real mutation however
# the attempt was labelled upstream.
if success and proof_fields:
return _decision(
SERVER_SIDE_MUTATION,
budget_consumed=True,
requires_readback=False,
reasons=[
"API reported success with durable proof field(s): "
+ ", ".join(proof_fields)
],
proof_fields=proof_fields,
api_called=True,
)
if api_called is False:
# Nothing reached the server; pick the precise pre-API class.
if data.get("transport_error") or data.get("transport_failed"):
return _decision(
TRANSPORT_FAILURE_BEFORE_API,
budget_consumed=False,
requires_readback=False,
reasons=["transport failed before the Gitea API was reached"],
proof_fields=[],
api_called=False,
)
if data.get("permission_report") or data.get("capability_blocked"):
return _decision(
CAPABILITY_GATE_REJECTION,
budget_consumed=False,
requires_readback=False,
reasons=["capability/permission gate refused before any API call"],
proof_fields=[],
api_called=False,
)
return _decision(
LOCAL_VALIDATOR_REJECTION,
budget_consumed=False,
requires_readback=False,
reasons=[
"local validator rejected the payload before any API call; "
"no server-side state was created"
],
proof_fields=[],
api_called=False,
)
if api_called is True:
if success:
reason = (
"API reported success but returned no durable proof field; "
"read-after-write verification required before counting budget"
)
else:
reason = (
"API was reached and the outcome carries no durable proof; "
"read-after-write verification required before counting budget"
)
return _decision(
AMBIGUOUS_REQUIRES_READBACK,
budget_consumed=True,
requires_readback=True,
reasons=[reason],
proof_fields=proof_fields,
api_called=True,
)
# ``api_called`` was not reported at all. Fail closed rather than assuming
# nothing happened.
return _decision(
AMBIGUOUS_REQUIRES_READBACK,
budget_consumed=True,
requires_readback=True,
reasons=[
"attempt did not report 'api_called'; cannot prove the request "
"stopped before the Gitea API, so the attempt fails closed"
],
proof_fields=proof_fields,
api_called=None,
)
def record_attempt(
ledger: list[dict] | None,
result: dict | None,
*,
operation: str = "",
timestamp: str | None = None,
) -> dict:
"""Append one classified attempt to the local ledger and return the entry.
Every attempt is recorded, including the ones that consume no budget: the
point of #617 is that failed local attempts stay visible without being
miscounted as Gitea mutations.
"""
entries = ledger if isinstance(ledger, list) else []
entry = {
"operation": _clean(operation),
"timestamp": _clean(timestamp) or datetime.now(timezone.utc).isoformat(),
**classify_mutation_attempt(result),
}
entries.append(entry)
return entry
def summarize_attempt_ledger(ledger: list[dict] | None) -> dict:
"""Summarize a ledger into the categories a final report must show."""
entries = [e for e in (ledger or []) if isinstance(e, dict)]
def _count(*classifications: str) -> int:
return sum(1 for e in entries if e.get("classification") in classifications)
return {
"total_attempts": len(entries),
"local_failed_attempts": _count(LOCAL_VALIDATOR_REJECTION),
"blocked_api_attempts": _count(
CAPABILITY_GATE_REJECTION, TRANSPORT_FAILURE_BEFORE_API
),
"successful_server_mutations": _count(SERVER_SIDE_MUTATION),
"ambiguous_attempts": _count(AMBIGUOUS_REQUIRES_READBACK),
"budget_consumed": sum(1 for e in entries if e.get("budget_consumed")),
"requires_readback": any(e.get("requires_readback") for e in entries),
"entries": entries,
}
def assess_final_report_mutation_accounting(
report: dict | None,
ledger: list[dict] | None,
) -> dict:
"""Fail closed when a report's mutation accounting contradicts the ledger."""
data = dict(report or {})
summary = summarize_attempt_ledger(ledger)
reasons: list[str] = []
for field in FINAL_REPORT_REQUIRED_FIELDS:
if field not in data:
reasons.append(f"final report omits required field '{field}'")
continue
claimed = data.get(field)
actual = summary[field]
if claimed != actual:
reasons.append(
f"final report claims {field}={claimed} but the attempt ledger "
f"shows {actual}"
)
if summary["requires_readback"] and not data.get("readback_verified"):
reasons.append(
"ledger contains an ambiguous attempt; final report must record "
"'readback_verified' proof before claiming mutation accounting"
)
return {
"valid": not reasons,
"reasons": reasons,
"ledger_summary": {k: v for k, v in summary.items() if k != "entries"},
}
+907
View File
@@ -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 = "<!-- sph:v1 -->"
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")
),
}
+8
View File
@@ -35,3 +35,11 @@ Install for Codex:
```
Preflight via MCP: `mcp_check_workflow_skill_preflight`.
## Tool inventory
Which tools actually exist is documented in
[`docs/mcp-tool-inventory.md`](../../docs/mcp-tool-inventory.md), which a test
holds equal to the registered set. Never plan a mutation against a tool that is
not listed there — that is the #781 failure mode, where a documented
`gitea_edit_issue` did not exist until execution time.
+20
View File
@@ -216,6 +216,15 @@ Helpers: `scripts/worktree-start`, `scripts/worktree-review`,
- Never place raw tokens in LLM/MCP config.
- Use `gitea_whoami` and `gitea_resolve_task_capability` before mutating.
## Tool inventory
[`docs/mcp-tool-inventory.md`](../../docs/mcp-tool-inventory.md) is the canonical
list of registered tools, held equal to the live registry by a test. A tool that
is not listed there does not exist — do not scope work around it (#781).
Issue content is edited with `gitea_edit_issue` (title/body only, read-after-write
verified). `gitea_edit_pr` is pull-request-only and never accepts an issue number.
## Controller Handoff
Every task must end with a section titled exactly `Controller Handoff`. Compact
@@ -224,6 +233,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/):
@@ -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.
@@ -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.
@@ -39,6 +39,7 @@ occurred).
- Git ref mutations:
- MCP/Gitea mutations:
- Reconciliation mutations:
- Terminal label cleanup:
- External-state mutations:
- Read-only diagnostics:
- Blockers:
@@ -50,4 +51,15 @@ occurred).
Identity format: `username / profile` (not personal email unless required — #305).
`git fetch` belongs under `Git ref mutations`, not read-only diagnostics (#297).
`git fetch` belongs under `Git ref mutations`, not read-only diagnostics (#297).
`Terminal label cleanup` (#780) reports the `pr_open_label_cleanup` record the
reconciliation tool returned — `clean` / `failed` / `not applicable (no linked
issue)`, with the labels removed and preserved. Reconciliation is a terminal
transition, so a non-`clean` record blocks any "reconciled" claim; recover with
`gitea_cleanup_terminal_pr_labels` (`terminal_reason='retry_recovery'`) and
confirm with `gitea_assess_terminal_label_hygiene`.
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.
@@ -99,6 +99,21 @@ Narrative final report and controller handoff must agree on eligibility class,
candidate/reviewed head SHA, mutation state, worktree usage, review decision,
terminal review mutation, merge result, and linked issue status.
### Terminal label state (#780)
A run that takes a PR to a terminal state — merged, closed without merge,
superseded, or reconciled as already landed — must report what happened to the
linked issue's `status:pr-open` label, quoting the `pr_open_label_cleanup`
record the terminal tool returned:
- Terminal label cleanup: `clean` / `failed` / `not applicable (no linked issue)`
- Labels removed and preserved per issue, with the read-after-write read-back
Never claim the transition is complete while that record is not `clean`. A
failed cleanup does not undo the merge; the safe next action is
`gitea_cleanup_terminal_pr_labels` with `terminal_reason='retry_recovery'`,
confirmed by `gitea_assess_terminal_label_hygiene`.
### Proof-backed claims (#395)
Proof-sensitive claims must cite explicit command/tool evidence in the report
@@ -116,4 +131,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.
`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.
@@ -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
<!-- sph:v1 -->
## Canonical Handoff
```text
REPOSITORY: <org>/<repo>
ISSUE: <number>
PR: <number or none>
WORKFLOW_STATE: <one of the workflow states below>
HEAD_SHA: <current head, or none before a branch exists>
BASE_BRANCH: <base branch>
BASE_OR_MERGE_SHA: <base SHA, or merge commit SHA after merge>
ACTING_ROLE: <author|reviewer|merger|controller|operator|reconciler>
ACTING_IDENTITY: <username (profile)>
COMPLETED_ACTIONS: <what this actor actually did>
VALIDATION_EVIDENCE: <commands run and their results>
MUTATION_LEDGER: <every durable mutation performed>
BLOCKERS: <active blockers, or none>
NEXT_ACTOR: <role authorized by WORKFLOW_STATE, or none when complete>
NEXT_ACTION: <exact next action, or none when complete>
PROHIBITED_ACTIONS: <what the next actor must not do>
NEXT_PROMPT: <complete ready-to-run prompt, or none when complete>
WORKFLOW_FAILURE_ISSUES: <durable issue refs for tooling defects, or none>
LAST_UPDATED: <UTC timestamp>
```
```
## 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.
@@ -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.
`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.
+16
View File
@@ -36,6 +36,20 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.issue.comment",
"role": "author",
},
# #781: editing an issue title/body is issue authoring, the same authority
# every other non-create/non-close issue mutation gates on. Deliberately not
# a new operation name: introducing one would silently strip the capability
# from every already-configured author profile.
"edit_issue": {
"permission": "gitea.issue.comment",
"role": "author",
},
# #780: retire status:pr-open after a terminal PR transition. Same label
# authority as set_issue_labels — it is a strictly narrower operation.
"cleanup_terminal_pr_labels": {
"permission": "gitea.issue.comment",
"role": "author",
},
"create_label": {
"permission": "gitea.issue.comment",
"role": "author",
@@ -504,9 +518,11 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
"gitea_create_issue": "create_issue",
"gitea_close_issue": "close_issue",
"gitea_edit_issue": "edit_issue",
"gitea_create_issue_comment": "comment_issue",
"gitea_mark_issue": "mark_issue",
"gitea_set_issue_labels": "set_issue_labels",
"gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels",
"gitea_create_label": "create_label",
"gitea_commit_files": "commit_files",
}
+308
View File
@@ -0,0 +1,308 @@
"""Authoritative terminal-transition cleanup for ``status:pr-open`` (#780).
``status:pr-open`` is applied by ``gitea_create_pr`` while a linked pull
request is open. Nothing removed it again: the workflow's terminal paths
(merge, close-without-merge, supersession, already-landed reconciliation,
controller closure) each ended without touching the label, so a repository
audit found 40 closed issues still carrying it.
This module is the single source of truth for that cleanup. Every sanctioned
terminal path plans its label mutation here rather than implementing its own
rule, so the paths cannot drift apart:
- :func:`plan_pr_open_cleanup` decides the exact resulting label set. It only
ever removes ``status:pr-open``; every other label is preserved verbatim,
including the case where the result is an empty label set.
- :func:`verify_pr_open_cleanup` is the read-after-write check. It proves the
label is gone *and* that no unrelated label was dropped or added.
- :func:`detect_residual_pr_open` is the terminal validation: given issues, it
reports any that still carry the label, so a controller closure or audit
fails loudly instead of leaving the leak behind.
The rule is idempotent by construction: an issue without the label plans no
mutation, so retries and recovery re-runs are harmless.
This module performs no I/O — callers own the Gitea API calls.
"""
from __future__ import annotations
from typing import Any, Iterable, Mapping, Sequence
import issue_workflow_labels
#: The single label this module is responsible for retiring.
PR_OPEN_LABEL = "status:pr-open"
# Canonical terminal reasons — the sanctioned ways an issue can end up
# associated with a pull request that is no longer open.
MERGED = "merged"
CLOSED_WITHOUT_MERGE = "closed_without_merge"
SUPERSEDED = "superseded"
ALREADY_LANDED = "already_landed"
CONTROLLER_CLOSURE = "controller_closure"
ABANDONED = "abandoned"
RETRY_RECOVERY = "retry_recovery"
TERMINAL_REASONS: tuple[str, ...] = (
MERGED,
CLOSED_WITHOUT_MERGE,
SUPERSEDED,
ALREADY_LANDED,
CONTROLLER_CLOSURE,
ABANDONED,
RETRY_RECOVERY,
)
_REASON_ALIASES: dict[str, str] = {
"merge": MERGED,
"merged": MERGED,
"pr_merged": MERGED,
"closed": CLOSED_WITHOUT_MERGE,
"close": CLOSED_WITHOUT_MERGE,
"closed_without_merge": CLOSED_WITHOUT_MERGE,
"pr_closed": CLOSED_WITHOUT_MERGE,
"supersede": SUPERSEDED,
"superseded": SUPERSEDED,
"supersession": SUPERSEDED,
"already_landed": ALREADY_LANDED,
"reconcile_already_landed": ALREADY_LANDED,
"controller_closure": CONTROLLER_CLOSURE,
"close_issue": CONTROLLER_CLOSURE,
"abandon": ABANDONED,
"abandoned": ABANDONED,
"retry": RETRY_RECOVERY,
"recovery": RETRY_RECOVERY,
"retry_recovery": RETRY_RECOVERY,
}
#: Human-readable phrasing used in audit comments and diagnostics.
REASON_DESCRIPTIONS: dict[str, str] = {
MERGED: "the linked PR was merged",
CLOSED_WITHOUT_MERGE: "the linked PR was closed without merging",
SUPERSEDED: "the linked PR was superseded by another merged PR",
ALREADY_LANDED: "the linked PR's change was already on the target branch",
CONTROLLER_CLOSURE: "the issue reached controller closure",
ABANDONED: "the linked PR was abandoned",
RETRY_RECOVERY: "a partial terminal transition is being recovered",
}
def canonical_terminal_reason(reason: str | None) -> str:
"""Normalize a terminal reason, failing closed on anything unrecognized."""
name = (reason or "").strip()
if name in TERMINAL_REASONS:
return name
normalized = name.lower().replace("-", "_").replace(" ", "_")
try:
return _REASON_ALIASES[normalized]
except KeyError as exc:
raise ValueError(
f"unknown terminal PR reason '{reason}' (expected one of: "
+ ", ".join(TERMINAL_REASONS)
+ ")"
) from exc
def plan_pr_open_cleanup(
current_labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
*,
terminal_reason: str,
) -> dict[str, Any]:
"""Plan the label set an issue must carry after a terminal PR transition.
The plan removes ``status:pr-open`` and nothing else. When the label is
absent the plan is an explicit no-op (``cleanup_required`` False), which is
what makes repeated cleanup calls harmless. When it was the only label the
resulting set is legitimately empty.
"""
reason = canonical_terminal_reason(terminal_reason)
before = issue_workflow_labels.label_names(current_labels)
after = [name for name in before if name != PR_OPEN_LABEL]
present = len(after) != len(before)
return {
"terminal_reason": reason,
"terminal_reason_description": REASON_DESCRIPTIONS[reason],
"label": PR_OPEN_LABEL,
"label_present": present,
"cleanup_required": present,
"idempotent_noop": not present,
"labels_before": before,
"labels_after": after,
"removed": [PR_OPEN_LABEL] if present else [],
"preserved": list(after),
"empty_label_set": not after,
}
def verify_pr_open_cleanup(
observed_labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
*,
plan: Mapping[str, Any],
) -> dict[str, Any]:
"""Read-after-write check for a planned cleanup.
Verifies the label is gone and that the observed set matches the plan
exactly, so an unrelated label silently dropped (or re-added) by the API is
reported rather than accepted.
"""
observed = issue_workflow_labels.label_names(observed_labels)
expected = list(plan.get("labels_after") or [])
observed_set = set(observed)
expected_set = set(expected)
residual = PR_OPEN_LABEL in observed_set
unexpected_removals = sorted(expected_set - observed_set)
unexpected_additions = sorted(observed_set - expected_set - {PR_OPEN_LABEL})
reasons: list[str] = []
if residual:
reasons.append(
f"'{PR_OPEN_LABEL}' is still present after terminal cleanup"
)
if unexpected_removals:
reasons.append(
"unrelated labels were dropped by the cleanup: "
+ ", ".join(unexpected_removals)
)
if unexpected_additions:
reasons.append(
"unexpected labels appeared during the cleanup: "
+ ", ".join(unexpected_additions)
)
verified = not reasons
return {
"verified": verified,
"residual": residual,
"observed_labels": observed,
"expected_labels": expected,
"unexpected_removals": unexpected_removals,
"unexpected_additions": unexpected_additions,
"empty_label_set": not observed,
"reasons": reasons,
"safe_next_action": (
""
if verified
else (
"Re-run the terminal cleanup for this issue with "
f"terminal_reason='{RETRY_RECOVERY}' and confirm the read-back "
f"no longer reports '{PR_OPEN_LABEL}'."
)
),
}
def summarize_cleanup_results(
results: Sequence[Mapping[str, Any]],
*,
terminal_reason: str,
) -> dict[str, Any]:
"""Aggregate per-issue cleanup outcomes into one reportable record."""
reason = canonical_terminal_reason(terminal_reason)
entries = [dict(entry) for entry in results]
removed = [e.get("issue_number") for e in entries if e.get("status") == "removed"]
absent = [
e.get("issue_number") for e in entries if e.get("status") == "not present"
]
failed = [
e.get("issue_number")
for e in entries
if e.get("status") not in ("removed", "not present") or not e.get("verified")
]
reasons: list[str] = []
for entry in entries:
for text in entry.get("reasons") or []:
reasons.append(f"issue #{entry.get('issue_number')}: {text}")
clean = not failed
return {
"label": PR_OPEN_LABEL,
"terminal_reason": reason,
"clean": clean,
"checked": [e.get("issue_number") for e in entries],
"removed": removed,
"already_absent": absent,
"failed": failed,
"results": entries,
"reasons": reasons,
"safe_next_action": (
""
if clean
else (
"Terminal label cleanup did not complete for "
+ ", ".join(f"#{num}" for num in failed)
+ ". Re-run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason='{RETRY_RECOVERY}' for those issues."
)
),
}
def detect_residual_pr_open(
issues: Iterable[Mapping[str, Any]],
*,
open_pr_issue_numbers: Iterable[int] = (),
) -> dict[str, Any]:
"""Terminal validation: report issues still carrying ``status:pr-open``.
An issue with a genuinely open pull request is allowed to keep the label,
so *open_pr_issue_numbers* is excluded from the residual set rather than
being reported as a leak.
"""
legitimate: set[int] = set()
for num in open_pr_issue_numbers or ():
try:
legitimate.add(int(num))
except (TypeError, ValueError):
continue
checked = 0
residual: list[dict[str, Any]] = []
exempt: list[int] = []
for issue in issues or []:
checked += 1
names = issue_workflow_labels.label_names(issue)
if PR_OPEN_LABEL not in names:
continue
try:
number = int(issue.get("number"))
except (TypeError, ValueError):
number = None
if number is not None and number in legitimate:
exempt.append(number)
continue
residual.append(
{
"number": number,
"state": issue.get("state"),
"labels": names,
}
)
clean = not residual
reasons = [
(
f"issue #{entry['number']} ({entry.get('state') or 'unknown state'}) "
f"still carries '{PR_OPEN_LABEL}' with no open PR"
)
for entry in residual
]
return {
"label": PR_OPEN_LABEL,
"clean": clean,
"checked_count": checked,
"residual_count": len(residual),
"residual_issues": residual,
"exempt_open_pr_issues": sorted(exempt),
"reasons": reasons,
"safe_next_action": (
""
if clean
else (
"Run gitea_cleanup_terminal_pr_labels with "
f"terminal_reason='{RETRY_RECOVERY}' for issues "
+ ", ".join(f"#{entry['number']}" for entry in residual)
+ " before declaring the terminal transition complete."
)
),
}
@@ -0,0 +1,380 @@
"""Allocator ownership exclusion tests (#765).
One session's active lease must never blockade the author queue for a
different controller. Covers: foreign lease skipped, next unclaimed candidate
selected, own task resumable, task-local blocker quarantined, all-claimed ->
wait, same profile + different controller_instance_id -> different ownership,
and claimed candidates reported in skipped results.
"""
from __future__ import annotations
import os
import tempfile
import unittest
from allocator_service import (
OUTCOME_OWNERSHIP_DEFECT,
OUTCOME_PREVIEW,
OUTCOME_WAIT,
OWNERSHIP_FOREIGN,
OWNERSHIP_OWN,
OWNERSHIP_UNKNOWN,
SKIP_CLAIMED_BY_OTHER_SESSION,
WorkCandidate,
allocate_next_work,
classify_claim_ownership,
resolve_controller_instance_id,
)
from control_plane_db import ControlPlaneDB
REMOTE = "prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"
MINE = "ctl-mine-0001"
THEIRS = "ctl-theirs-0002"
def _issue(number: int, **kwargs) -> WorkCandidate:
base = dict(
kind="issue",
number=number,
state="open",
labels=("status:ready", "type:bug"),
title=f"issue {number}",
priority=20,
)
base.update(kwargs)
return WorkCandidate(**base)
def _claim(number: int, *, session_id: str, instance: str | None, kind: str = "issue"):
return {
"lease_id": f"lease-{number}",
"session_id": session_id,
"controller_instance_id": instance,
"role": "author",
"profile": "prgs-author",
"expires_at": "2026-07-20T07:06:09Z",
"work_kind": kind,
"work_number": number,
}
class AllocatorOwnershipTestCase(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
def _allocate(
self,
candidates,
*,
claims,
session_id="sess-mine",
instance=MINE,
apply=False,
role="author",
):
return allocate_next_work(
self.db,
session_id=session_id,
role=role,
remote=REMOTE,
org=ORG,
repo=REPO,
candidates=candidates,
apply=apply,
profile_name="prgs-author",
controller_instance_id=instance,
claims=claims,
)
class TestOwnershipClassification(AllocatorOwnershipTestCase):
def test_no_claim_returns_none(self):
self.assertIsNone(
classify_claim_ownership(
None, session_id="s", controller_instance_id=MINE
)
)
def test_same_controller_instance_is_own(self):
claim = _claim(1, session_id="other-session", instance=MINE)
self.assertEqual(
classify_claim_ownership(
claim, session_id="sess-mine", controller_instance_id=MINE
),
OWNERSHIP_OWN,
)
def test_same_profile_different_instance_is_foreign(self):
"""Shared profile must not imply shared ownership."""
claim = _claim(1, session_id="other-session", instance=THEIRS)
self.assertEqual(
classify_claim_ownership(
claim, session_id="sess-mine", controller_instance_id=MINE
),
OWNERSHIP_FOREIGN,
)
def test_exact_session_match_is_own(self):
claim = _claim(1, session_id="sess-mine", instance=None)
self.assertEqual(
classify_claim_ownership(
claim, session_id="sess-mine", controller_instance_id=None
),
OWNERSHIP_OWN,
)
def test_legacy_claim_with_neither_side_identified_is_foreign(self):
"""No identities anywhere: a different session id is simply not ours."""
claim = _claim(1, session_id="someone-else", instance=None)
self.assertEqual(
classify_claim_ownership(
claim, session_id="sess-mine", controller_instance_id=None
),
OWNERSHIP_FOREIGN,
)
def test_claim_identified_but_local_undeclared_is_unknown(self):
"""Only one side identified: not comparable, so never adopt."""
claim = _claim(1, session_id="someone-else", instance=THEIRS)
self.assertEqual(
classify_claim_ownership(
claim, session_id="sess-mine", controller_instance_id=None
),
OWNERSHIP_UNKNOWN,
)
def test_local_identified_but_claim_undeclared_is_unknown(self):
"""A legacy lease may be our own under an old session id; do not guess."""
claim = _claim(1, session_id="someone-else", instance=None)
self.assertEqual(
classify_claim_ownership(
claim, session_id="sess-mine", controller_instance_id=MINE
),
OWNERSHIP_UNKNOWN,
)
def test_resolve_controller_instance_id_reads_env(self):
self.assertEqual(
resolve_controller_instance_id({"GITEA_CONTROLLER_INSTANCE_ID": MINE}),
MINE,
)
self.assertIsNone(resolve_controller_instance_id({}))
self.assertIsNone(
resolve_controller_instance_id({"GITEA_CONTROLLER_INSTANCE_ID": " "})
)
class TestForeignLeaseDoesNotBlockade(AllocatorOwnershipTestCase):
def test_foreign_claim_skipped_and_next_issue_selected(self):
"""Skip the claimed issue, select the next unclaimed one."""
candidates = [_issue(607), _issue(615), _issue(617)]
claims = {
("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS)
}
result = self._allocate(candidates, claims=claims)
self.assertEqual(result["outcome"], OUTCOME_PREVIEW)
self.assertEqual(result["selected"]["number"], 615)
skipped_607 = [s for s in result["skipped"] if s["number"] == 607]
self.assertEqual(len(skipped_607), 1)
self.assertEqual(
skipped_607[0]["reason_code"], SKIP_CLAIMED_BY_OTHER_SESSION
)
self.assertIn(SKIP_CLAIMED_BY_OTHER_SESSION, skipped_607[0]["reason"])
def test_claimed_candidate_appears_in_skipped_inventory(self):
"""Skipped reporting must reflect claimed candidates."""
candidates = [_issue(607), _issue(615)]
claims = {
("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS)
}
result = self._allocate(candidates, claims=claims)
self.assertEqual(len(result["skipped"]), 1)
self.assertEqual(len(result["claims_excluded"]), 1)
excluded = result["claims_excluded"][0]
self.assertEqual(excluded["number"], 607)
self.assertEqual(excluded["ownership"], OWNERSHIP_FOREIGN)
self.assertEqual(excluded["owner_controller_instance_id"], THEIRS)
def test_task_local_blocker_does_not_freeze_unrelated_work(self):
"""A quarantined task must not stop the rest of the queue."""
candidates = [_issue(607), _issue(615), _issue(617)]
claims = {
("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS)
}
first = self._allocate(candidates, claims=claims)
self.assertEqual(first["selected"]["number"], 615)
# 615 then gets claimed by yet another controller; queue still advances.
claims[("issue", 615)] = _claim(
615, session_id="sess-third", instance="ctl-third-0003"
)
second = self._allocate(candidates, claims=claims)
self.assertEqual(second["selected"]["number"], 617)
def test_multiple_controllers_get_different_issues(self):
"""Concurrent author sessions work on different issues."""
candidates = [_issue(607), _issue(615)]
claims = {
("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS)
}
mine = self._allocate(candidates, claims=claims, instance=MINE)
theirs = self._allocate(
candidates, claims=claims, session_id="sess-theirs", instance=THEIRS
)
self.assertEqual(mine["selected"]["number"], 615)
# The other controller may still be handed its own in-progress task.
self.assertEqual(theirs["selected"]["number"], 607)
def test_unclaimed_queue_is_unaffected(self):
candidates = [_issue(607), _issue(615)]
result = self._allocate(candidates, claims={})
self.assertEqual(result["selected"]["number"], 607)
self.assertEqual(result["skipped"], [])
self.assertEqual(result["claims_excluded"], [])
class TestOwnTaskResume(AllocatorOwnershipTestCase):
def test_controller_may_resume_its_own_active_task(self):
"""Own claim stays selectable across a new session id."""
candidates = [_issue(607), _issue(615)]
claims = {
("issue", 607): _claim(607, session_id="sess-mine-old", instance=MINE)
}
result = self._allocate(
candidates, claims=claims, session_id="sess-mine-new", instance=MINE
)
self.assertEqual(result["selected"]["number"], 607)
self.assertEqual(result["claims_excluded"], [])
def test_own_claim_by_exact_session_is_selectable(self):
candidates = [_issue(607)]
claims = {("issue", 607): _claim(607, session_id="sess-mine", instance=None)}
result = self._allocate(
candidates, claims=claims, session_id="sess-mine", instance=None
)
self.assertEqual(result["selected"]["number"], 607)
class TestAllCandidatesClaimed(AllocatorOwnershipTestCase):
def test_all_claimed_returns_wait_not_a_claimed_selection(self):
"""Never hand back a claimed issue; report waiting instead."""
candidates = [_issue(607), _issue(615)]
claims = {
("issue", 607): _claim(607, session_id="sess-a", instance=THEIRS),
("issue", 615): _claim(615, session_id="sess-b", instance="ctl-c-0003"),
}
result = self._allocate(candidates, claims=claims)
self.assertIsNone(result["selected"])
self.assertEqual(result["outcome"], OUTCOME_WAIT)
self.assertEqual(len(result["claims_excluded"]), 2)
def test_unidentifiable_owner_reports_ownership_defect(self):
"""Refuse to adopt when ownership cannot be established."""
candidates = [_issue(607)]
claims = {("issue", 607): _claim(607, session_id="sess-legacy", instance=None)}
result = self._allocate(candidates, claims=claims)
self.assertIsNone(result["selected"])
self.assertEqual(result["outcome"], OUTCOME_OWNERSHIP_DEFECT)
self.assertEqual(len(result["ownership_defects"]), 1)
self.assertEqual(
result["ownership_defects"][0]["ownership"], OWNERSHIP_UNKNOWN
)
class TestClaimsFromControlPlaneDb(AllocatorOwnershipTestCase):
"""End-to-end against the real substrate, not injected claim dicts."""
def _seed_lease(self, number: int, *, session_id: str, instance: str | None):
self.db.upsert_session(
session_id=session_id,
role="author",
profile="prgs-author",
pid=4242,
controller_instance_id=instance,
)
return self.db.assign_and_lease(
session_id=session_id,
role="author",
remote=REMOTE,
org=ORG,
repo=REPO,
kind="issue",
number=number,
)
def test_controller_instance_id_persists_on_session(self):
row = self.db.upsert_session(
session_id="sess-x",
role="author",
profile="prgs-author",
pid=1,
controller_instance_id=MINE,
)
self.assertEqual(row["controller_instance_id"], MINE)
def test_heartbeat_without_instance_does_not_erase_ownership(self):
self.db.upsert_session(
session_id="sess-x",
role="author",
profile="prgs-author",
pid=1,
controller_instance_id=MINE,
)
row = self.db.upsert_session(
session_id="sess-x", role="author", profile="prgs-author", pid=1
)
self.assertEqual(row["controller_instance_id"], MINE)
def test_list_active_claims_surfaces_owner_instance(self):
self._seed_lease(607, session_id="sess-theirs", instance=THEIRS)
claims = self.db.list_active_claims(remote=REMOTE, org=ORG, repo=REPO)
self.assertIn(("issue", 607), claims)
self.assertEqual(claims[("issue", 607)]["controller_instance_id"], THEIRS)
def test_live_foreign_lease_is_excluded_without_injected_claims(self):
self._seed_lease(607, session_id="sess-theirs", instance=THEIRS)
result = allocate_next_work(
self.db,
session_id="sess-mine",
role="author",
remote=REMOTE,
org=ORG,
repo=REPO,
candidates=[_issue(607), _issue(615)],
apply=False,
profile_name="prgs-author",
controller_instance_id=MINE,
)
self.assertEqual(result["selected"]["number"], 615)
self.assertEqual(
result["skipped"][0]["reason_code"], SKIP_CLAIMED_BY_OTHER_SESSION
)
def test_apply_reserves_the_unclaimed_issue(self):
self._seed_lease(607, session_id="sess-theirs", instance=THEIRS)
result = allocate_next_work(
self.db,
session_id="sess-mine",
role="author",
remote=REMOTE,
org=ORG,
repo=REPO,
candidates=[_issue(607), _issue(615)],
apply=True,
profile_name="prgs-author",
controller_instance_id=MINE,
)
self.assertEqual(result["outcome"], "assigned_work")
self.assertEqual(result["selected"]["number"], 615)
self.assertEqual(result["assignment"]["work_number"], 615)
if __name__ == "__main__":
unittest.main()
+349
View File
@@ -0,0 +1,349 @@
"""Allocator pre-rank exclusions and candidates_json transport (#776).
Covers:
* #617 excluded before ranking (never leased when exclude_issue_numbers=[617]);
* excluded top candidate selects the next safe candidate;
* all candidates excluded → WAIT, no lease;
* decoded-list and JSON-string candidates_json;
* malformed / type-invalid fail-closed cases;
* dry-run/apply fingerprint match and drift rejection;
* foreign lease and same-owner lease on excluded issue;
* skipped-accounting reason parity (excluded_by_controller);
* public MCP entry-point coverage for exclude_issue_numbers.
"""
from __future__ import annotations
import json
import os
import tempfile
import unittest
from unittest.mock import patch
import gitea_mcp_server as srv
from allocator_service import (
OUTCOME_ASSIGNED,
OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE,
OUTCOME_CANDIDATE_SET_DRIFT,
OUTCOME_NO_SAFE,
OUTCOME_PREVIEW,
OUTCOME_WAIT,
SKIP_CLAIMED_BY_OTHER_SESSION,
SKIP_EXCLUDED_BY_CONTROLLER,
WorkCandidate,
allocate_next_work,
candidate_from_dict,
candidate_set_fingerprint,
normalize_candidates_payload,
normalize_exclude_issue_numbers,
)
from control_plane_db import ControlPlaneDB
REMOTE = "prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"
MINE = "ctl-mine-776"
THEIRS = "ctl-theirs-776"
def _issue(number: int, **kwargs) -> WorkCandidate:
base = dict(
kind="issue",
number=number,
state="open",
labels=("status:ready", "type:bug"),
title=f"issue {number}",
priority=20,
)
base.update(kwargs)
return WorkCandidate(**base)
def _claim(number: int, *, session_id: str, instance: str | None, kind: str = "issue"):
return {
"lease_id": f"lease-{number}",
"session_id": session_id,
"controller_instance_id": instance,
"role": "author",
"profile": "prgs-author",
"expires_at": "2026-07-21T12:00:00Z",
"work_kind": kind,
"work_number": number,
}
def _cand_dict(number: int, **kwargs) -> dict:
d = {
"kind": "issue",
"number": number,
"state": "open",
"labels": ["status:ready", "type:bug"],
"title": f"issue {number}",
"priority": 20,
}
d.update(kwargs)
return d
class AllocatorExcludeServiceTest(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-776",
role="author",
remote=REMOTE,
org=ORG,
repo=REPO,
profile_name="prgs-author",
controller_instance_id=MINE,
claims={},
apply=False,
)
defaults.update(kwargs)
return allocate_next_work(self.db, candidates=candidates, **defaults)
def test_exclude_617_before_ranking_never_selects(self) -> None:
"""AC2/AC8: highest-ranked #617 is removed before ranking."""
cands = [_issue(617), _issue(700), _issue(701)]
res = self._alloc(cands, exclude_issue_numbers=[617])
self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
self.assertEqual(res["selected"]["number"], 700)
skipped = {s["number"]: s for s in res["skipped"]}
self.assertIn(617, skipped)
self.assertEqual(
skipped[617]["reason_code"], SKIP_EXCLUDED_BY_CONTROLLER
)
self.assertIn(SKIP_EXCLUDED_BY_CONTROLLER, skipped[617]["reason"])
def test_excluded_top_selects_next_safe(self) -> None:
"""AC2: excluding the oldest ready issue promotes the next number."""
cands = [_issue(600), _issue(601), _issue(602)]
res = self._alloc(cands, exclude_issue_numbers=[600])
self.assertEqual(res["selected"]["number"], 601)
def test_all_candidates_excluded_wait_no_lease(self) -> None:
"""AC7: every candidate excluded → WAIT, no assignment."""
cands = [_issue(617), _issue(700)]
res = self._alloc(cands, exclude_issue_numbers=[617, 700], apply=True)
self.assertEqual(res["outcome"], OUTCOME_WAIT)
self.assertIsNone(res["selected"])
self.assertIsNone(res["assignment"])
self.assertEqual(len(res["controller_excluded"]), 2)
def test_omit_exclude_retains_existing_behavior(self) -> None:
"""AC1/AC9: omit exclusions → #617 still wins when oldest ready."""
cands = [_issue(617), _issue(700)]
res = self._alloc(cands)
self.assertEqual(res["selected"]["number"], 617)
self.assertEqual(res.get("exclude_issue_numbers"), [])
def test_foreign_lease_still_skipped(self) -> None:
"""AC6/AC9: foreign claims keep SKIP_CLAIMED_BY_OTHER_SESSION."""
cands = [_issue(617), _issue(700)]
claims = {
("issue", 700): _claim(700, session_id="other", instance=THEIRS),
}
res = self._alloc(
cands, exclude_issue_numbers=[617], claims=claims
)
# 617 excluded, 700 foreign → wait, no selection
self.assertEqual(res["outcome"], OUTCOME_WAIT)
self.assertIsNone(res["selected"])
codes = {s["reason_code"] for s in res["skipped"]}
self.assertIn(SKIP_EXCLUDED_BY_CONTROLLER, codes)
self.assertIn(SKIP_CLAIMED_BY_OTHER_SESSION, codes)
def test_same_owner_lease_on_excluded_blocks_resume_release(self) -> None:
"""AC5: excluded + live same-owner lease → structured blocker."""
cands = [_issue(617), _issue(700)]
claims = {
("issue", 617): _claim(617, session_id="sess-776", instance=MINE),
}
res = self._alloc(
cands, exclude_issue_numbers=[617], claims=claims, apply=True
)
self.assertEqual(res["outcome"], OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE)
self.assertIsNone(res["assignment"])
self.assertEqual(res["blocked_lease"]["number"], 617)
self.assertIn("resume", res["blocked_lease"]["safe_next_action"])
def test_dry_run_apply_fingerprint_match(self) -> None:
"""AC4: dry-run and apply share the same fingerprint."""
cands = [_issue(617), _issue(700)]
dry = self._alloc(cands, exclude_issue_numbers=[617], apply=False)
apply_res = self._alloc(
cands,
exclude_issue_numbers=[617],
apply=True,
expected_candidate_set_fingerprint=dry["candidate_set_fingerprint"],
)
self.assertEqual(
dry["candidate_set_fingerprint"],
apply_res["candidate_set_fingerprint"],
)
self.assertEqual(apply_res["outcome"], OUTCOME_ASSIGNED)
self.assertEqual(apply_res["selected"]["number"], 700)
def test_apply_rejects_fingerprint_drift(self) -> None:
"""AC4: material candidate-set drift fails closed on apply."""
cands = [_issue(617), _issue(700)]
res = self._alloc(
cands,
exclude_issue_numbers=[617],
apply=True,
expected_candidate_set_fingerprint="0" * 64,
)
self.assertFalse(res["success"])
self.assertEqual(res["outcome"], OUTCOME_CANDIDATE_SET_DRIFT)
self.assertIsNone(res["assignment"])
def test_fingerprint_stable_helper(self) -> None:
cands = [_issue(700), _issue(617)]
a = candidate_set_fingerprint(cands, exclude_issue_numbers=[617])
b = candidate_set_fingerprint(
list(reversed(cands)), exclude_issue_numbers=[617]
)
self.assertEqual(a, b)
def test_normalize_exclude_rejects_bool(self) -> None:
with self.assertRaises(ValueError):
normalize_exclude_issue_numbers([True])
def test_normalize_exclude_rejects_scalar(self) -> None:
with self.assertRaises(ValueError):
normalize_exclude_issue_numbers(617)
class CandidatesJsonNormalizeTest(unittest.TestCase):
def test_decoded_list(self) -> None:
"""AC3: already-decoded list from MCP transport."""
cands = normalize_candidates_payload([_cand_dict(617), _cand_dict(700)])
self.assertEqual([c.number for c in cands], [617, 700])
def test_json_string(self) -> None:
"""AC3: backward-compatible JSON string."""
raw = json.dumps([_cand_dict(617)])
cands = normalize_candidates_payload(raw)
self.assertEqual(cands[0].number, 617)
def test_malformed_json_fail_closed(self) -> None:
with self.assertRaises(ValueError) as ctx:
normalize_candidates_payload("{not json")
self.assertIn("malformed", str(ctx.exception).lower())
def test_scalar_fail_closed(self) -> None:
with self.assertRaises(ValueError):
normalize_candidates_payload(42)
def test_bool_number_fail_closed(self) -> None:
with self.assertRaises(ValueError):
normalize_candidates_payload([_cand_dict(True)]) # type: ignore[arg-type]
def test_invalid_record_fail_closed(self) -> None:
with self.assertRaises(ValueError):
normalize_candidates_payload(["not-a-dict"])
def test_object_not_list_fail_closed(self) -> None:
with self.assertRaises(ValueError):
normalize_candidates_payload(json.dumps({"number": 1}))
def test_candidate_from_dict_rejects_bool_number(self) -> None:
with self.assertRaises(ValueError):
candidate_from_dict({"kind": "issue", "number": True})
class AllocateNextWorkMcpExcludeTest(unittest.TestCase):
"""Public MCP entry-point coverage (#776 AC8)."""
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
def tearDown(self) -> None:
self._tmp.cleanup()
def _call(self, **kwargs):
with patch("gitea_mcp_server._profile_operation_gate", return_value=None), patch(
"gitea_mcp_server._resolve", return_value=("h", ORG, REPO)
), patch(
"gitea_mcp_server.get_profile",
return_value={"profile_name": "prgs-author", "role": "author"},
), patch(
"gitea_mcp_server._authenticated_username", return_value="jcwalker3"
), patch(
"gitea_mcp_server._control_plane_db_or_error", return_value=(self.db, [])
), patch(
"gitea_mcp_server.sentry_observability.monitor_checkin", return_value=None
):
return srv.gitea_allocate_next_work(
remote="prgs", org=ORG, repo=REPO, role="author", **kwargs
)
def test_mcp_exclude_617_decoded_list_never_selects(self) -> None:
"""AC8: public tool with decoded list + exclude_issue_numbers=[617]."""
candidates = [_cand_dict(617), _cand_dict(700)]
res = self._call(
candidates_json=candidates,
exclude_issue_numbers=[617],
apply=False,
)
self.assertTrue(res.get("success"), res)
self.assertEqual(res["selected"]["number"], 700)
skipped = {s["number"]: s for s in res["skipped"]}
self.assertEqual(
skipped[617]["reason_code"], SKIP_EXCLUDED_BY_CONTROLLER
)
self.assertNotEqual(res["selected"]["number"], 617)
def test_mcp_exclude_617_json_string_apply(self) -> None:
"""AC8: JSON-string transport + apply never leases #617."""
raw = json.dumps([_cand_dict(617), _cand_dict(700)])
res = self._call(
candidates_json=raw,
exclude_issue_numbers=[617],
apply=True,
)
self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
self.assertEqual(res["assignment"]["work_number"], 700)
self.assertNotEqual(res["selected"]["number"], 617)
def test_mcp_malformed_candidates_json_fail_closed(self) -> None:
res = self._call(candidates_json="{bad", apply=False)
self.assertFalse(res["success"])
self.assertIsNone(res["assignment"])
self.assertTrue(any("fail closed" in r for r in res["reasons"]))
def test_mcp_bool_number_fail_closed(self) -> None:
res = self._call(
candidates_json=[{"kind": "issue", "number": True, "priority": 20}],
apply=False,
)
self.assertFalse(res["success"])
self.assertIsNone(res["assignment"])
def test_mcp_fingerprint_dry_run_apply_parity(self) -> None:
candidates = [_cand_dict(617), _cand_dict(700)]
dry = self._call(
candidates_json=candidates,
exclude_issue_numbers=[617],
apply=False,
)
apply_res = self._call(
candidates_json=candidates,
exclude_issue_numbers=[617],
apply=True,
expected_candidate_set_fingerprint=dry["candidate_set_fingerprint"],
)
self.assertEqual(
dry["candidate_set_fingerprint"],
apply_res["candidate_set_fingerprint"],
)
self.assertEqual(apply_res["selected"]["number"], 700)
if __name__ == "__main__":
unittest.main()
+11 -1
View File
@@ -238,7 +238,17 @@ class TestSimpleToolAudit(_AuditWiringBase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_close_issue_audited(self, _auth, mock_api):
mock_api.side_effect = [{"state": "closed"}, {"login": "mgr-bot"}]
# Keyed rather than positional: closing an issue also reads its labels
# before and after the state change for the #780 terminal cleanup and
# its read-after-write check, so call order is not a fixed sequence.
def api(method, url, auth, payload=None):
if method == "PATCH":
return {"state": "closed"}
if "/issues/" in url:
return {"number": 42, "labels": []}
return {"login": "mgr-bot"}
mock_api.side_effect = api
with patch.dict(os.environ, self._env(), clear=True):
gitea_close_issue(issue_number=42, remote="prgs")
recs = self._records()
+635
View File
@@ -0,0 +1,635 @@
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from mutation_profile_fixture import install_deterministic_remote_urls # noqa: E402
install_deterministic_remote_urls()
"""#781: sanctioned issue title/body editing, and the documentation drift guard.
Two defects are covered here. The first is that no MCP path could edit an issue
title or body at all, so an authorized correction had to be recorded as a
comment. The second is why nobody noticed: documentation named a tool that was
never registered, and nothing compared the two lists.
"""
import glob
import json
import os
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import anti_stomp_preflight # noqa: E402
import edit_issue # noqa: E402
import mcp_server # noqa: E402
import mcp_tool_inventory # noqa: E402
import task_capability_map # noqa: E402
REPO_ROOT = Path(__file__).resolve().parent.parent
CONFIG = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
}
},
"profiles": {
"edit-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "author-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": ["gitea.read", "gitea.issue.comment"],
"forbidden_operations": [],
"allowed_repositories": [
"Scaled-Tech-Consulting/Gitea-Tools",
"Example-Org/Example-Repo",
"913443/eAgenda",
],
"execution_profile": "edit-author",
},
"read-only-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "reader-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": ["gitea.read"],
"forbidden_operations": ["gitea.issue.comment"],
"allowed_repositories": [
"Scaled-Tech-Consulting/Gitea-Tools",
"Example-Org/Example-Repo",
"913443/eAgenda",
],
"execution_profile": "read-only-author",
},
},
"rules": {"allow_runtime_switching": False},
}
ISSUE_NUMBER = 9
ORIGINAL_TITLE = "fix(mcp): original title"
ORIGINAL_BODY = "Original body.\n"
NEW_TITLE = "fix(mcp): corrected title"
NEW_BODY = "Corrected body.\n"
def _registered_tool_names() -> set[str]:
manager = mcp_server.mcp._tool_manager
return set((getattr(manager, "_tools", None) or {}).keys())
# ---------------------------------------------------------------------------
# Rule: request validation
# ---------------------------------------------------------------------------
class TestValidateEditRequest(unittest.TestCase):
def test_no_field_is_rejected(self):
with self.assertRaises(ValueError) as ctx:
edit_issue.validate_edit_request()
self.assertIn("At least one field", str(ctx.exception))
def test_blank_title_is_rejected(self):
with self.assertRaises(ValueError) as ctx:
edit_issue.validate_edit_request(title=" ")
self.assertIn("cannot be blank", str(ctx.exception))
def test_non_string_title_is_rejected(self):
with self.assertRaises(ValueError):
edit_issue.validate_edit_request(title=42)
def test_non_string_body_is_rejected(self):
with self.assertRaises(ValueError):
edit_issue.validate_edit_request(body=["not", "a", "string"])
def test_empty_body_is_a_legitimate_edit(self):
self.assertEqual(edit_issue.validate_edit_request(body=""), {"body": ""})
# ---------------------------------------------------------------------------
# Rule: planning against the pre-image
# ---------------------------------------------------------------------------
class TestPlanIssueEdit(unittest.TestCase):
def _current(self, **overrides):
issue = {
"number": ISSUE_NUMBER,
"title": ORIGINAL_TITLE,
"body": ORIGINAL_BODY,
"state": "open",
"labels": [{"name": "type:bug"}, {"name": "mcp"}],
"assignees": [{"login": "author-user"}],
"milestone": {"title": "v1.2.0"},
}
issue.update(overrides)
return issue
def test_title_only_sends_only_the_title(self):
plan = edit_issue.plan_issue_edit(self._current(), title=NEW_TITLE)
self.assertEqual(plan["payload"], {"title": NEW_TITLE})
self.assertEqual(plan["requested_fields"], ["title"])
self.assertFalse(plan["no_op"])
def test_body_only_sends_only_the_body(self):
plan = edit_issue.plan_issue_edit(self._current(), body=NEW_BODY)
self.assertEqual(plan["payload"], {"body": NEW_BODY})
def test_combined_edit_sends_both(self):
plan = edit_issue.plan_issue_edit(
self._current(), title=NEW_TITLE, body=NEW_BODY
)
self.assertEqual(plan["payload"], {"title": NEW_TITLE, "body": NEW_BODY})
self.assertEqual(plan["requested_fields"], ["body", "title"])
def test_identical_content_is_an_explicit_no_op(self):
plan = edit_issue.plan_issue_edit(
self._current(), title=ORIGINAL_TITLE, body=ORIGINAL_BODY
)
self.assertTrue(plan["no_op"])
self.assertEqual(plan["payload"], {})
self.assertTrue(plan["reasons"])
self.assertTrue(plan["safe_next_action"])
def test_partially_unchanged_request_sends_only_the_difference(self):
plan = edit_issue.plan_issue_edit(
self._current(), title=ORIGINAL_TITLE, body=NEW_BODY
)
self.assertFalse(plan["no_op"])
self.assertEqual(plan["payload"], {"body": NEW_BODY})
self.assertEqual(plan["unchanged_fields"], ["title"])
def test_missing_body_is_compared_as_empty(self):
current = self._current()
current.pop("body")
plan = edit_issue.plan_issue_edit(current, body="")
self.assertTrue(plan["no_op"])
def test_preserved_snapshot_captures_untouched_fields(self):
plan = edit_issue.plan_issue_edit(self._current(), title=NEW_TITLE)
before = plan["preserved_before"]
self.assertEqual(before["state"], "open")
self.assertEqual(before["labels"], ["type:bug", "mcp"])
self.assertEqual(before["assignees"], ["author-user"])
self.assertEqual(before["milestone"], "v1.2.0")
# ---------------------------------------------------------------------------
# Rule: pull requests are refused
# ---------------------------------------------------------------------------
class TestAssessIssueTarget(unittest.TestCase):
def test_issue_is_accepted(self):
target = edit_issue.assess_issue_target(
{"number": 9, "title": "t"}, issue_number=9
)
self.assertTrue(target["is_issue"])
self.assertEqual(target["reasons"], [])
def test_pull_request_is_refused_with_a_next_action(self):
target = edit_issue.assess_issue_target(
{"number": 9, "pull_request": {"merged": False}}, issue_number=9
)
self.assertFalse(target["is_issue"])
self.assertTrue(target["is_pull_request"])
self.assertIn("gitea_edit_pr", target["safe_next_action"])
# ---------------------------------------------------------------------------
# Rule: read-after-write verification
# ---------------------------------------------------------------------------
class TestVerifyIssueEdit(unittest.TestCase):
def _plan(self, **kwargs):
current = {
"number": ISSUE_NUMBER,
"title": ORIGINAL_TITLE,
"body": ORIGINAL_BODY,
"state": "open",
"labels": [{"name": "type:bug"}],
"assignees": [],
"milestone": None,
}
return edit_issue.plan_issue_edit(current, **kwargs)
def test_applied_content_verifies(self):
plan = self._plan(title=NEW_TITLE)
observed = {
"title": NEW_TITLE,
"body": ORIGINAL_BODY,
"state": "open",
"labels": [{"name": "type:bug"}],
"assignees": [],
"milestone": None,
}
result = edit_issue.verify_issue_edit(observed, plan=plan)
self.assertTrue(result["verified"])
self.assertTrue(result["preserved_intact"])
self.assertEqual(result["applied"], {"title": NEW_TITLE})
def test_unapplied_content_fails_closed(self):
plan = self._plan(title=NEW_TITLE)
observed = {
"title": ORIGINAL_TITLE,
"state": "open",
"labels": [{"name": "type:bug"}],
}
result = edit_issue.verify_issue_edit(observed, plan=plan)
self.assertFalse(result["verified"])
self.assertEqual(result["mismatches"][0]["field"], "title")
self.assertTrue(result["safe_next_action"])
def test_dropped_label_fails_closed(self):
plan = self._plan(title=NEW_TITLE)
observed = {"title": NEW_TITLE, "state": "open", "labels": []}
result = edit_issue.verify_issue_edit(observed, plan=plan)
self.assertFalse(result["verified"])
self.assertFalse(result["preserved_intact"])
self.assertEqual(result["preserved_changed"][0]["field"], "labels")
def test_changed_state_fails_closed(self):
plan = self._plan(body=NEW_BODY)
observed = {
"body": NEW_BODY,
"state": "closed",
"labels": [{"name": "type:bug"}],
}
result = edit_issue.verify_issue_edit(observed, plan=plan)
self.assertFalse(result["verified"])
self.assertEqual(result["preserved_changed"][0]["field"], "state")
# ---------------------------------------------------------------------------
# Tool: gitea_edit_issue against a fake Gitea
# ---------------------------------------------------------------------------
class _EditIssueToolHarness(unittest.TestCase):
def setUp(self):
self._remotes = patch.dict(
mcp_server.REMOTES,
{
"prgs": {
"host": "gitea.example.com",
"org": "Example-Org",
"repo": "Example-Repo",
}
},
)
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG))
self.issue = {
"number": ISSUE_NUMBER,
"title": ORIGINAL_TITLE,
"body": ORIGINAL_BODY,
"state": "open",
"labels": [{"name": "type:bug"}, {"name": "mcp"}],
"assignees": [{"login": "author-user"}],
"milestone": {"title": "v1.2.0"},
"html_url": "https://gitea.example.com/Example-Org/Example-Repo/issues/9",
}
self.calls: list[tuple[str, str]] = []
self.patched_payloads: list[dict] = []
patch("gitea_audit.audit_enabled", return_value=False).start()
patch("mcp_server.get_auth_header", return_value="token author-pass").start()
patch("mcp_server.api_request", side_effect=self._api).start()
self.addCleanup(patch.stopall)
def tearDown(self):
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
self._dir.cleanup()
def _api(self, method, url, auth, payload=None):
self.calls.append((method, url))
if url.endswith("/user"):
return {"login": "author-user"}
if "/issues/" in url:
if method == "GET":
return dict(self.issue)
if method == "PATCH":
self.patched_payloads.append(dict(payload or {}))
self.issue.update(payload or {})
return dict(self.issue)
raise AssertionError(f"unexpected API call: {method} {url}")
def _env(self, profile: str = "edit-author") -> dict:
return {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
"PYTEST_CURRENT_TEST": os.environ.get(
"PYTEST_CURRENT_TEST", "issue_781_edit_issue"
),
}
def _edit(self, profile: str = "edit-author", **kwargs):
with patch.dict(os.environ, self._env(profile), clear=True):
return mcp_server.gitea_edit_issue(
issue_number=ISSUE_NUMBER, remote="prgs", **kwargs
)
def _patch_methods(self) -> list[str]:
return [method for method, _url in self.calls if method == "PATCH"]
class TestEditIssueSucceeds(_EditIssueToolHarness):
def test_title_only_edit(self):
result = self._edit(title=NEW_TITLE)
self.assertTrue(result["success"])
self.assertTrue(result["verified"])
self.assertEqual(result["changed_fields"], ["title"])
self.assertEqual(self.patched_payloads, [{"title": NEW_TITLE}])
self.assertEqual(self.issue["title"], NEW_TITLE)
self.assertEqual(self.issue["body"], ORIGINAL_BODY)
def test_body_only_edit(self):
result = self._edit(body=NEW_BODY)
self.assertTrue(result["success"])
self.assertEqual(self.patched_payloads, [{"body": NEW_BODY}])
self.assertEqual(self.issue["title"], ORIGINAL_TITLE)
self.assertEqual(self.issue["body"], NEW_BODY)
def test_combined_edit(self):
result = self._edit(title=NEW_TITLE, body=NEW_BODY)
self.assertTrue(result["success"])
self.assertEqual(
self.patched_payloads, [{"title": NEW_TITLE, "body": NEW_BODY}]
)
self.assertEqual(result["applied"], {"title": NEW_TITLE, "body": NEW_BODY})
def test_body_can_be_cleared(self):
result = self._edit(body="")
self.assertTrue(result["success"])
self.assertEqual(self.issue["body"], "")
def test_targets_the_issue_endpoint_never_the_pull_endpoint(self):
self._edit(title=NEW_TITLE)
patched = [url for method, url in self.calls if method == "PATCH"]
self.assertTrue(patched)
for url in patched:
self.assertIn("/issues/", url)
self.assertNotIn("/pulls/", url)
def test_labels_state_assignee_and_milestone_are_provably_unchanged(self):
result = self._edit(title=NEW_TITLE)
proof = result["read_after_write"]
self.assertTrue(proof["preserved_intact"])
self.assertEqual(proof["preserved_before"], proof["preserved_after"])
self.assertEqual(proof["preserved_after"]["labels"], ["type:bug", "mcp"])
self.assertEqual(proof["preserved_after"]["state"], "open")
self.assertEqual(proof["preserved_after"]["assignees"], ["author-user"])
self.assertEqual(proof["preserved_after"]["milestone"], "v1.2.0")
def test_read_after_write_re_reads_the_issue(self):
self._edit(title=NEW_TITLE)
issue_calls = [method for method, url in self.calls if "/issues/" in url]
self.assertEqual(issue_calls, ["GET", "PATCH", "GET"])
class TestEditIssueFailsClosed(_EditIssueToolHarness):
def test_no_op_request_is_rejected_without_a_patch(self):
result = self._edit(title=ORIGINAL_TITLE)
self.assertFalse(result["success"])
self.assertFalse(result["performed"])
self.assertTrue(result["no_op"])
self.assertEqual(self._patch_methods(), [])
self.assertTrue(result["reasons"])
self.assertTrue(result["safe_next_action"])
def test_invalid_request_raises_before_any_api_call(self):
with self.assertRaises(ValueError):
self._edit()
self.assertEqual(self.calls, [])
def test_blank_title_raises_before_any_api_call(self):
with self.assertRaises(ValueError):
self._edit(title=" ")
self.assertEqual(self.calls, [])
def test_authorization_failure_blocks_before_any_api_call(self):
result = self._edit(profile="read-only-author", title=NEW_TITLE)
self.assertFalse(result["success"])
self.assertFalse(result["performed"])
self.assertIn("permission_report", result)
self.assertEqual(
result["permission_report"]["missing_permission"],
task_capability_map.required_permission("edit_issue"),
)
self.assertEqual(self.calls, [])
def test_pull_request_target_is_refused_without_a_patch(self):
self.issue["pull_request"] = {"merged": False}
result = self._edit(title=NEW_TITLE)
self.assertFalse(result["success"])
self.assertFalse(result["performed"])
self.assertEqual(self._patch_methods(), [])
self.assertIn("gitea_edit_pr", result["safe_next_action"])
def test_pre_read_transport_error_is_reported(self):
def boom(method, url, auth, payload=None):
raise RuntimeError("connection reset by peer")
with patch("mcp_server.api_request", side_effect=boom):
result = self._edit(title=NEW_TITLE)
self.assertFalse(result["success"])
self.assertFalse(result["performed"])
self.assertTrue(result["reasons"])
self.assertTrue(result["safe_next_action"])
def test_patch_transport_error_is_reported_not_swallowed(self):
def flaky(method, url, auth, payload=None):
if method == "PATCH":
raise RuntimeError("gitea exploded")
return self._api(method, url, auth, payload)
with patch("mcp_server.api_request", side_effect=flaky):
result = self._edit(title=NEW_TITLE)
self.assertFalse(result["success"])
self.assertFalse(result["performed"])
self.assertIn("issue edit failed", result["reasons"][0])
self.assertEqual(self.issue["title"], ORIGINAL_TITLE)
def test_read_back_transport_error_reports_an_unverified_edit(self):
state = {"gets": 0}
def flaky(method, url, auth, payload=None):
if method == "GET" and "/issues/" in url:
state["gets"] += 1
if state["gets"] > 1:
raise RuntimeError("read timed out")
return self._api(method, url, auth, payload)
with patch("mcp_server.api_request", side_effect=flaky):
result = self._edit(title=NEW_TITLE)
self.assertFalse(result["success"])
self.assertTrue(result["performed"])
self.assertFalse(result["verified"])
self.assertTrue(result["safe_next_action"])
def test_unapplied_edit_fails_verification(self):
def sticky(method, url, auth, payload=None):
if method == "PATCH":
self.calls.append((method, url))
# Report success but store nothing.
return dict(self.issue)
return self._api(method, url, auth, payload)
with patch("mcp_server.api_request", side_effect=sticky):
result = self._edit(title=NEW_TITLE)
self.assertFalse(result["success"])
self.assertTrue(result["performed"])
self.assertFalse(result["verified"])
self.assertEqual(
result["read_after_write"]["mismatches"][0]["field"], "title"
)
def test_edit_that_drops_a_label_fails_verification(self):
def label_eating(method, url, auth, payload=None):
if method == "PATCH":
self.calls.append((method, url))
self.issue.update(payload or {})
self.issue["labels"] = []
return dict(self.issue)
return self._api(method, url, auth, payload)
with patch("mcp_server.api_request", side_effect=label_eating):
result = self._edit(title=NEW_TITLE)
self.assertFalse(result["success"])
self.assertFalse(result["read_after_write"]["preserved_intact"])
# ---------------------------------------------------------------------------
# Registration and gate wiring
# ---------------------------------------------------------------------------
class TestEditIssueRegistrationAndGates(unittest.TestCase):
def test_tool_is_registered(self):
self.assertIn("gitea_edit_issue", _registered_tool_names())
def test_resolver_task_exists_with_author_role(self):
self.assertEqual(
task_capability_map.required_permission("edit_issue"),
"gitea.issue.comment",
)
self.assertEqual(task_capability_map.required_role("edit_issue"), "author")
def test_tool_gate_matches_the_resolver_task(self):
self.assertEqual(
task_capability_map.ISSUE_MUTATION_TOOL_TASKS["gitea_edit_issue"],
"edit_issue",
)
self.assertEqual(
task_capability_map.tool_required_permission("gitea_edit_issue"),
task_capability_map.required_permission("edit_issue"),
)
def test_declared_as_an_anti_stomp_mutation_task(self):
self.assertIn("edit_issue", anti_stomp_preflight.MUTATION_TASKS)
def test_edit_pr_remains_pull_request_only(self):
import inspect
params = inspect.signature(mcp_server.gitea_edit_pr).parameters
self.assertIn("pr_number", params)
self.assertNotIn("issue_number", params)
def test_edit_issue_cannot_change_state_or_labels(self):
import inspect
params = inspect.signature(mcp_server.gitea_edit_issue).parameters
self.assertEqual(
[name for name in params if name in ("title", "body")],
["title", "body"],
)
for forbidden in ("state", "labels", "assignee", "assignees", "milestone"):
self.assertNotIn(forbidden, params)
# ---------------------------------------------------------------------------
# The drift guard itself
# ---------------------------------------------------------------------------
class TestInventoryDriftRule(unittest.TestCase):
def test_missing_markers_fail_closed(self):
with self.assertRaises(ValueError):
mcp_tool_inventory.parse_documented_inventory("no markers here")
def test_documented_but_unregistered_is_drift(self):
result = mcp_tool_inventory.assess_inventory_drift(
["gitea_edit_issue", "gitea_view_issue"], ["gitea_view_issue"]
)
self.assertFalse(result["in_sync"])
self.assertEqual(result["documented_not_registered"], ["gitea_edit_issue"])
self.assertTrue(result["safe_next_action"])
def test_registered_but_undocumented_is_drift(self):
result = mcp_tool_inventory.assess_inventory_drift(
["gitea_view_issue"], ["gitea_view_issue", "gitea_edit_issue"]
)
self.assertFalse(result["in_sync"])
self.assertEqual(result["registered_not_documented"], ["gitea_edit_issue"])
def test_unsorted_inventory_is_drift(self):
result = mcp_tool_inventory.assess_inventory_drift(
["gitea_view_issue", "gitea_edit_issue"],
["gitea_view_issue", "gitea_edit_issue"],
)
self.assertFalse(result["in_sync"])
self.assertFalse(result["sorted"])
def test_module_names_are_not_treated_as_tools(self):
self.assertFalse(mcp_tool_inventory.looks_like_tool_name("gitea_auth"))
self.assertTrue(mcp_tool_inventory.looks_like_tool_name("gitea_view_issue"))
def test_unregistered_doc_reference_is_reported(self):
result = mcp_tool_inventory.assess_doc_references(
{"skills/example.md": {"gitea_edit_issue"}}, ["gitea_view_issue"]
)
self.assertFalse(result["clean"])
self.assertEqual(result["unregistered"][0]["tool"], "gitea_edit_issue")
def test_rendered_block_round_trips(self):
block = mcp_tool_inventory.render_inventory_block(
["gitea_view_issue", "gitea_edit_issue"]
)
self.assertEqual(
mcp_tool_inventory.parse_documented_inventory(block),
["gitea_edit_issue", "gitea_view_issue"],
)
class TestDocumentationMatchesRegistry(unittest.TestCase):
"""The live guard: docs and the registry must not drift apart."""
def test_documented_inventory_equals_registered_tools(self):
doc = REPO_ROOT / mcp_tool_inventory.INVENTORY_DOC_PATH
self.assertTrue(doc.exists(), f"{doc} is missing")
documented = mcp_tool_inventory.parse_documented_inventory(
doc.read_text(encoding="utf-8")
)
result = mcp_tool_inventory.assess_inventory_drift(
documented, _registered_tool_names()
)
self.assertTrue(result["in_sync"], "; ".join(result["reasons"]))
def test_every_tool_named_in_the_skills_is_registered(self):
references: dict[str, set[str]] = {}
pattern = str(REPO_ROOT / "skills" / "**" / "*.md")
paths = glob.glob(pattern, recursive=True)
self.assertTrue(paths, "no skill documents found to check")
for path in paths:
text = Path(path).read_text(encoding="utf-8")
names = mcp_tool_inventory.extract_tool_references(text)
if names:
references[str(Path(path).relative_to(REPO_ROOT))] = names
result = mcp_tool_inventory.assess_doc_references(
references, _registered_tool_names()
)
self.assertTrue(result["clean"], "; ".join(result["reasons"]))
if __name__ == "__main__":
unittest.main()
+388
View File
@@ -0,0 +1,388 @@
"""Tests for the #617 mutation-budget classifier.
Covers every acceptance criterion on issue #617:
* AC1 — the classifier distinguishes local validator rejection, capability-gate
rejection, transport failure before API, and successful server-side mutation.
* AC2 — pre-API validator failures do not consume server-side mutation budget.
* AC3 — failed attempts are still logged in the local attempt ledger.
* AC4 — the final report separately shows local failed attempts, blocked API
attempts, and successful server-side mutations.
* AC5 — the six named scenarios, including the #615 reproduction where two
local validator rejections precede one successful comment.
"""
from __future__ import annotations
import unittest
from mutation_budget_classifier import (
AMBIGUOUS_REQUIRES_READBACK,
CAPABILITY_GATE_REJECTION,
LOCAL_VALIDATOR_REJECTION,
SERVER_SIDE_MUTATION,
TRANSPORT_FAILURE_BEFORE_API,
assess_final_report_mutation_accounting,
classify_mutation_attempt,
record_attempt,
summarize_attempt_ledger,
)
# The two pre-API rejections observed on the #615 comment flow.
MISSING_LEDGER_BLOCK = {
"success": False,
"performed": False,
"api_called": False,
"reasons": ["missing [THREAD STATE LEDGER] block"],
}
MISSING_CANONICAL_STATE = {
"success": False,
"performed": False,
"api_called": False,
"reasons": ["missing ## Canonical Issue State block"],
}
# The corrected comment that actually landed as #615 comment 9137.
SUCCESSFUL_COMMENT = {
"success": True,
"performed": True,
"api_called": True,
"comment_id": 9137,
"issue_number": 615,
}
class TestAC1Classification(unittest.TestCase):
"""AC1: the four outcome classes are distinguished."""
def test_local_validator_rejection_is_its_own_class(self):
result = classify_mutation_attempt(MISSING_LEDGER_BLOCK)
self.assertEqual(result["classification"], LOCAL_VALIDATOR_REJECTION)
self.assertTrue(result["pre_api"])
def test_capability_gate_rejection_is_its_own_class(self):
result = classify_mutation_attempt(
{
"success": False,
"api_called": False,
"permission_report": {"missing_permission": "gitea.issue.comment"},
}
)
self.assertEqual(result["classification"], CAPABILITY_GATE_REJECTION)
self.assertTrue(result["pre_api"])
def test_transport_failure_before_api_is_its_own_class(self):
result = classify_mutation_attempt(
{"success": False, "api_called": False, "transport_error": "EOF"}
)
self.assertEqual(result["classification"], TRANSPORT_FAILURE_BEFORE_API)
self.assertTrue(result["pre_api"])
def test_successful_server_mutation_is_its_own_class(self):
result = classify_mutation_attempt(SUCCESSFUL_COMMENT)
self.assertEqual(result["classification"], SERVER_SIDE_MUTATION)
self.assertFalse(result["pre_api"])
def test_each_class_is_distinct(self):
classes = {
classify_mutation_attempt(payload)["classification"]
for payload in (
MISSING_LEDGER_BLOCK,
{"success": False, "api_called": False, "capability_blocked": True},
{"success": False, "api_called": False, "transport_failed": True},
SUCCESSFUL_COMMENT,
)
}
self.assertEqual(len(classes), 4)
class TestAC2BudgetAccounting(unittest.TestCase):
"""AC2: pre-API failures never consume server-side mutation budget."""
def test_missing_thread_state_ledger_not_counted_as_mutation(self):
result = classify_mutation_attempt(MISSING_LEDGER_BLOCK)
self.assertFalse(result["budget_consumed"])
self.assertIs(result["api_called"], False)
def test_missing_canonical_issue_state_not_counted_as_mutation(self):
result = classify_mutation_attempt(MISSING_CANONICAL_STATE)
self.assertFalse(result["budget_consumed"])
self.assertIs(result["api_called"], False)
def test_transport_failure_before_api_not_counted_as_mutation(self):
result = classify_mutation_attempt(
{
"success": False,
"api_called": False,
"transport_error": "connection reset",
}
)
self.assertFalse(result["budget_consumed"])
def test_capability_gate_block_not_counted_as_mutation(self):
result = classify_mutation_attempt(
{
"success": False,
"api_called": False,
"permission_report": {"missing_permission": "gitea.pr.merge"},
}
)
self.assertFalse(result["budget_consumed"])
def test_successful_comment_with_comment_id_counts_as_one_mutation(self):
result = classify_mutation_attempt(SUCCESSFUL_COMMENT)
self.assertTrue(result["budget_consumed"])
self.assertEqual(result["proof_fields"], ["comment_id"])
class TestAC2FailsClosed(unittest.TestCase):
"""AC2 must not become a loophole: ambiguity still fails closed."""
def test_api_reached_without_proof_is_ambiguous_and_consumes_budget(self):
result = classify_mutation_attempt({"success": True, "api_called": True})
self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK)
self.assertTrue(result["budget_consumed"])
self.assertTrue(result["requires_readback"])
def test_missing_api_called_signal_fails_closed(self):
result = classify_mutation_attempt({"success": False})
self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK)
self.assertTrue(result["budget_consumed"])
self.assertIsNone(result["api_called"])
def test_empty_and_none_results_fail_closed(self):
for payload in ({}, None):
result = classify_mutation_attempt(payload)
self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK)
self.assertTrue(result["budget_consumed"])
def test_success_with_proof_counts_even_when_api_called_absent(self):
result = classify_mutation_attempt({"success": True, "comment_id": 13320})
self.assertEqual(result["classification"], SERVER_SIDE_MUTATION)
self.assertTrue(result["budget_consumed"])
def test_blank_proof_field_is_not_proof(self):
result = classify_mutation_attempt(
{"success": True, "api_called": True, "merge_commit_sha": " "}
)
self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK)
class TestAC3AttemptLedger(unittest.TestCase):
"""AC3: failed attempts are still logged locally."""
def test_failed_attempts_are_recorded(self):
ledger: list[dict] = []
record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="create_issue_comment")
record_attempt(ledger, MISSING_CANONICAL_STATE, operation="create_issue_comment")
self.assertEqual(len(ledger), 2)
self.assertTrue(
all(e["classification"] == LOCAL_VALIDATOR_REJECTION for e in ledger)
)
def test_recorded_entry_carries_operation_and_timestamp(self):
ledger: list[dict] = []
entry = record_attempt(
ledger,
SUCCESSFUL_COMMENT,
operation="create_issue_comment",
timestamp="2026-07-20T18:15:04+00:00",
)
self.assertEqual(entry["operation"], "create_issue_comment")
self.assertEqual(entry["timestamp"], "2026-07-20T18:15:04+00:00")
def test_timestamp_is_generated_when_omitted(self):
ledger: list[dict] = []
entry = record_attempt(ledger, SUCCESSFUL_COMMENT)
self.assertTrue(entry["timestamp"])
class TestAC5CorrectedCommentAllowed(unittest.TestCase):
"""AC5: the #615 reproduction — two local rejections then one success."""
def _replay_615_flow(self) -> list[dict]:
ledger: list[dict] = []
record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="create_issue_comment")
record_attempt(ledger, MISSING_CANONICAL_STATE, operation="create_issue_comment")
record_attempt(ledger, SUCCESSFUL_COMMENT, operation="create_issue_comment")
return ledger
def test_corrected_comment_after_two_rejections_is_allowed(self):
summary = summarize_attempt_ledger(self._replay_615_flow())
# The regression: budget must show ONE mutation, not three attempts.
self.assertEqual(summary["successful_server_mutations"], 1)
self.assertEqual(summary["budget_consumed"], 1)
def test_all_three_attempts_remain_visible(self):
summary = summarize_attempt_ledger(self._replay_615_flow())
self.assertEqual(summary["total_attempts"], 3)
self.assertEqual(summary["local_failed_attempts"], 2)
def test_no_readback_required_for_clean_flow(self):
summary = summarize_attempt_ledger(self._replay_615_flow())
self.assertFalse(summary["requires_readback"])
class TestAC4FinalReportAccounting(unittest.TestCase):
"""AC4: the report must show the three categories, and match the ledger."""
def _mixed_ledger(self) -> list[dict]:
ledger: list[dict] = []
record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="comment")
record_attempt(ledger, MISSING_CANONICAL_STATE, operation="comment")
record_attempt(
ledger,
{"success": False, "api_called": False, "transport_error": "EOF"},
operation="comment",
)
record_attempt(
ledger,
{"success": False, "api_called": False, "capability_blocked": True},
operation="merge",
)
record_attempt(ledger, SUCCESSFUL_COMMENT, operation="comment")
return ledger
def test_summary_separates_the_three_categories(self):
summary = summarize_attempt_ledger(self._mixed_ledger())
self.assertEqual(summary["local_failed_attempts"], 2)
self.assertEqual(summary["blocked_api_attempts"], 2)
self.assertEqual(summary["successful_server_mutations"], 1)
def test_matching_report_is_valid(self):
result = assess_final_report_mutation_accounting(
{
"local_failed_attempts": 2,
"blocked_api_attempts": 2,
"successful_server_mutations": 1,
},
self._mixed_ledger(),
)
self.assertTrue(result["valid"], result["reasons"])
def test_omitted_category_fails_closed(self):
result = assess_final_report_mutation_accounting(
{"local_failed_attempts": 2, "blocked_api_attempts": 2},
self._mixed_ledger(),
)
self.assertFalse(result["valid"])
self.assertTrue(
any("successful_server_mutations" in r for r in result["reasons"])
)
def test_inflated_mutation_count_fails_closed(self):
# The #617 bug shape: claiming three mutations when only one landed.
result = assess_final_report_mutation_accounting(
{
"local_failed_attempts": 2,
"blocked_api_attempts": 2,
"successful_server_mutations": 3,
},
self._mixed_ledger(),
)
self.assertFalse(result["valid"])
self.assertTrue(
any("successful_server_mutations=3" in r for r in result["reasons"])
)
def test_ambiguous_attempt_requires_readback_proof(self):
ledger: list[dict] = []
record_attempt(ledger, {"success": True, "api_called": True}, operation="comment")
report = {
"local_failed_attempts": 0,
"blocked_api_attempts": 0,
"successful_server_mutations": 0,
}
blocked = assess_final_report_mutation_accounting(report, ledger)
self.assertFalse(blocked["valid"])
self.assertTrue(any("readback_verified" in r for r in blocked["reasons"]))
allowed = assess_final_report_mutation_accounting(
{**report, "readback_verified": True}, ledger
)
self.assertTrue(allowed["valid"], allowed["reasons"])
def test_ledger_summary_is_returned_without_raw_entries(self):
result = assess_final_report_mutation_accounting({}, self._mixed_ledger())
self.assertNotIn("entries", result["ledger_summary"])
self.assertEqual(result["ledger_summary"]["total_attempts"], 5)
class TestEmptyLedger(unittest.TestCase):
def test_empty_ledger_summarizes_to_zero(self):
summary = summarize_attempt_ledger([])
self.assertEqual(summary["total_attempts"], 0)
self.assertEqual(summary["successful_server_mutations"], 0)
self.assertFalse(summary["requires_readback"])
def test_none_ledger_is_tolerated(self):
self.assertEqual(summarize_attempt_ledger(None)["total_attempts"], 0)
if __name__ == "__main__":
unittest.main()
class TestValidatorIntegration(unittest.TestCase):
"""The classifier is wired into the shared final-report validator (AC4)."""
def _ledger_two_rejections_one_success(self) -> list[dict]:
ledger: list[dict] = []
record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="comment")
record_attempt(ledger, MISSING_CANONICAL_STATE, operation="comment")
record_attempt(ledger, SUCCESSFUL_COMMENT, operation="comment")
return ledger
def test_rule_is_noop_without_a_ledger(self):
from final_report_validator import assess_final_report_validator
result = assess_final_report_validator("some report", "review_pr")
self.assertFalse(
any(
f["rule_id"] == "shared.mutation_budget_accounting"
for f in result["findings"]
)
)
def test_report_matching_ledger_produces_no_finding(self):
from final_report_validator import assess_final_report_validator
report = (
"Local failed attempts: 2\n"
"Blocked API attempts: 0\n"
"Successful server-side mutations: 1\n"
)
result = assess_final_report_validator(
report,
"review_pr",
mutation_attempt_ledger=self._ledger_two_rejections_one_success(),
)
self.assertFalse(
any(
f["rule_id"] == "shared.mutation_budget_accounting"
for f in result["findings"]
)
)
def test_counting_rejections_as_mutations_is_blocked(self):
from final_report_validator import assess_final_report_validator
# The #617 bug: three attempts reported as three server-side mutations.
report = (
"Local failed attempts: 0\n"
"Blocked API attempts: 0\n"
"Successful server-side mutations: 3\n"
)
result = assess_final_report_validator(
report,
"review_pr",
mutation_attempt_ledger=self._ledger_two_rejections_one_success(),
)
findings = [
f
for f in result["findings"]
if f["rule_id"] == "shared.mutation_budget_accounting"
]
self.assertTrue(findings)
self.assertTrue(all(f["severity"] == "block" for f in findings))
+613
View File
@@ -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()
+555
View File
@@ -0,0 +1,555 @@
"""#780: ``status:pr-open`` must not survive a terminal PR transition.
The leak this file locks down: ``gitea_create_pr`` applied ``status:pr-open``
and no terminal path ever removed it, so a repository audit found 40 closed
issues still advertising an open PR that had long since merged or closed.
Coverage mirrors the issue's acceptance criteria: merge, close-without-merge,
supersession, already-landed reconciliation, controller closure, retry /
idempotency, unrelated-label preservation, the only-label (empty set) case,
and terminal validation of any residual label.
"""
from __future__ import annotations
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
import terminal_pr_label_cleanup as tplc
FAKE_AUTH = "token test-token"
PR_OPEN = tplc.PR_OPEN_LABEL
def _lb(name: str, lid: int) -> dict:
return {"id": lid, "name": name, "color": "000000"}
# ---------------------------------------------------------------------------
# Pure rule: planning
# ---------------------------------------------------------------------------
class TestPlanPrOpenCleanup(unittest.TestCase):
def test_removes_only_the_pr_open_label(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", PR_OPEN, "workflow-hardening"],
terminal_reason=tplc.MERGED,
)
self.assertTrue(plan["cleanup_required"])
self.assertEqual(plan["removed"], [PR_OPEN])
self.assertEqual(plan["labels_after"], ["type:bug", "workflow-hardening"])
def test_preserves_unrelated_labels_in_original_order(self):
labels = ["workflow-hardening", "type:bug", PR_OPEN, "role:author", "leases"]
plan = tplc.plan_pr_open_cleanup(labels, terminal_reason=tplc.MERGED)
self.assertEqual(
plan["labels_after"],
["workflow-hardening", "type:bug", "role:author", "leases"],
)
self.assertNotIn(PR_OPEN, plan["labels_after"])
def test_only_label_yields_empty_set(self):
plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=tplc.MERGED)
self.assertTrue(plan["cleanup_required"])
self.assertEqual(plan["labels_after"], [])
self.assertTrue(plan["empty_label_set"])
def test_absent_label_is_an_idempotent_noop(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", "status:done"], terminal_reason=tplc.RETRY_RECOVERY
)
self.assertFalse(plan["cleanup_required"])
self.assertTrue(plan["idempotent_noop"])
self.assertEqual(plan["labels_after"], ["type:bug", "status:done"])
def test_accepts_gitea_label_objects(self):
plan = tplc.plan_pr_open_cleanup(
{"labels": [{"name": PR_OPEN}, {"name": "type:bug"}]},
terminal_reason=tplc.SUPERSEDED,
)
self.assertEqual(plan["labels_after"], ["type:bug"])
def test_every_terminal_reason_is_planable(self):
for reason in tplc.TERMINAL_REASONS:
plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=reason)
self.assertEqual(plan["terminal_reason"], reason)
self.assertTrue(plan["terminal_reason_description"])
def test_unknown_terminal_reason_fails_closed(self):
with self.assertRaises(ValueError):
tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason="whenever")
def test_reason_aliases_normalize(self):
self.assertEqual(tplc.canonical_terminal_reason("merge"), tplc.MERGED)
self.assertEqual(
tplc.canonical_terminal_reason("already-landed"), tplc.ALREADY_LANDED
)
self.assertEqual(
tplc.canonical_terminal_reason("controller-closure"),
tplc.CONTROLLER_CLOSURE,
)
# ---------------------------------------------------------------------------
# Pure rule: read-after-write verification
# ---------------------------------------------------------------------------
class TestVerifyPrOpenCleanup(unittest.TestCase):
def test_verified_when_observed_matches_plan(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
)
result = tplc.verify_pr_open_cleanup(["type:bug"], plan=plan)
self.assertTrue(result["verified"])
self.assertFalse(result["residual"])
self.assertEqual(result["reasons"], [])
def test_residual_label_is_reported(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
)
result = tplc.verify_pr_open_cleanup(["type:bug", PR_OPEN], plan=plan)
self.assertFalse(result["verified"])
self.assertTrue(result["residual"])
self.assertIn(PR_OPEN, result["reasons"][0])
self.assertTrue(result["safe_next_action"])
def test_dropped_unrelated_label_is_reported(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", "leases", PR_OPEN], terminal_reason=tplc.MERGED
)
result = tplc.verify_pr_open_cleanup(["type:bug"], plan=plan)
self.assertFalse(result["verified"])
self.assertEqual(result["unexpected_removals"], ["leases"])
def test_unexpected_added_label_is_reported(self):
plan = tplc.plan_pr_open_cleanup(
["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
)
result = tplc.verify_pr_open_cleanup(["type:bug", "surprise"], plan=plan)
self.assertFalse(result["verified"])
self.assertEqual(result["unexpected_additions"], ["surprise"])
def test_empty_observed_set_verifies_for_only_label_case(self):
plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=tplc.MERGED)
result = tplc.verify_pr_open_cleanup([], plan=plan)
self.assertTrue(result["verified"])
self.assertTrue(result["empty_label_set"])
# ---------------------------------------------------------------------------
# Terminal validation
# ---------------------------------------------------------------------------
class TestDetectResidualPrOpen(unittest.TestCase):
def test_clean_repository(self):
issues = [
{"number": 1, "state": "closed", "labels": [{"name": "type:bug"}]},
{"number": 2, "state": "open", "labels": []},
]
result = tplc.detect_residual_pr_open(issues)
self.assertTrue(result["clean"])
self.assertEqual(result["residual_count"], 0)
self.assertEqual(result["checked_count"], 2)
def test_reports_each_stale_issue(self):
issues = [
{"number": 626, "state": "closed", "labels": [{"name": PR_OPEN}]},
{"number": 772, "state": "closed", "labels": [{"name": PR_OPEN}]},
{"number": 9, "state": "open", "labels": [{"name": "type:bug"}]},
]
result = tplc.detect_residual_pr_open(issues)
self.assertFalse(result["clean"])
self.assertEqual(result["residual_count"], 2)
self.assertEqual(
[entry["number"] for entry in result["residual_issues"]], [626, 772]
)
self.assertTrue(result["safe_next_action"])
def test_issue_with_a_live_open_pr_is_not_residual(self):
issues = [{"number": 42, "state": "open", "labels": [{"name": PR_OPEN}]}]
result = tplc.detect_residual_pr_open(issues, open_pr_issue_numbers=[42])
self.assertTrue(result["clean"])
self.assertEqual(result["exempt_open_pr_issues"], [42])
# ---------------------------------------------------------------------------
# Executor: one authoritative rule, with read-after-write proof
# ---------------------------------------------------------------------------
class _ExecutorHarness(unittest.TestCase):
"""Drives mcp_server.clear_pr_open_label against a fake Gitea."""
def setUp(self):
self.issue_labels: dict[int, list[str]] = {}
self.repo_labels = {
PR_OPEN: 4,
"type:bug": 1,
"workflow-hardening": 2,
"leases": 3,
"status:done": 5,
}
self.puts: list[tuple[int, list[int]]] = []
patch("mcp_server._resolve", return_value=("h", "o", "r")).start()
patch("mcp_server._auth", return_value=FAKE_AUTH).start()
patch(
"mcp_server.repo_api_url",
return_value="https://gitea.example/api/v1/repos/o/r",
).start()
patch("gitea_audit.audit_enabled", return_value=False).start()
patch("mcp_server.api_request", side_effect=self._api).start()
# api_get_all resolves api_request inside gitea_auth, so patching the
# mcp_server binding alone would let the label inventory hit the network.
patch("mcp_server.api_get_all", side_effect=self._api_get_all).start()
self.addCleanup(patch.stopall)
def _api_get_all(self, url, auth, **_kwargs):
if "/labels" in url:
return [_lb(name, lid) for name, lid in self.repo_labels.items()]
raise AssertionError(f"unexpected paginated GET: {url}")
def _api(self, method, url, auth, payload=None):
if method == "GET" and "/issues/" in url:
num = int(url.rsplit("/issues/", 1)[1].split("?")[0])
return {
"number": num,
"labels": [
{"name": n, "id": self.repo_labels[n]}
for n in self.issue_labels.get(num, [])
],
}
if method == "PUT" and url.endswith("/labels"):
num = int(url.rsplit("/issues/", 1)[1].split("/")[0])
ids = payload["labels"]
by_id = {lid: name for name, lid in self.repo_labels.items()}
names = [by_id[i] for i in ids]
self.puts.append((num, ids))
self.issue_labels[num] = names
return [_lb(n, self.repo_labels[n]) for n in names]
raise AssertionError(f"unexpected API call: {method} {url}")
def _clear(self, numbers, reason=tplc.MERGED):
return mcp_server.clear_pr_open_label(
numbers, "prgs", None, None, None, terminal_reason=reason
)
class TestClearPrOpenLabel(_ExecutorHarness):
def test_removes_label_and_preserves_the_rest(self):
self.issue_labels[780] = ["type:bug", PR_OPEN, "workflow-hardening"]
summary = self._clear([780])
self.assertTrue(summary["clean"])
self.assertEqual(summary["removed"], [780])
self.assertEqual(
self.issue_labels[780], ["type:bug", "workflow-hardening"]
)
def test_only_label_results_in_empty_set(self):
self.issue_labels[626] = [PR_OPEN]
summary = self._clear([626])
self.assertTrue(summary["clean"])
self.assertEqual(self.issue_labels[626], [])
self.assertEqual(self.puts, [(626, [])])
self.assertTrue(summary["results"][0]["empty_label_set"])
def test_read_after_write_proof_is_returned(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
summary = self._clear([780])
entry = summary["results"][0]
self.assertTrue(entry["verified"])
self.assertEqual(entry["labels_before"], ["type:bug", PR_OPEN])
self.assertEqual(entry["labels_after"], ["type:bug"])
self.assertEqual(entry["verification"]["observed_labels"], ["type:bug"])
def test_repeated_cleanup_is_harmless(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
first = self._clear([780])
second = self._clear([780], reason=tplc.RETRY_RECOVERY)
third = self._clear([780], reason=tplc.RETRY_RECOVERY)
self.assertTrue(first["clean"] and second["clean"] and third["clean"])
self.assertEqual(second["already_absent"], [780])
self.assertEqual(third["already_absent"], [780])
# Exactly one mutation across three calls.
self.assertEqual(len(self.puts), 1)
self.assertEqual(self.issue_labels[780], ["type:bug"])
def test_noop_path_never_reads_the_label_inventory(self):
self.issue_labels[780] = ["type:bug"]
with patch("mcp_server._repo_label_id_map") as mock_map:
summary = self._clear([780])
self.assertTrue(summary["clean"])
mock_map.assert_not_called()
def test_duplicate_issue_numbers_are_collapsed(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
summary = self._clear([780, 780, "780"])
self.assertEqual(summary["checked"], [780])
self.assertEqual(len(self.puts), 1)
def test_no_issue_numbers_is_a_clean_noop(self):
summary = self._clear([])
self.assertTrue(summary["clean"])
self.assertEqual(summary["checked"], [])
def test_failed_mutation_is_reported_not_swallowed(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
def boom(*_a, **_kw):
raise RuntimeError("gitea exploded")
with patch("mcp_server._put_issue_label_names", side_effect=boom):
summary = self._clear([780])
self.assertFalse(summary["clean"])
self.assertEqual(summary["failed"], [780])
self.assertTrue(summary["safe_next_action"])
self.assertIn(PR_OPEN, self.issue_labels[780])
def test_residual_label_after_write_fails_verification(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
real_api = self._api
# Simulate a write that reports success but leaves the label behind.
def sticky(method, url, auth, payload=None):
if method == "PUT" and url.endswith("/labels"):
num = int(url.rsplit("/issues/", 1)[1].split("/")[0])
self.puts.append((num, payload["labels"]))
return [_lb("type:bug", 1), _lb(PR_OPEN, 4)]
return real_api(method, url, auth, payload)
with patch("mcp_server.api_request", side_effect=sticky):
summary = self._clear([780])
self.assertFalse(summary["clean"])
self.assertEqual(summary["failed"], [780])
# ---------------------------------------------------------------------------
# Terminal workflow paths
# ---------------------------------------------------------------------------
class TestTerminalPathsUseTheSharedRule(_ExecutorHarness):
"""Merge, close-without-merge, supersession and already-landed."""
def test_merge_path_clears_the_label_for_linked_issues(self):
self.issue_labels[780] = ["type:bug", PR_OPEN]
merged_pr = {
"title": "fix: terminal label cleanup",
"body": "Closes #780",
"head": {"ref": "fix/issue-780-terminal-pr-open-label-cleanup"},
}
with patch(
"mcp_server.release_in_progress_label", return_value={780: "released"}
):
result = mcp_server.cleanup_in_progress_for_pr(
merged_pr, "prgs", None, None, None, terminal_reason=tplc.MERGED
)
cleanup = result["pr_open_label_cleanup"]
self.assertTrue(cleanup["clean"])
self.assertEqual(cleanup["terminal_reason"], tplc.MERGED)
self.assertEqual(self.issue_labels[780], ["type:bug"])
def test_close_without_merge_clears_the_label(self):
self.issue_labels[781] = ["type:bug", PR_OPEN, "leases"]
closed_pr = {
"title": "chore: abandoned",
"body": "Closes #781",
"head": {"ref": "chore/issue-781-abandoned"},
}
with patch(
"mcp_server.release_in_progress_label", return_value={781: "released"}
):
result = mcp_server.cleanup_in_progress_for_pr(
closed_pr,
"prgs",
None,
None,
None,
terminal_reason=tplc.CLOSED_WITHOUT_MERGE,
)
cleanup = result["pr_open_label_cleanup"]
self.assertTrue(cleanup["clean"])
self.assertEqual(cleanup["terminal_reason"], tplc.CLOSED_WITHOUT_MERGE)
self.assertEqual(self.issue_labels[781], ["type:bug", "leases"])
def test_pr_without_linked_issue_reports_an_empty_cleanup(self):
pr = {"title": "chore: no link", "body": "", "head": {"ref": "chore/none"}}
result = mcp_server.cleanup_in_progress_for_pr(
pr, "prgs", None, None, None, terminal_reason=tplc.MERGED
)
self.assertEqual(result["cleanup_status"], "no linked issue found")
self.assertTrue(result["pr_open_label_cleanup"]["clean"])
self.assertEqual(result["pr_open_label_cleanup"]["checked"], [])
def test_supersession_reason_is_recorded(self):
self.issue_labels[600] = [PR_OPEN, "type:bug"]
summary = self._clear([600], reason=tplc.SUPERSEDED)
self.assertTrue(summary["clean"])
self.assertEqual(summary["terminal_reason"], tplc.SUPERSEDED)
self.assertEqual(self.issue_labels[600], ["type:bug"])
def test_already_landed_reconciliation_reason_is_recorded(self):
self.issue_labels[601] = [PR_OPEN]
summary = self._clear([601], reason=tplc.ALREADY_LANDED)
self.assertTrue(summary["clean"])
self.assertEqual(summary["terminal_reason"], tplc.ALREADY_LANDED)
self.assertEqual(self.issue_labels[601], [])
def test_issue_780_regression_stale_label_survived_every_terminal_path(self):
"""Regression for the observed leak.
Before the fix each terminal path finished without touching
``status:pr-open``, so the audit found closed issues still carrying it.
Every path now routes through the one shared rule and leaves nothing
behind — while preserving each issue's other labels.
"""
stale = {
626: (["type:bug", PR_OPEN], tplc.CONTROLLER_CLOSURE),
772: (["workflow-hardening", PR_OPEN], tplc.MERGED),
768: ([PR_OPEN], tplc.CLOSED_WITHOUT_MERGE),
758: (["leases", PR_OPEN, "type:bug"], tplc.SUPERSEDED),
755: (["status:done", PR_OPEN], tplc.ALREADY_LANDED),
}
for number, (labels, _reason) in stale.items():
self.issue_labels[number] = list(labels)
for number, (_labels, reason) in stale.items():
summary = self._clear([number], reason=reason)
self.assertTrue(summary["clean"], msg=f"issue #{number}")
audit = tplc.detect_residual_pr_open(
[
{"number": num, "state": "closed", "labels": names}
for num, names in self.issue_labels.items()
]
)
self.assertTrue(audit["clean"])
self.assertEqual(audit["residual_count"], 0)
# Unrelated labels survived every path.
self.assertEqual(self.issue_labels[626], ["type:bug"])
self.assertEqual(self.issue_labels[772], ["workflow-hardening"])
self.assertEqual(self.issue_labels[768], [])
self.assertEqual(self.issue_labels[758], ["leases", "type:bug"])
self.assertEqual(self.issue_labels[755], ["status:done"])
# ---------------------------------------------------------------------------
# Controller closure
# ---------------------------------------------------------------------------
class TestControllerClosure(unittest.TestCase):
def test_close_issue_clears_label_before_closing_and_validates(self):
calls: list[str] = []
def fake_clear(numbers, *_a, **kwargs):
calls.append(f"clear:{kwargs['terminal_reason']}")
return {
"label": PR_OPEN,
"clean": True,
"checked": list(numbers),
"removed": list(numbers),
"already_absent": [],
"failed": [],
"results": [],
"reasons": [],
"safe_next_action": "",
"terminal_reason": kwargs["terminal_reason"],
}
def fake_api(method, url, auth, payload=None):
if method == "PATCH":
calls.append("patch:closed")
return {"state": "closed"}
return {"labels": [{"name": "type:bug"}]}
with patch("mcp_server.clear_pr_open_label", side_effect=fake_clear), \
patch("mcp_server.api_request", side_effect=fake_api), \
patch("mcp_server._profile_permission_block", return_value=None), \
patch("mcp_server.verify_preflight_purity", return_value=None), \
patch("mcp_server.release_in_progress_label", return_value={}), \
patch("mcp_server._resolve", return_value=("h", "o", "r")), \
patch("mcp_server._auth", return_value=FAKE_AUTH), \
patch("gitea_audit.audit_enabled", return_value=False):
result = mcp_server.gitea_close_issue(issue_number=780, remote="prgs")
self.assertTrue(result["success"])
# Cleanup precedes the state change: closing first would bake in the leak.
self.assertEqual(calls[0], f"clear:{tplc.CONTROLLER_CLOSURE}")
self.assertIn("patch:closed", calls)
self.assertTrue(result["terminal_label_validation"]["clean"])
def test_close_issue_fails_closed_when_cleanup_cannot_complete(self):
def fake_clear(numbers, *_a, **kwargs):
return {
"label": PR_OPEN,
"clean": False,
"checked": list(numbers),
"removed": [],
"already_absent": [],
"failed": list(numbers),
"results": [],
"reasons": ["label replacement failed: boom"],
"safe_next_action": "retry",
"terminal_reason": kwargs["terminal_reason"],
}
def fail_on_patch(method, url, auth, payload=None):
if method == "PATCH":
raise AssertionError("issue must not be closed when cleanup failed")
return {}
with patch("mcp_server.clear_pr_open_label", side_effect=fake_clear), \
patch("mcp_server.api_request", side_effect=fail_on_patch), \
patch("mcp_server._profile_permission_block", return_value=None), \
patch("mcp_server.verify_preflight_purity", return_value=None), \
patch("mcp_server._resolve", return_value=("h", "o", "r")), \
patch("mcp_server._auth", return_value=FAKE_AUTH), \
patch("gitea_audit.audit_enabled", return_value=False):
result = mcp_server.gitea_close_issue(issue_number=780, remote="prgs")
self.assertFalse(result["success"])
self.assertTrue(result["blocked"])
self.assertFalse(result["performed"])
self.assertIn("#780", result["message"])
self.assertTrue(result["safe_next_action"])
# ---------------------------------------------------------------------------
# Capability wiring
# ---------------------------------------------------------------------------
class TestCapabilityWiring(unittest.TestCase):
def test_task_is_registered_with_label_authority(self):
import task_capability_map
self.assertEqual(
task_capability_map.required_permission("cleanup_terminal_pr_labels"),
"gitea.issue.comment",
)
self.assertEqual(
task_capability_map.required_role("cleanup_terminal_pr_labels"),
"author",
)
self.assertEqual(
task_capability_map.tool_required_permission(
"gitea_cleanup_terminal_pr_labels"
),
"gitea.issue.comment",
)
def test_recovery_tool_rejects_an_unknown_reason_without_mutating(self):
with patch("mcp_server._profile_permission_block", return_value=None), \
patch("mcp_server.verify_preflight_purity", return_value=None), \
patch("mcp_server.clear_pr_open_label") as mock_clear:
result = mcp_server.gitea_cleanup_terminal_pr_labels(
issue_numbers=[780], terminal_reason="sometime", remote="prgs"
)
self.assertFalse(result["success"])
self.assertFalse(result["clean"])
mock_clear.assert_not_called()
if __name__ == "__main__":
unittest.main()
+40 -4
View File
@@ -14,7 +14,7 @@ Design rules:
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Iterable, Sequence
from typing import Any, Iterable, Mapping, Sequence
from allocator_service import (
ROLE_AUTHOR,
@@ -22,7 +22,11 @@ from allocator_service import (
ROLE_MERGER,
ROLE_RECONCILER,
ROLE_REVIEWER,
OWNERSHIP_FOREIGN,
OWNERSHIP_UNKNOWN,
SKIP_CLAIMED_BY_OTHER_SESSION,
WorkCandidate,
classify_claim_ownership,
classify_skip,
expected_role_for_candidate,
sort_candidates,
@@ -265,9 +269,13 @@ def _entry_for_candidate(
c: WorkCandidate,
*,
terminal_pr: int | None,
claim_ownership: str | None = None,
) -> QueueEntry:
expected = expected_role_for_candidate(c)
badges = _badges_for_candidate(c, terminal_pr=terminal_pr)
claimed_by_other = claim_ownership in (OWNERSHIP_FOREIGN, OWNERSHIP_UNKNOWN)
if claimed_by_other and "claimed" not in badges:
badges = tuple(list(badges) + ["claimed"])
safe_roles: list[str] = []
block_reason: str | None = None
@@ -276,6 +284,12 @@ def _entry_for_candidate(
block_reason = "status blocked"
elif c.dependency_unmet:
block_reason = c.dependency_reason or "unmet dependency"
elif claimed_by_other:
# #765: never advertise another controller's active task as safe work.
block_reason = (
f"{SKIP_CLAIMED_BY_OTHER_SESSION}: active lease held by another "
"controller"
)
elif c.already_claimed_elsewhere:
block_reason = "already claimed elsewhere"
elif c.kind == "pr" and not (c.head_sha or "").strip():
@@ -291,7 +305,12 @@ def _entry_for_candidate(
if block_reason is None:
# Safe only for the expected role, and only when classify_skip agrees.
skip = classify_skip(c, role=expected, terminal_pr=terminal_pr)
skip = classify_skip(
c,
role=expected,
terminal_pr=terminal_pr,
claim_ownership=claim_ownership,
)
if skip is None:
safe_roles.append(expected)
else:
@@ -450,8 +469,16 @@ def build_workflow_dashboard(
terminal_lock: dict[str, Any] | None = None,
inventory_complete: bool = True,
inventory_reasons: Sequence[str] | None = None,
claims: Mapping[tuple[str, int], dict[str, Any]] | None = None,
session_id: str | None = None,
controller_instance_id: str | None = None,
) -> DashboardSnapshot:
"""Build a full dashboard snapshot from injected inventory (pure)."""
"""Build a full dashboard snapshot from injected inventory (pure).
*claims* (#765) maps ``(kind, number)`` to the live lease holding that work
item. Items claimed by a different controller are never presented as safe
next work for this one.
"""
reasons = [str(r) for r in (inventory_reasons or ()) if str(r).strip()]
if not inventory_complete:
reasons.append(
@@ -461,7 +488,16 @@ def build_workflow_dashboard(
ranked = sort_candidates(list(candidates))
entries = [
_entry_for_candidate(c, terminal_pr=terminal_pr) for c in ranked
_entry_for_candidate(
c,
terminal_pr=terminal_pr,
claim_ownership=classify_claim_ownership(
(claims or {}).get((c.kind, int(c.number))),
session_id=session_id,
controller_instance_id=controller_instance_id,
),
)
for c in ranked
]
open_prs = [e for e in entries if e.kind == "pr"]