Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 81 Pull Requests 7 Actions Packages Projects Releases Wiki Activity

fix(session): keep KIND_DECISION_LOCK durable past generic 4h TTL (Closes #720)

Browse Source 
Terminal review-decision ledgers are recovery-critical provenance, not
disposable session cache. A generic four-hour TTL previously made
fresh_review_on_current_head_allowed unreachable after age expiry on open
PRs (PR #616 / review 443 reproduction).

- Classify KIND_DECISION_LOCK as RECOVERY_CRITICAL so load/mark_final work
  after >4h without hand-editing session-state files
- Stamp kind + recovery_critical on save for compatibility
- Add inspect_state_envelope so assessment reports on-disk evidence instead
  of silent "no lock" when TTL would hide non-critical kinds
- Surface disk_inspect on stale decision-lock cleanup assessment

Preserves same-head #332 hard-stop, #620 head-scoped fresh review, #594
moot cleanup, and #709 irrecoverable authorization (no ordinary-profile
permission grant).
This commit is contained in:
sysadmin
2026-07-16 15:47:52 -04:00
parent 77746b08fc
commit 80f59b334e
3 changed files with 668 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+44
View File
@@ -3344,6 +3344,11 @@ def _save_review_decision_lock(data):
)
payload["session_profile_lock"] = binding["session_profile_lock"]
payload["profile_identity"] = binding["profile_identity"]
# #720: durable decision locks are recovery-critical terminal provenance,
# not generic TTL session cache. Stamp kind + recovery_critical for
# pre-existing readers and identity_match_reasons flag-based exempt.
payload["kind"] = mcp_session_state.KIND_DECISION_LOCK
payload["recovery_critical"] = True
if binding.get("remote") and not payload.get("remote"):
payload["remote"] = binding["remote"]
# #695 AC6: stamp native transport provenance on durable decision locks.
@@ -5027,6 +5032,12 @@ def gitea_cleanup_stale_review_decision_lock(
binding = _decision_lock_binding()
active_identity = binding.get("profile_identity")
lock = _load_review_decision_lock()
# #720: when normal load yields no lock, inspect disk so assessment does not
# silently report "absent" while an expired/non-critical envelope remains.
disk_inspect = mcp_session_state.inspect_state_envelope(
kind=mcp_session_state.KIND_DECISION_LOCK,
profile_identity=active_identity,
)
last = stale_review_decision_lock.last_terminal_mutation(lock)
pr_live = None
pr_lookup_error = None
@@ -5048,6 +5059,26 @@ def gitea_cleanup_stale_review_decision_lock(
pr_lookup_error=pr_lookup_error,
active_profile_identity=active_identity,
)
if lock is None and disk_inspect.get("on_disk"):
assessment = dict(assessment)
assessment["reasons"] = list(assessment.get("reasons") or []) + [
"decision-lock file is present on disk but not loadable via normal "
f"TTL/identity gates: {disk_inspect.get('summary')} "
"(do not rm session-state files; #720)"
]
assessment["disk_inspect"] = {
k: disk_inspect.get(k)
for k in (
"on_disk",
"has_payload",
"age_hours",
"age_exceeds_default_ttl",
"recovery_critical",
"ttl_exempt",
"would_ttl_reject",
"summary",
)
}
# Optional pin: refuse apply against a different terminal PR than expected.
if (
@@ -5102,6 +5133,19 @@ def gitea_cleanup_stale_review_decision_lock(
"pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
"merge_commit_sha": assessment.get("merge_commit_sha"),
"lock_summary": assessment.get("lock_summary"),
"disk_inspect": assessment.get("disk_inspect") or {
k: disk_inspect.get(k)
for k in (
"on_disk",
"has_payload",
"age_hours",
"age_exceeds_default_ttl",
"recovery_critical",
"ttl_exempt",
"would_ttl_reject",
"summary",
)
},
"audit": audit,
"audit_comment_id": None,
"reasons": list(assessment.get("reasons") or []),
mcp_session_state.py
+125 -3
View File
@@ -46,8 +46,15 @@ KIND_IRRECOVERABLE_DECISION_PROVENANCE = "irrecoverable_decision_provenance"
KIND_IRRECOVERABLE_PROVENANCE_AUTH = "irrecoverable_provenance_authorization"
# Kinds that must survive the default session-state TTL (forensic / recovery).
#
# KIND_DECISION_LOCK is recovery-critical (#720): terminal review provenance is
# not disposable cache. A generic four-hour TTL must not drop old-head evidence
# or make ``fresh_review_on_current_head_allowed`` unreachable. Same-head / same-
# run #332 protections still apply once the ledger is loadable. Other session
# kinds (workflow load, drafts, etc.) remain TTL-bound.
RECOVERY_CRITICAL_KINDS = frozenset(
{
KIND_DECISION_LOCK,
KIND_DECISION_LOCK_ARCHIVE,
KIND_POST_MERGE_DECISION_RECOVERY,
KIND_IRRECOVERABLE_DECISION_PROVENANCE,
@@ -245,6 +252,16 @@ def _write_json(path: str, data: dict[str, Any]) -> None:
pass
def is_recovery_critical_record(record: dict[str, Any] | None, kind: str | None = None) -> bool:
"""True when a durable record must outlive the generic session-state TTL."""
if not record and not kind:
return False
record_kind = ((record or {}).get("kind") or kind or "").strip()
if record_kind in RECOVERY_CRITICAL_KINDS:
return True
return bool((record or {}).get("recovery_critical"))
def identity_match_reasons(
record: dict[str, Any] | None,
*,
@@ -288,9 +305,7 @@ def identity_match_reasons(
else:
age = _now_utc() - recorded_at
kind = (record.get("kind") or "").strip()
ttl_exempt = kind in RECOVERY_CRITICAL_KINDS or bool(
record.get("recovery_critical")
)
ttl_exempt = is_recovery_critical_record(record, kind=kind)
if age > timedelta(hours=ttl_hours()) and not ttl_exempt:
reasons.append(
f"session state expired after {ttl_hours():g}h (fail closed)"
@@ -300,6 +315,113 @@ def identity_match_reasons(
return reasons
def inspect_state_envelope(
*,
kind: str,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
profile_identity: str | None = None,
state_dir: str | None = None,
) -> dict[str, Any]:
"""Read-only disk inspection for assessment when TTL would otherwise hide state (#720).
Does **not** apply identity/TTL rejection to the returned presence flags.
Callers use this to distinguish:
* no file on disk
* file present but TTL would reject a non-critical kind
* recovery-critical ledger (e.g. KIND_DECISION_LOCK) still loadable
Never mutates files. Never returns secrets.
"""
profile = current_profile_identity(profile_identity=profile_identity)
path = state_file_path(
kind=kind,
remote=remote,
org=org,
repo=repo,
profile_identity=profile,
state_dir=state_dir,
)
result: dict[str, Any] = {
"kind": kind,
"profile_identity": profile,
"path_basename": os.path.basename(path),
"on_disk": False,
"has_payload": False,
"recorded_at": None,
"updated_at": None,
"age_hours": None,
"ttl_hours": ttl_hours(),
"age_exceeds_default_ttl": False,
"recovery_critical": kind in RECOVERY_CRITICAL_KINDS,
"ttl_exempt": kind in RECOVERY_CRITICAL_KINDS,
"would_ttl_reject": False,
"identity_reasons": [],
"summary": "no session-state file on disk",
}
if not path or not os.path.exists(path):
return result
result["on_disk"] = True
envelope = _read_json(path)
if not envelope:
result["summary"] = "session-state file present but unreadable or empty"
return result
payload = envelope.get("payload")
merged: dict[str, Any] = dict(payload) if isinstance(payload, dict) else {}
result["has_payload"] = isinstance(payload, dict)
for key in (
"kind",
"remote",
"org",
"repo",
"profile_identity",
"session_profile_lock",
"recorded_at",
"updated_at",
"writer_pid",
"recovery_critical",
):
if key in envelope and key not in merged:
merged[key] = envelope[key]
if not merged.get("kind"):
merged["kind"] = kind
recorded_at = _parse_iso(merged.get("recorded_at") or merged.get("updated_at"))
result["recorded_at"] = merged.get("recorded_at") or merged.get("updated_at")
result["updated_at"] = merged.get("updated_at") or merged.get("recorded_at")
if recorded_at is not None:
age = _now_utc() - recorded_at
age_hours = age.total_seconds() / 3600.0
result["age_hours"] = age_hours
result["age_exceeds_default_ttl"] = age > timedelta(hours=ttl_hours())
ttl_exempt = is_recovery_critical_record(merged, kind=kind)
result["recovery_critical"] = ttl_exempt
result["ttl_exempt"] = ttl_exempt
identity_reasons = identity_match_reasons(
merged,
remote=remote,
org=org,
repo=repo,
profile_identity=profile,
)
result["identity_reasons"] = list(identity_reasons)
result["would_ttl_reject"] = any("expired" in r for r in identity_reasons)
if result["has_payload"] and ttl_exempt:
result["summary"] = (
"recovery-critical session-state present on disk and TTL-exempt; "
"load via load_state for full payload"
)
elif result["has_payload"] and result["would_ttl_reject"]:
result["summary"] = (
"session-state file present on disk but generic TTL would reject load "
f"(age_hours={result.get('age_hours')!r}, ttl={ttl_hours():g}h)"
)
elif result["has_payload"]:
result["summary"] = "session-state file present and within TTL / identity gates"
else:
result["summary"] = "session-state file present without a dict payload"
return result
def load_state(
*,
kind: str,
tests/test_issue_720_expired_decision_lock.py
+499
View File
@@ -0,0 +1,499 @@
"""#720: Expired old-head KIND_DECISION_LOCK must not block fresh review.
Reproduction shape (PR #616):
* REQUEST_CHANGES terminal at head A
* open PR advanced to head B
* durable decision lock age > default 4h TTL
* fresh_review_on_current_head_allowed is true but unreachable under TTL-first reject
* mark_final fails with "session state expired after 4h"
* assessment may report "no lock" while the file remains on disk
Security invariants preserved:
* same-head second terminal remains fail-closed (#332/#620)
* REQUEST_CHANGES is never treated as approval
* historical mutations remain on the ledger
* merged/closed moot cleanup still allowed (#594)
* irrecoverable provenance still requires #709 authorization
* ordinary profiles do not gain gitea.decision_lock.irrecoverable_recovery
"""
from __future__ import annotations
import os
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
import sys
ROOT = str(Path(__file__).resolve().parent.parent)
if ROOT not in sys.path:
sys.path.insert(0, ROOT)
import mcp_session_state as ss
import mcp_server
import stale_review_decision_lock as srdl
import task_capability_map
HEAD_A = "a0fffae576673ba7df7456b32e6aec916581bdfb"
HEAD_B = "a6a2243aad9c3e385fc70509f4941a0f0ec33162"
HEAD_SAME = HEAD_A
RC_616_A = {
"pr_number": 616,
"action": "request_changes",
"review_id": 443,
"review_state": "request_changes",
"head_sha": HEAD_A,
}
def _lock(mutations=None, **kwargs):
base = {
"task": "review_pr",
"kind": ss.KIND_DECISION_LOCK,
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"session_pid": os.getpid(),
"session_profile": "prgs-reviewer",
"session_profile_lock": "prgs-reviewer",
"profile_identity": "prgs-reviewer",
"final_review_decision_ready": False,
"ready_pr_number": kwargs.get("ready_pr"),
"ready_action": kwargs.get("ready_action"),
"ready_expected_head_sha": kwargs.get("ready_head"),
"ready_remote": "prgs" if kwargs.get("ready_pr") else None,
"ready_org": "Scaled-Tech-Consulting" if kwargs.get("ready_pr") else None,
"ready_repo": "Gitea-Tools" if kwargs.get("ready_pr") else None,
"live_mutations": list(mutations or []),
"correction_authorized": False,
"correction_reason": None,
}
return base
def _open_pr(pr_number=616, head=HEAD_B, merged=False, closed=False):
state = "closed" if closed or merged else "open"
return {
"number": pr_number,
"state": state,
"merged": merged,
"merged_at": "2026-07-16T12:00:00Z" if merged else None,
"merge_commit_sha": "m" * 40 if merged else None,
"head": {"sha": head},
}
def _no_lease():
return {"block": False, "reasons": [], "mutation_allowed": True}
def _feedback(blocking=False, stale=True):
return {
"success": True,
"has_blocking_change_requests": blocking,
"review_feedback_stale": stale,
"current_head_sha": HEAD_B,
}
def _age_lock_payload(lock: dict, hours: float = 5.0) -> dict:
"""Rewrite timestamps to *hours* ago (simulates durable age without touching prod)."""
aged = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat().replace(
"+00:00", "Z"
)
out = dict(lock)
out["recorded_at"] = aged
out["updated_at"] = aged
return out
class TestIssue720ExpiredDecisionLockLifecycle(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.env = patch.dict(
os.environ,
{
ss.STATE_DIR_ENV: self._tmp.name,
ss.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
"GITEA_MCP_PROFILE": "prgs-reviewer",
"GITEA_PROFILE_NAME": "prgs-reviewer",
},
clear=False,
)
self.env.start()
mcp_server._REVIEW_DECISION_LOCK = None
import review_workflow_load
review_workflow_load.clear_review_workflow_load()
review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
mcp_server.gitea_load_review_workflow()
def tearDown(self):
mcp_server._REVIEW_DECISION_LOCK = None
import review_workflow_load
review_workflow_load.clear_review_workflow_load()
self.env.stop()
self._tmp.cleanup()
def _persist_aged_lock(self, mutations, hours: float = 5.0, **kwargs):
"""Write decision lock aged > TTL to the temp durable store (test-only)."""
payload = _age_lock_payload(_lock(mutations, **kwargs), hours=hours)
saved = ss.save_state(
kind=ss.KIND_DECISION_LOCK,
payload=payload,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
profile_identity="prgs-reviewer",
)
# save_state refreshes updated_at; re-age the on-disk envelope for TTL tests.
path = ss.state_file_path(
kind=ss.KIND_DECISION_LOCK,
profile_identity="prgs-reviewer",
state_dir=self._tmp.name,
)
aged = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat().replace(
"+00:00", "Z"
)
import json
with open(path, encoding="utf-8") as fh:
envelope = json.load(fh)
envelope["recorded_at"] = aged
envelope["updated_at"] = aged
if isinstance(envelope.get("payload"), dict):
envelope["payload"]["recorded_at"] = aged
envelope["payload"]["updated_at"] = aged
with open(path, "w", encoding="utf-8") as fh:
json.dump(envelope, fh, indent=2, sort_keys=True)
fh.write("\n")
# Drop memory so subsequent loads hit durable store.
mcp_server._REVIEW_DECISION_LOCK = None
return saved
def _mark(self, pr, action, head, feedback=None):
with patch("mcp_server._list_pr_lease_comments", return_value=[]), patch(
"mcp_server._pr_work_lease_reviewer_block", return_value=_no_lease()
), patch.object(
mcp_server,
"gitea_get_pr_review_feedback",
return_value=feedback or _feedback(blocking=False, stale=True),
), patch.object(
mcp_server,
"gitea_check_pr_eligibility",
return_value={"eligible": True, "head_sha": head},
), patch.object(
mcp_server.mcp_daemon_guard,
"assert_sanctioned_mutation_runtime",
return_value=None,
), patch.object(
mcp_server.mcp_daemon_guard,
"assert_no_direct_import_bypass",
return_value=None,
):
return mcp_server.gitea_mark_final_review_decision(
pr_number=pr,
action=action,
expected_head_sha=head,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
# --- AC regression matrix ---
def test_under_four_hours_fresh_review_at_b_allowed(self):
self._persist_aged_lock(
[RC_616_A],
hours=1.0,
ready_pr=616,
ready_head=HEAD_A,
ready_action="request_changes",
)
res = self._mark(616, "approve", HEAD_B)
self.assertTrue(res.get("marked_ready"), res)
def test_over_four_hours_fresh_review_at_b_still_allowed(self):
"""Primary #720 defect: age > TTL must not block head-B mark_final."""
self._persist_aged_lock(
[RC_616_A],
hours=5.0,
ready_pr=616,
ready_head=HEAD_A,
ready_action="request_changes",
)
# Prove durable load is possible for recovery-critical decision locks.
loaded = ss.load_state(
kind=ss.KIND_DECISION_LOCK,
profile_identity="prgs-reviewer",
)
self.assertIsNotNone(
loaded,
"expired KIND_DECISION_LOCK must remain loadable (not generic TTL cache)",
)
res = self._mark(616, "approve", HEAD_B)
self.assertTrue(
res.get("marked_ready"),
f"expected mark_ready on head B after >4h; got {res}",
)
reasons = " ".join(res.get("reasons") or [])
self.assertNotIn("session state expired", reasons)
def test_historical_review_at_a_preserved_and_stale(self):
self._persist_aged_lock(
[RC_616_A],
hours=5.0,
ready_pr=616,
ready_head=HEAD_A,
ready_action="request_changes",
)
res = self._mark(616, "approve", HEAD_B)
self.assertTrue(res.get("marked_ready"), res)
lock = mcp_server._load_review_decision_lock()
self.assertIsNotNone(lock)
hist = [m for m in lock["live_mutations"] if m.get("review_id") == 443]
self.assertEqual(len(hist), 1)
self.assertEqual(hist[0]["head_sha"], HEAD_A)
self.assertEqual(hist[0]["action"], "request_changes")
a = srdl.assess_stale_review_decision_lock(
lock, pr_live=_open_pr(616, HEAD_B)
)
self.assertTrue(a["stale_by_head"])
self.assertTrue(a["fresh_review_on_current_head_allowed"])
self.assertEqual(a["locked_head_sha"], HEAD_A)
def test_no_reuse_approval_from_head_a(self):
"""REQUEST_CHANGES at A is never an approval credential for B."""
self._persist_aged_lock([RC_616_A], hours=5.0)
lock = ss.load_state(
kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
)
last = srdl.last_terminal_mutation(lock)
self.assertEqual(last.get("action"), "request_changes")
self.assertNotEqual(last.get("action"), "approve")
# Gate must still require a new mark/submit for head B; prior RC is not approve.
reasons = mcp_server.terminal_review_hard_stop_reasons(
616, "mark_ready", expected_head_sha=HEAD_B
)
self.assertEqual(reasons, [])
def test_second_terminal_mutation_at_b_same_run_blocked(self):
self._persist_aged_lock(
[RC_616_A],
hours=5.0,
ready_pr=616,
ready_head=HEAD_A,
ready_action="request_changes",
)
res = self._mark(616, "approve", HEAD_B)
self.assertTrue(res.get("marked_ready"), res)
# Record terminal approve at B (same run).
lock = mcp_server._load_review_decision_lock()
lock["final_review_decision_ready"] = True
lock["ready_pr_number"] = 616
lock["ready_action"] = "approve"
lock["ready_expected_head_sha"] = HEAD_B
mcp_server._save_review_decision_lock(lock)
mcp_server.record_live_review_mutation(616, "approve", review_id=999)
# Second terminal at same head B must hard-stop.
hard = mcp_server.terminal_review_hard_stop_reasons(
616, "mark_ready", expected_head_sha=HEAD_B
)
self.assertTrue(hard)
res2 = self._mark(616, "approve", HEAD_B)
self.assertFalse(res2.get("marked_ready"))
def test_open_pr_same_head_expired_still_fail_closed(self):
self._persist_aged_lock(
[RC_616_A],
hours=5.0,
ready_pr=616,
ready_head=HEAD_A,
ready_action="request_changes",
)
res = self._mark(616, "approve", HEAD_SAME)
self.assertFalse(res.get("marked_ready"), res)
self.assertTrue(
any("#332" in r or "already consumed" in r for r in (res.get("reasons") or [])),
res,
)
def test_merged_pr_moot_cleanup_still_allowed(self):
self._persist_aged_lock(
[RC_616_A],
hours=5.0,
ready_pr=616,
ready_head=HEAD_A,
ready_action="request_changes",
)
lock = ss.load_state(
kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
)
a = srdl.assess_stale_review_decision_lock(
lock, pr_live=_open_pr(616, HEAD_A, merged=True)
)
self.assertTrue(a["is_moot"])
self.assertTrue(a["cleanup_allowed"])
self.assertFalse(a["fresh_review_on_current_head_allowed"])
def test_assessment_reports_expired_old_head_not_absent(self):
self._persist_aged_lock(
[RC_616_A],
hours=5.0,
ready_pr=616,
ready_head=HEAD_A,
ready_action="request_changes",
)
# Disk presence + load after lifecycle fix.
path = ss.state_file_path(
kind=ss.KIND_DECISION_LOCK,
profile_identity="prgs-reviewer",
state_dir=self._tmp.name,
)
self.assertTrue(os.path.exists(path))
loaded = ss.load_state(
kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
)
self.assertIsNotNone(loaded)
inspect = ss.inspect_state_envelope(
kind=ss.KIND_DECISION_LOCK,
profile_identity="prgs-reviewer",
state_dir=self._tmp.name,
)
self.assertTrue(inspect.get("on_disk"))
self.assertTrue(inspect.get("has_payload"))
self.assertTrue(inspect.get("recovery_critical") or inspect.get("ttl_exempt"))
self.assertTrue(inspect.get("age_hours", 0) >= 4.0)
a = srdl.assess_stale_review_decision_lock(
loaded, pr_live=_open_pr(616, HEAD_B)
)
self.assertTrue(a["has_lock"])
self.assertNotIn("no review decision lock present", " ".join(a["reasons"]))
self.assertTrue(a["stale_by_head"])
self.assertEqual(a["last_terminal_action"], "request_changes")
def test_existing_serialized_records_compatible(self):
"""Pre-fix ledgers without recovery_critical flag still load via kind."""
payload = _age_lock_payload(_lock([RC_616_A]), hours=8.0)
payload.pop("recovery_critical", None)
# Manually write pre-#720-shaped envelope (kind only on envelope).
path = ss.state_file_path(
kind=ss.KIND_DECISION_LOCK,
profile_identity="prgs-reviewer",
state_dir=self._tmp.name,
)
import json
os.makedirs(self._tmp.name, exist_ok=True)
aged = payload["recorded_at"]
envelope = {
"kind": ss.KIND_DECISION_LOCK,
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"profile_identity": "prgs-reviewer",
"session_profile_lock": "prgs-reviewer",
"recorded_at": aged,
"updated_at": aged,
"writer_pid": os.getpid(),
"payload": payload,
}
with open(path, "w", encoding="utf-8") as fh:
json.dump(envelope, fh, indent=2, sort_keys=True)
fh.write("\n")
loaded = ss.load_state(
kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
)
self.assertIsNotNone(loaded)
self.assertEqual(loaded["live_mutations"][0]["review_id"], 443)
def test_decision_lock_is_recovery_critical_kind(self):
self.assertIn(ss.KIND_DECISION_LOCK, ss.RECOVERY_CRITICAL_KINDS)
def test_workflow_load_still_ttl_expires(self):
"""Do not make every session-state kind permanently TTL-exempt."""
aged = (datetime.now(timezone.utc) - timedelta(hours=5)).isoformat().replace(
"+00:00", "Z"
)
rec = {
"kind": ss.KIND_WORKFLOW_LOAD,
"recorded_at": aged,
"updated_at": aged,
"profile_identity": "prgs-reviewer",
"session_profile_lock": "prgs-reviewer",
}
reasons = ss.identity_match_reasons(
rec, profile_identity="prgs-reviewer"
)
self.assertTrue(any("expired" in r for r in reasons), reasons)
def test_irrecoverable_permission_not_on_ordinary_profiles(self):
perm = "gitea.decision_lock.irrecoverable_recovery"
# Capability map must not map ordinary author/reviewer/merger tasks to it.
for task in (
"create_issue",
"comment_issue",
"review_pr",
"merge_pr",
"lock_issue",
"create_pr",
):
req = task_capability_map.required_permission(task)
self.assertNotEqual(req, perm, msg=task)
def test_structured_error_when_same_head_expired(self):
self._persist_aged_lock(
[RC_616_A],
hours=5.0,
ready_pr=616,
ready_head=HEAD_A,
ready_action="request_changes",
)
res = self._mark(616, "approve", HEAD_A)
self.assertFalse(res.get("marked_ready"))
reasons = res.get("reasons") or []
self.assertTrue(reasons)
blob = " ".join(reasons)
# Actionable recovery text from hard-stop (not generic internal_error).
self.assertIn("#332", blob)
self.assertTrue(
"#620" in blob or "head moved" in blob or "new expected_head_sha" in blob
or "already consumed" in blob
)
self.assertNotIn("internal_error", blob.lower())
class TestIssue720InspectEnvelope(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.env = patch.dict(
os.environ,
{
ss.STATE_DIR_ENV: self._tmp.name,
ss.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
},
clear=False,
)
self.env.start()
def tearDown(self):
self.env.stop()
self._tmp.cleanup()
def test_inspect_missing_file(self):
info = ss.inspect_state_envelope(
kind=ss.KIND_DECISION_LOCK,
profile_identity="prgs-reviewer",
state_dir=self._tmp.name,
)
self.assertFalse(info["on_disk"])
self.assertFalse(info["has_payload"])
if __name__ == "__main__":
unittest.main()