feat: expose explicit adoption proof in gitea_lock_issue response (Closes #477) #481

Merged
sysadmin merged 1 commits from feat/issue-477-lock-adoption-proof into master 2026-07-08 02:31:47 -05:00
5 changed files with 263 additions and 0 deletions
+11
View File
@@ -303,6 +303,17 @@ shared state and manual writes can clobber another session's live lease. Use
3. Operator override only when explicitly authorized — record 3. Operator override only when explicitly authorized — record
`External-state mutations` and `operator override proof` in the final report. `External-state mutations` and `operator override proof` in the final report.
**Adoption proof in the live lock response (#477):** when `gitea_lock_issue`
adopts an existing own branch, the response carries an `adoption` block with
citable fields — `adoption_decision` (`ADOPT`), `adopted` (`true`),
`adopted_branch`, `adopted_branch_head`, `matcher_summary` (boundary-safe reason
the branch qualified), `competing_branch_check`, and `safe_next_action`. A normal
lock instead returns an `adoption_check` block with `adoption_decision`
(`NO_MATCH`) and `adopted: false`, so a non-adoption response can never be misread
as claiming adoption. Recovery reports should quote the live lock response
`adoption`/`adoption_check` block directly instead of inferring adoption from
separate offline checks.
`gitea_create_pr` rejects lock files that lack sanctioned `lock_provenance` `gitea_create_pr` rejects lock files that lack sanctioned `lock_provenance`
metadata. Final-report validation blocks handoffs that hide lock read/write/delete metadata. Final-report validation blocks handoffs that hide lock read/write/delete
under `External-state mutations: none` or mix author PR creation with reviewer under `External-state mutations: none` or mix author PR creation with reviewer
+8
View File
@@ -1427,6 +1427,14 @@ def gitea_lock_issue(
f"Adopted existing branch '{branch_name}' and locked issue " f"Adopted existing branch '{branch_name}' and locked issue "
f"#{issue_number} for recovery (fail-closed check complete)." f"#{issue_number} for recovery (fail-closed check complete)."
) )
else:
# #477 AC2: normal (no-adoption) lock responses carry explicit,
# adoption-free proof metadata so they stay clear and cannot be
# misread as claiming a branch was adopted.
result["adoption_check"] = issue_lock_adoption.build_non_adoption_lock_proof(
issue_number=issue_number,
branch_name=branch_name,
)
if agent_artifacts: if agent_artifacts:
result["warnings"] = [ result["warnings"] = [
"Agent temp artifacts at repo root (delete before implementation): " "Agent temp artifacts at repo root (delete before implementation): "
+116
View File
@@ -20,6 +20,43 @@ ADOPT = "adopt_existing_branch"
BLOCK_COMPETING = "block_competing_branch" BLOCK_COMPETING = "block_competing_branch"
NO_MATCH = "no_matching_branch" NO_MATCH = "no_matching_branch"
# Citable decision labels aligned with the ``assess_own_branch_adoption``
# outcomes, surfaced verbatim in the live ``gitea_lock_issue`` response so
# recovery reports (#473-style) can quote the lock tool output directly
# instead of inferring adoption from separate offline checks (#477).
DECISION_LABELS = {
ADOPT: "ADOPT",
BLOCK_COMPETING: "BLOCK_COMPETING",
NO_MATCH: "NO_MATCH",
}
_SAFE_NEXT_ACTIONS = {
ADOPT: (
"Own existing branch adopted for lock recovery; proceed to "
"gitea_create_pr for this issue and cite this adoption proof."
),
BLOCK_COMPETING: (
"Competing same-issue branch(es) exist; resolve branch ownership "
"before locking. No adoption performed (fail closed)."
),
NO_MATCH: (
"No existing branch carries this issue marker; normal lock path "
"applied. No adoption performed."
),
}
def decision_label(outcome: str) -> str:
"""Map an ``assess_own_branch_adoption`` outcome to its citable label."""
return DECISION_LABELS.get(outcome, "UNKNOWN")
def safe_next_action(outcome: str) -> str:
"""Return the safe next action string for an adoption *outcome*."""
return _SAFE_NEXT_ACTIONS.get(
outcome, "Unknown adoption outcome; treat as fail closed."
)
def _branch_name(entry) -> str: def _branch_name(entry) -> str:
if isinstance(entry, dict): if isinstance(entry, dict):
@@ -128,6 +165,42 @@ def assess_own_branch_adoption(
} }
def _matcher_summary(issue_number: int, assessment: dict) -> str:
"""Explain, citably, why the assessed branch did or did not qualify.
Names the numeric word-boundary rule so reports can show that
``issue-42`` was not matched inside ``issue-420`` (#440 / #477 AC3).
"""
outcome = assessment.get("outcome")
matched = assessment.get("matched_branch")
competing = assessment.get("competing_branches") or []
if outcome == ADOPT and matched:
return (
f"branch '{matched}' exactly matches the issue-{int(issue_number)} "
f"marker (numeric word-boundary; 'issue-{int(issue_number)}' is not "
f"matched inside 'issue-{int(issue_number)}0')"
)
if outcome == BLOCK_COMPETING:
return (
f"competing same-issue branch(es) {competing} carry the "
f"issue-{int(issue_number)} marker but are not the requested "
f"branch; ownership is ambiguous (fail closed)"
)
return (
f"no existing branch carries the issue-{int(issue_number)} marker "
f"under the numeric word-boundary rule"
)
def _competing_branch_check(assessment: dict) -> dict:
"""Structured competing-branch verdict for the proof block."""
competing = list(assessment.get("competing_branches") or [])
return {
"result": "blocked" if competing else "clear",
"competing_branches": competing,
}
def build_adoption_proof( def build_adoption_proof(
*, *,
issue_number: int, issue_number: int,
@@ -143,7 +216,18 @@ def build_adoption_proof(
Requirement #4: adoption results must carry issue number, branch name, Requirement #4: adoption results must carry issue number, branch name,
branch head commit, adoption reason, no-existing-PR proof, no-competing- branch head commit, adoption reason, no-existing-PR proof, no-competing-
live-lock proof, and lock file path/status. live-lock proof, and lock file path/status.
#477: additionally surface explicit, citable adoption-proof fields tied to
the ``assess_own_branch_adoption`` outcome (``adoption_decision``,
``adopted``, ``adopted_branch``, ``adopted_branch_head``,
``matcher_summary``, ``competing_branch_check``, ``safe_next_action``) so a
recovery session can quote the live lock response directly. The explicit
fields are populated for any outcome; ``adopted_branch`` /
``adopted_branch_head`` are set only when the outcome is ADOPT so a
non-adoption proof can never be misread as claiming adoption.
""" """
outcome = assessment.get("outcome")
adopted = outcome == ADOPT
return { return {
"issue_number": issue_number, "issue_number": issue_number,
"branch_name": branch_name, "branch_name": branch_name,
@@ -153,4 +237,36 @@ def build_adoption_proof(
"no_competing_live_lock_proof": bool(competing_lock_checked), "no_competing_live_lock_proof": bool(competing_lock_checked),
"lock_file_path": lock_file_path, "lock_file_path": lock_file_path,
"lock_file_status": lock_file_status, "lock_file_status": lock_file_status,
# Explicit citable fields (#477).
"adoption_decision": decision_label(outcome),
"adopted": adopted,
"adopted_branch": branch_name if adopted else None,
"adopted_branch_head": assessment.get("matched_head_sha") if adopted else None,
"matcher_summary": _matcher_summary(issue_number, assessment),
"competing_branch_check": _competing_branch_check(assessment),
"safe_next_action": safe_next_action(outcome),
}
def build_non_adoption_lock_proof(*, issue_number: int, branch_name: str) -> dict:
"""Safe, adoption-free proof metadata for a normal (NO_MATCH) lock.
Requirement #477 AC2: non-adoption lock responses must stay clear and must
not imply adoption. This returns explicit ``adopted: False`` metadata with
the ``NO_MATCH`` decision so a normal lock response can carry citable proof
without ever asserting a branch was adopted.
"""
return {
"issue_number": issue_number,
"branch_name": branch_name,
"adoption_decision": DECISION_LABELS[NO_MATCH],
"adopted": False,
"adopted_branch": None,
"adopted_branch_head": None,
"matcher_summary": (
f"no existing branch carries the issue-{int(issue_number)} marker; "
f"normal lock path (no adoption)"
),
"competing_branch_check": {"result": "clear", "competing_branches": []},
"safe_next_action": safe_next_action(NO_MATCH),
} }
+89
View File
@@ -11,6 +11,7 @@ from issue_lock_adoption import ( # noqa: E402
NO_MATCH, NO_MATCH,
assess_own_branch_adoption, assess_own_branch_adoption,
build_adoption_proof, build_adoption_proof,
build_non_adoption_lock_proof,
) )
REQ = "feat/issue-420-server-code-parity" REQ = "feat/issue-420-server-code-parity"
@@ -134,5 +135,93 @@ class TestBuildAdoptionProof(unittest.TestCase):
self.assertTrue(proof["no_competing_live_lock_proof"]) self.assertTrue(proof["no_competing_live_lock_proof"])
class TestExplicitAdoptionProofFields(unittest.TestCase):
"""#477: explicit, citable adoption-proof fields for all outcomes."""
def _proof(self, assessment, branch):
return build_adoption_proof(
issue_number=420,
branch_name=branch,
assessment=assessment,
open_pr_checked=True,
competing_lock_checked=True,
lock_file_path="/tmp/example-lock.json",
lock_file_status="written",
)
def test_adopt_proof_exposes_explicit_fields(self):
assessment = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
)
proof = self._proof(assessment, REQ)
self.assertEqual(proof["adoption_decision"], "ADOPT")
self.assertTrue(proof["adopted"])
self.assertEqual(proof["adopted_branch"], REQ)
self.assertEqual(proof["adopted_branch_head"], "934688a")
self.assertEqual(proof["competing_branch_check"]["result"], "clear")
self.assertEqual(proof["competing_branch_check"]["competing_branches"], [])
self.assertIn("gitea_create_pr", proof["safe_next_action"])
self.assertIn("exactly matches", proof["matcher_summary"])
def test_block_proof_reports_competing_and_does_not_claim_adoption(self):
assessment = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": "feat/issue-420-rogue"}],
)
proof = self._proof(assessment, REQ)
self.assertEqual(proof["adoption_decision"], "BLOCK_COMPETING")
self.assertFalse(proof["adopted"])
self.assertIsNone(proof["adopted_branch"])
self.assertIsNone(proof["adopted_branch_head"])
self.assertEqual(proof["competing_branch_check"]["result"], "blocked")
self.assertIn(
"feat/issue-420-rogue",
proof["competing_branch_check"]["competing_branches"],
)
self.assertIn("fail closed", proof["safe_next_action"])
def test_no_match_proof_does_not_claim_adoption(self):
assessment = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": "feat/issue-999-unrelated"}],
)
proof = self._proof(assessment, REQ)
self.assertEqual(proof["adoption_decision"], "NO_MATCH")
self.assertFalse(proof["adopted"])
self.assertIsNone(proof["adopted_branch"])
self.assertEqual(proof["competing_branch_check"]["result"], "clear")
def test_substring_collision_stays_boundary_safe(self):
# issue-42 must not adopt/claim against an issue-420 branch (#440/#477).
own = "feat/issue-42-widget"
assessment = assess_own_branch_adoption(
issue_number=42,
requested_branch=own,
existing_branches=[
{"name": own, "commit_sha": "abc1234"},
{"name": "feat/issue-420-server-code-parity"},
],
)
proof = self._proof(assessment, own)
self.assertEqual(proof["adoption_decision"], "ADOPT")
self.assertEqual(proof["adopted_branch"], own)
self.assertEqual(proof["competing_branch_check"]["competing_branches"], [])
def test_non_adoption_lock_proof_is_adoption_free(self):
proof = build_non_adoption_lock_proof(
issue_number=196, branch_name="feat/issue-196-mutations"
)
self.assertEqual(proof["adoption_decision"], "NO_MATCH")
self.assertFalse(proof["adopted"])
self.assertIsNone(proof["adopted_branch"])
self.assertIsNone(proof["adopted_branch_head"])
self.assertEqual(proof["competing_branch_check"]["result"], "clear")
self.assertIn("no adoption", proof["safe_next_action"].lower())
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+39
View File
@@ -3325,6 +3325,45 @@ class TestIssueLocking(unittest.TestCase):
self.assertIn("adoption", res) self.assertIn("adoption", res)
self.assertEqual(res["adoption"]["branch_head_commit"], "abc123") self.assertEqual(res["adoption"]["branch_head_commit"], "abc123")
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_adoption_response_has_explicit_proof(self, _auth, mock_api, _git_state):
# #477 AC1: the live lock response must carry citable adoption proof.
branch = "feat/issue-196-mutations"
self.mock_dup_fetcher.return_value = ([], [branch], {"status": "not_claimed"})
mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}]
res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs")
proof = res["adoption"]
self.assertEqual(proof["adoption_decision"], "ADOPT")
self.assertTrue(proof["adopted"])
self.assertEqual(proof["adopted_branch"], branch)
self.assertEqual(proof["adopted_branch_head"], "abc123")
self.assertEqual(proof["competing_branch_check"]["result"], "clear")
self.assertIn("gitea_create_pr", proof["safe_next_action"])
self.assertIn("196", proof["matcher_summary"])
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_no_match_response_does_not_claim_adoption(self, _auth, _api, _git_state):
# #477 AC2/AC3: a normal (NO_MATCH) lock must carry adoption-free proof
# and must NOT expose an ``adoption`` block.
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertTrue(res["success"])
self.assertNotIn("adoption", res)
check = res["adoption_check"]
self.assertEqual(check["adoption_decision"], "NO_MATCH")
self.assertFalse(check["adopted"])
self.assertIsNone(check["adopted_branch"])
self.assertEqual(check["competing_branch_check"]["result"], "clear")
@patch( @patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state", "mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(), return_value=_clean_master_git_state_for_lock(),