From 795f54404798a1eb70be74419a6260153e423652 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 16:18:42 -0400 Subject: [PATCH] feat: block manual issue-lock seeding and require lock disclosure (#447) Add lock_provenance metadata on gitea_lock_issue writes and fail closed at gitea_create_pr when provenance is missing. Final-report validation now requires explicit External-state mutations disclosure for issue-lock read/write/delete and blocks mixed author PR creation with reviewer approval. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/llm-workflow-runbooks.md | 15 + final_report_validator.py | 62 ++++ gitea_mcp_server.py | 13 + issue_lock_provenance.py | 269 ++++++++++++++++++ .../workflows/work-issue.md | 6 + tests/test_commit_payloads.py | 14 + tests/test_issue_lock_provenance.py | 130 +++++++++ tests/test_mcp_server.py | 36 +++ 8 files changed, 545 insertions(+) create mode 100644 issue_lock_provenance.py create mode 100644 tests/test_issue_lock_provenance.py diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index f3d82e3..09b4d5c 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -281,6 +281,21 @@ same-issue/same-operation lease blocks duplicate work. An expired lease still blocks takeover until a recovery review records why the prior work is abandoned, completed, or unsafe to continue. +**Issue-lock recovery (#447):** Do not manually seed, restore, or delete +`/tmp/gitea_issue_lock.json` as a normal recovery path. That file is global +shared state and manual writes can clobber another session's live lease. Use +`sanctioned recovery` instead: + +1. `gitea_lock_issue` on a clean `branches/` worktree (normal path). +2. Own-branch adoption via #442 when the issue's exact branch is already pushed. +3. Operator override only when explicitly authorized — record + `External-state mutations` and `operator override proof` in the final report. + +`gitea_create_pr` rejects lock files that lack sanctioned `lock_provenance` +metadata. Final-report validation blocks handoffs that hide lock read/write/delete +under `External-state mutations: none` or mix author PR creation with reviewer +approval in one run. See also #438 (global lock redesign). + Remote branches matching the issue number are also treated as active work unless the recovery review proves the branch is abandoned or superseded. Never delete or clean up a branch when it has an active lease, dirty worktree, open PR, or is diff --git a/final_report_validator.py b/final_report_validator.py index 3fbac0f..bbc4b6c 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -11,6 +11,7 @@ import inspect import re from typing import Any, Callable +import issue_lock_provenance from review_proofs import ( HANDOFF_HEADING, assess_controller_handoff, @@ -870,6 +871,54 @@ def _rule_reviewer_mutation_ledger( ) +def _rule_shared_issue_lock_external_state(report_text: str) -> list[dict[str, str]]: + result = issue_lock_provenance.assess_issue_lock_external_state_report(report_text) + if result.get("proven"): + return [] + return _findings_from_reasons( + "shared.issue_lock_external_state", + result.get("reasons") or [], + field="External-state mutations", + severity="block", + safe_next_action=( + "disclose gitea_issue_lock.json read/write/delete under " + "External-state mutations; never claim none after lock seeding" + ), + ) + + +def _rule_shared_manual_lock_pr_override(report_text: str) -> list[dict[str, str]]: + result = issue_lock_provenance.assess_manual_lock_pr_without_override(report_text) + if result.get("proven"): + return [] + return _findings_from_reasons( + "shared.manual_lock_pr_override", + result.get("reasons") or [], + field="External-state mutations", + severity="block", + safe_next_action=( + "use gitea_lock_issue or #442 adoption instead of manual lock seeding; " + "if operator override was authorized, cite override proof" + ), + ) + + +def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, str]]: + result = issue_lock_provenance.assess_author_reviewer_same_run_report(report_text) + if result.get("proven"): + return [] + return _findings_from_reasons( + "shared.author_reviewer_same_run", + result.get("reasons") or [], + field="Review mutations", + severity="block", + safe_next_action=( + "split author PR creation and reviewer approval across separate " + "sessions and handoffs" + ), + ) + + def _rule_reviewer_review_mutation( report_text: str, *, @@ -889,10 +938,17 @@ def _rule_reviewer_review_mutation( ) +_SHARED_ISSUE_LOCK_RULES = ( + _rule_shared_issue_lock_external_state, + _rule_shared_manual_lock_pr_override, + _rule_shared_author_reviewer_same_run, +) + _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { "review_pr": [ _rule_shared_controller_handoff, _rule_shared_email_disclosure, + *_SHARED_ISSUE_LOCK_RULES, _rule_reviewer_legacy_workspace_mutations, _rule_reviewer_vague_mutations_none, _rule_reviewer_mutation_categories, @@ -913,6 +969,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { "reconcile_already_landed": [ _rule_reconcile_controller_handoff, _rule_shared_email_disclosure, + *_SHARED_ISSUE_LOCK_RULES, _rule_reconcile_stale_author_fields, _rule_reconcile_eligible_reviewed, _rule_reconcile_linked_issue_live, @@ -924,25 +981,30 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { "author_issue": [ _rule_shared_controller_handoff, _rule_shared_email_disclosure, + *_SHARED_ISSUE_LOCK_RULES, _rule_reviewer_vague_mutations_none, ], "work_issue": [ _rule_shared_controller_handoff, _rule_shared_email_disclosure, + *_SHARED_ISSUE_LOCK_RULES, _rule_reviewer_vague_mutations_none, ], "issue_filing": [ _rule_shared_controller_handoff, _rule_shared_email_disclosure, + *_SHARED_ISSUE_LOCK_RULES, ], "inventory": [ _rule_shared_controller_handoff, _rule_shared_email_disclosure, + *_SHARED_ISSUE_LOCK_RULES, _rule_reconcile_pagination_proof, ], "issue_selection": [ _rule_shared_controller_handoff, _rule_shared_email_disclosure, + *_SHARED_ISSUE_LOCK_RULES, ], } diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index beb6084..25eb9f3 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -538,6 +538,7 @@ import task_capability_map # noqa: E402 import review_proofs # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 +import issue_lock_provenance # noqa: E402 import already_landed_reconcile # noqa: E402 import author_mutation_worktree # noqa: E402 import issue_claim_heartbeat # noqa: E402 @@ -1208,6 +1209,10 @@ def gitea_lock_issue( "repo": r, "worktree_path": resolved_worktree, "work_lease": work_lease, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant=work_lease.get("claimant"), + ), } try: @@ -1298,6 +1303,14 @@ def gitea_create_pr( except Exception as e: raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)") + lock_provenance_check = issue_lock_provenance.assess_lock_file_for_create_pr( + lock_data + ) + if lock_provenance_check["block"]: + raise RuntimeError( + issue_lock_provenance.format_lock_provenance_error(lock_provenance_check) + ) + locked_issue = lock_data.get("issue_number") locked_branch = lock_data.get("branch_name") locked_worktree = lock_data.get("worktree_path") diff --git a/issue_lock_provenance.py b/issue_lock_provenance.py new file mode 100644 index 0000000..87ee38f --- /dev/null +++ b/issue_lock_provenance.py @@ -0,0 +1,269 @@ +"""Issue-lock provenance and external-state disclosure (#447). + +Sanctioned locks are written only by ``gitea_lock_issue`` (or adoption recovery +#442). Manual seeding of ``/tmp/gitea_issue_lock.json`` is unsafe and must be +blocked at PR creation unless explicit operator override proof is recorded. +""" + +from __future__ import annotations + +import os +import re +from datetime import datetime, timezone + +ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json") + +SOURCE_LOCK_ISSUE = "gitea_lock_issue" +SOURCE_LOCK_ADOPTION = "gitea_lock_issue_adoption" +SOURCE_OPERATOR_OVERRIDE = "operator_override" + +SANCTIONED_LOCK_SOURCES = frozenset({ + SOURCE_LOCK_ISSUE, + SOURCE_LOCK_ADOPTION, + SOURCE_OPERATOR_OVERRIDE, +}) + +_OPERATOR_OVERRIDE_ENV = "GITEA_ISSUE_LOCK_OPERATOR_OVERRIDE" + +_ISSUE_LOCK_PATH_RE = re.compile( + r"(?:/tmp/)?gitea_issue_lock\.json", + re.IGNORECASE, +) +_LOCK_SEED_RE = re.compile( + r"(?:seed(?:ed|ing)?|restor(?:e|ed|ing)|wrote|written|write|programmatically|" + r"hand[- ]forg|manual(?:ly)?).{0,80}gitea_issue_lock", + re.IGNORECASE | re.DOTALL, +) +_LOCK_REMOVE_RE = re.compile( + r"(?:\brm\b|remove|deleted?|unlink).{0,80}gitea_issue_lock", + re.IGNORECASE | re.DOTALL, +) +_LOCK_READ_RE = re.compile( + r"(?:read|loaded?|parsed?).{0,80}gitea_issue_lock", + re.IGNORECASE | re.DOTALL, +) +_EXTERNAL_NONE_RE = re.compile( + r"external[- ]state mutations\s*:\s*none\b", + re.IGNORECASE, +) +_EXTERNAL_FIELD_RE = re.compile( + r"external[- ]state mutations\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_CLEANUP_ONLY_RE = re.compile( + r"cleanup mutations\s*:\s*(?:none|lock removed|removed issue lock)", + re.IGNORECASE, +) +_PR_CREATED_RE = re.compile( + r"(?:\bgitea_create_pr\b|PR\s*#\s*\d+\s+created|created\s+PR\s*#|opened\s+PR\s*#|" + r"PR\s+creation\s+(?:succeeded|complete))", + re.IGNORECASE, +) +_REVIEW_APPROVE_RE = re.compile( + r"(?:submitted\s+(?:['\"]approve['\"]|approve\s+review)|" + r"review decision\s*:\s*approve|approved\s+PR\s*#|gitea_review_pr.*approve)", + re.IGNORECASE, +) +_OVERRIDE_PROOF_RE = re.compile( + r"operator[- ]override\s+proof\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def build_sanctioned_lock_provenance( + *, + tool: str, + source: str = SOURCE_LOCK_ISSUE, + claimant: dict | None = None, + adoption: dict | None = None, +) -> dict: + """Return provenance metadata stored with a sanctioned lock write.""" + entry = { + "source": source, + "written_at": _utc_now_iso(), + "written_by_tool": tool, + "lock_file_path": ISSUE_LOCK_FILE, + } + if claimant: + entry["claimant"] = claimant + if adoption: + entry["adoption"] = adoption + return entry + + +def operator_override_requested() -> bool: + return os.environ.get(_OPERATOR_OVERRIDE_ENV, "").strip().lower() in { + "1", + "true", + "yes", + } + + +def build_operator_override_provenance(*, reason: str, claimant: dict | None = None) -> dict: + text = (reason or "").strip() + if not text: + raise ValueError( + "operator override requires a non-empty override reason (fail closed)" + ) + entry = build_sanctioned_lock_provenance( + tool="operator_override", + source=SOURCE_OPERATOR_OVERRIDE, + claimant=claimant, + ) + entry["override_reason"] = text + return entry + + +def assess_lock_file_for_create_pr(lock_data: dict | None) -> dict: + """Fail closed when lock file lacks sanctioned provenance (#447).""" + data = lock_data if isinstance(lock_data, dict) else {} + reasons: list[str] = [] + provenance = data.get("lock_provenance") + if not isinstance(provenance, dict): + reasons.append( + "issue lock file lacks sanctioned lock_provenance; manual seeding is " + "not a normal recovery path — call gitea_lock_issue or use #442 adoption" + ) + return _provenance_result(False, reasons, provenance) + + source = str(provenance.get("source") or "").strip() + if source not in SANCTIONED_LOCK_SOURCES: + reasons.append( + f"issue lock provenance source '{source or '(missing)'}' is not sanctioned" + ) + + if source == SOURCE_OPERATOR_OVERRIDE and not str( + provenance.get("override_reason") or "" + ).strip(): + reasons.append( + "operator_override lock provenance requires override_reason proof" + ) + + if not data.get("work_lease"): + reasons.append("issue lock file missing work_lease metadata") + + if not str(provenance.get("written_by_tool") or "").strip(): + reasons.append("issue lock provenance missing written_by_tool") + + proven = not reasons + return _provenance_result(proven, reasons, provenance) + + +def _provenance_result(proven: bool, reasons: list[str], provenance: dict | None) -> dict: + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "lock_provenance": provenance, + } + + +def format_lock_provenance_error(assessment: dict) -> str: + reasons = "; ".join(assessment.get("reasons") or ["unknown lock provenance violation"]) + return f"Issue lock provenance guard (#447): {reasons} (fail closed)" + + +def _lock_activity_detected(text: str) -> dict[str, bool]: + body = text or "" + return { + "seed_or_restore": bool(_LOCK_SEED_RE.search(body)), + "remove": bool(_LOCK_REMOVE_RE.search(body)), + "read": bool(_LOCK_READ_RE.search(body)), + } + + +def _external_state_discloses_lock(text: str) -> bool: + match = _EXTERNAL_FIELD_RE.search(text or "") + if not match: + return False + value = (match.group(1) or "").strip().lower() + if value in {"", "none", "n/a"}: + return False + return "lock" in value or "gitea_issue_lock" in value or "issue-lock" in value + + +def assess_issue_lock_external_state_report(report_text: str) -> dict: + """Require explicit external-state disclosure for issue-lock mutations (#447).""" + text = report_text or "" + activity = _lock_activity_detected(text) + if not any(activity.values()): + return {"proven": True, "block": False, "reasons": [], "activity": activity} + + reasons: list[str] = [] + disclosed = _external_state_discloses_lock(text) + + if activity["seed_or_restore"] and _EXTERNAL_NONE_RE.search(text): + reasons.append( + "report mentions seeding/restoring gitea_issue_lock.json but claims " + "External-state mutations: none" + ) + elif activity["seed_or_restore"] and not disclosed: + reasons.append( + "report mentions issue-lock file activity but External-state mutations " + "does not disclose read/write of gitea_issue_lock.json" + ) + + if activity["remove"]: + if _EXTERNAL_NONE_RE.search(text): + reasons.append( + "report mentions removing gitea_issue_lock.json but claims " + "External-state mutations: none" + ) + elif not disclosed and _CLEANUP_ONLY_RE.search(text): + reasons.append( + "report removes issue lock but classifies it as cleanup only; " + "record under External-state mutations" + ) + elif not disclosed: + reasons.append( + "report mentions deleting issue lock without External-state " + "mutation disclosure" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "activity": activity, + } + + +def assess_manual_lock_pr_without_override(report_text: str) -> dict: + """Block reports that created a PR via manual lock seed without override proof.""" + text = report_text or "" + seeded = bool(_LOCK_SEED_RE.search(text)) + created = bool(_PR_CREATED_RE.search(text)) + if not (seeded and created): + return {"proven": True, "block": False, "reasons": []} + + if _OVERRIDE_PROOF_RE.search(text): + return {"proven": True, "block": False, "reasons": []} + + return { + "proven": False, + "block": True, + "reasons": [ + "report created/opened a PR after manual issue-lock seeding without " + "operator override proof" + ], + } + + +def assess_author_reviewer_same_run_report(report_text: str) -> dict: + """Reviewer handoff must not create and approve the same PR in one run (#447).""" + text = report_text or "" + if not (_PR_CREATED_RE.search(text) and _REVIEW_APPROVE_RE.search(text)): + return {"proven": True, "block": False, "reasons": []} + return { + "proven": False, + "block": True, + "reasons": [ + "report mixes author-side PR creation and reviewer approval in one " + "final handoff; split author and reviewer sessions" + ], + } \ No newline at end of file diff --git a/skills/llm-project-workflow/workflows/work-issue.md b/skills/llm-project-workflow/workflows/work-issue.md index f450cd7..8f7a227 100644 --- a/skills/llm-project-workflow/workflows/work-issue.md +++ b/skills/llm-project-workflow/workflows/work-issue.md @@ -717,6 +717,12 @@ Use only precise categories: * External-state mutations: * Read-only diagnostics: +Issue-lock file (`/tmp/gitea_issue_lock.json`) read/write/delete is always an +external-state mutation. Never claim `External-state mutations: none` after +seeding, restoring, or removing that file. Manual lock seeding is not a normal +recovery path (#447); use `gitea_lock_issue` or the #442 adoption recovery path +instead. Link broader redesign: #438. + `git fetch`, `git remote update`, and any command that updates refs must be listed under `Git ref mutations`, not read-only diagnostics. If `git reset --hard`, checkout, clean, worktree add/remove, merge simulation, merge abort, or similar commands occurred, report them under `Worktree/index mutations`. diff --git a/tests/test_commit_payloads.py b/tests/test_commit_payloads.py index e8e4123..eb712a0 100644 --- a/tests/test_commit_payloads.py +++ b/tests/test_commit_payloads.py @@ -67,6 +67,15 @@ class TestCommitPayloads(unittest.TestCase): self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name) self.lock_file_path = "/tmp/gitea_issue_lock.json" + import issue_lock_provenance + + work_lease = { + "operation_type": "author_issue_work", + "issue_number": 263, + "branch": "feat/issue-263-native-commit-payloads", + "claimant": {"username": "test-user", "profile": "test-author"}, + "expires_at": "2999-01-01T00:00:00Z", + } self.lock_data = { "issue_number": 263, "branch_name": "feat/issue-263-native-commit-payloads", @@ -74,6 +83,11 @@ class TestCommitPayloads(unittest.TestCase): "org": "Example-Org", "repo": "Example-Repo", "worktree_path": self.locked_worktree_path, + "work_lease": work_lease, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant=work_lease.get("claimant"), + ), } with open(self.lock_file_path, "w", encoding="utf-8") as fh: fh.write(json.dumps(self.lock_data)) diff --git a/tests/test_issue_lock_provenance.py b/tests/test_issue_lock_provenance.py new file mode 100644 index 0000000..3e64090 --- /dev/null +++ b/tests/test_issue_lock_provenance.py @@ -0,0 +1,130 @@ +"""Tests for issue-lock provenance and external-state disclosure (#447).""" + +from __future__ import annotations + +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import issue_lock_provenance as ilp # noqa: E402 +from final_report_validator import assess_final_report_validator # noqa: E402 + + +def _sanctioned_lock(**overrides): + work_lease = { + "operation_type": "author_issue_work", + "issue_number": 447, + "branch": "feat/issue-447-lock-provenance", + "claimant": {"username": "jcwalker3", "profile": "prgs-author"}, + "expires_at": "2999-01-01T00:00:00Z", + } + data = { + "issue_number": 447, + "branch_name": "feat/issue-447-lock-provenance", + "work_lease": work_lease, + "lock_provenance": ilp.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant=work_lease["claimant"], + ), + } + data.update(overrides) + return data + + +class TestLockProvenanceForCreatePr(unittest.TestCase): + def test_sanctioned_lock_passes(self): + result = ilp.assess_lock_file_for_create_pr(_sanctioned_lock()) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_manual_seed_without_provenance_blocked(self): + result = ilp.assess_lock_file_for_create_pr( + {"issue_number": 420, "branch_name": "feat/x", "work_lease": {}} + ) + self.assertTrue(result["block"]) + self.assertIn("lock_provenance", result["reasons"][0]) + + def test_operator_override_requires_reason(self): + result = ilp.assess_lock_file_for_create_pr( + _sanctioned_lock( + lock_provenance=ilp.build_sanctioned_lock_provenance( + tool="operator_override", + source=ilp.SOURCE_OPERATOR_OVERRIDE, + ) + ) + ) + self.assertTrue(result["block"]) + + +class TestExternalStateReportRules(unittest.TestCase): + def test_seed_with_external_none_blocked(self): + report = ( + "Restored /tmp/gitea_issue_lock.json to unblock PR creation.\n" + "- External-state mutations: none\n" + ) + result = ilp.assess_issue_lock_external_state_report(report) + self.assertTrue(result["block"]) + + def test_seed_with_disclosure_passes(self): + report = ( + "Restored /tmp/gitea_issue_lock.json after MCP restart.\n" + "- External-state mutations: wrote /tmp/gitea_issue_lock.json\n" + ) + result = ilp.assess_issue_lock_external_state_report(report) + self.assertTrue(result["proven"]) + + def test_remove_claimed_as_cleanup_only_blocked(self): + report = ( + "rm /tmp/gitea_issue_lock.json after PR creation.\n" + "- Cleanup mutations: lock removed\n" + "- External-state mutations: none\n" + ) + result = ilp.assess_issue_lock_external_state_report(report) + self.assertTrue(result["block"]) + + def test_manual_lock_pr_without_override_blocked(self): + report = ( + "Programmatically seeded gitea_issue_lock.json then gitea_create_pr.\n" + "PR #444 created.\n" + ) + result = ilp.assess_manual_lock_pr_without_override(report) + self.assertTrue(result["block"]) + + def test_author_reviewer_same_run_blocked(self): + report = ( + "gitea_create_pr opened PR #444.\n" + "Submitted approve review on PR #444.\n" + ) + result = ilp.assess_author_reviewer_same_run_report(report) + self.assertTrue(result["block"]) + + +class TestFinalReportValidatorIntegration(unittest.TestCase): + def test_work_issue_blocks_hidden_lock_mutation(self): + report = ( + "## Controller Handoff\n" + "- Task: work issue #420\n" + "- External-state mutations: none\n" + "Restored /tmp/gitea_issue_lock.json before PR creation.\n" + ) + result = assess_final_report_validator(report, "work_issue") + rule_ids = {f["rule_id"] for f in result["findings"]} + self.assertIn("shared.issue_lock_external_state", rule_ids) + self.assertTrue(result["blocked"]) + + def test_review_pr_blocks_create_and_approve(self): + report = ( + "## Controller Handoff\n" + "- Task: review PR #444\n" + "- Review decision: approve\n" + "Created PR #444 via gitea_create_pr earlier in this run.\n" + ) + result = assess_final_report_validator(report, "review_pr") + rule_ids = {f["rule_id"] for f in result["findings"]} + self.assertIn("shared.author_reviewer_same_run", rule_ids) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 4002b5e..9f39ae1 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -101,12 +101,26 @@ ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json" def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides): + import issue_lock_provenance + + work_lease = { + "operation_type": "author_issue_work", + "issue_number": issue_number, + "branch": branch_name, + "claimant": {"username": "test-user", "profile": "test-author"}, + "expires_at": "2999-01-01T00:00:00Z", + } record = { "issue_number": issue_number, "branch_name": branch_name, "remote": "dadeschools", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", + "work_lease": work_lease, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant=work_lease.get("claimant"), + ), } record.update(overrides) return record @@ -3284,6 +3298,28 @@ class TestIssueLocking(unittest.TestCase): ) self.assertIn("does not match locked worktree", str(ctx.exception)) + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_create_pr_manual_lock_seed_blocked(self, _auth, _role): + with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: + json.dump( + _sample_issue_lock( + issue_number=447, + branch_name="feat/issue-447-lock-provenance", + lock_provenance=None, + ), + f, + ) + with patch.dict(os.environ, CREATE_PR_ENV, clear=True): + with self.assertRaises(RuntimeError) as ctx: + gitea_create_pr( + title="feat: lock provenance Closes #447", + head="feat/issue-447-lock-provenance", + remote="prgs", + ) + self.assertIn("lock provenance", str(ctx.exception).lower()) + @patch("mcp_server.api_request") @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))