Merge branch 'master' into feat/issue-483-role-boundaries

This commit is contained in:
2026-07-08 21:51:38 -05:00
22 changed files with 2560 additions and 40 deletions
+52
View File
@@ -0,0 +1,52 @@
# Controller Issue-Acceptance Gate
A merged PR does not automatically prove an issue is fully satisfied. After
merge, a controller must audit the linked issue against its acceptance criteria
and post a durable handoff before the issue is treated as complete.
## Workflow position
1. Author implements the issue and opens a PR.
2. Reviewer reviews the PR.
3. Merger merges the approved PR.
4. **Controller performs issue-acceptance audit.**
5. Controller posts a `## Controller Issue Acceptance` comment with either:
- `STATE: accepted` and checked criteria, or
- a rejection path (`more-work-required`, `needs-tests`, `needs-docs`, etc.)
with `MISSING_WORK` and a paste-ready `NEXT_PROMPT`.
Gitea may auto-close an issue via `Closes #N` in the PR body. That closure is
merge mechanics only. Controller acceptance is still required before any final
report or queue controller treats the issue as complete.
## Template
Use `issue_acceptance_gate.render_controller_acceptance_template()` or the
copy in
[`skills/llm-project-workflow/templates/controller-issue-acceptance.md`](../skills/llm-project-workflow/templates/controller-issue-acceptance.md).
## Final-report rules
Final reports must not claim `issue complete` solely because a PR merged.
Either:
- include a valid `## Controller Issue Acceptance` block with
`STATE: accepted`, or
- explicitly state `controller acceptance pending` and identify the controller
as the next actor.
`final_report_validator` enforces this through
`issue_acceptance_gate.validate_final_report_issue_acceptance()`.
## Role boundaries
- Authors must not mark their own issues accepted.
- Reviewers must not mark issue acceptance unless acting under controller
capability.
- Mergers merge PRs; they do not substitute for controller acceptance.
## Related
- #495 — canonical next-action comment templates
- #496 — fail-closed canonical comment validation before posting
- #303 — controller handoff schema for reconciliation workflows
+1
View File
@@ -929,6 +929,7 @@ When posting a Canonical Thread Handoff after a binding blocker:
## Related documents
- [`issue-acceptance-gate.md`](issue-acceptance-gate.md) — controller issue-acceptance audit after PR merge (#500).
- [`../skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md) — portable cross-project LLM workflow skill.
- [`gitea-execution-profiles.md`](gitea-execution-profiles.md) — the profile model.
- [`gitea-dual-namespace-deployment.md`](gitea-dual-namespace-deployment.md) — static author/reviewer namespace deployment (#139 decision).
+11 -2
View File
@@ -46,7 +46,8 @@ Optional environment variables:
| `/runtime` | Stub — MCP runtime health (#430) |
| `/audit` | Stub — report audit paste (#431) |
| `/worktrees` | Stub — hygiene dashboard (#432) |
| `/leases` | Stub — lease visibility (#433) |
| `/leases` | Lease and collision visibility (#433) |
| `/api/leases` | JSON lease/collision export |
All routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
`read-only-mvp`.
@@ -82,8 +83,16 @@ credentials. The UI surfaces pagination proof (returned count, pages fetched,
If credentials are missing or the fetch fails, the page shows an explicit error
instead of an empty queue (fail closed).
## Lease visibility (#433)
`/leases` surfaces read-only lease and collision state: local issue lock file,
in-progress claim inventory (#268), reviewer PR lease comments when present
(`<!-- mcp-review-lease:v1 -->`, #407), duplicate open PRs per issue (#400),
and duplicate local branches per issue. Links to collision-history backend
issues (#267, #268, #400, #407) are included. No lease acquire/release from UI.
## Tests
```bash
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py -q
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_lease_visibility.py -q
```
+84
View File
@@ -11,6 +11,7 @@ import inspect
import re
from typing import Any, Callable
import issue_acceptance_gate
import issue_lock_provenance
from mcp_native_cleanup_proof import assess_mcp_native_cleanup_proof
from post_merge_cleanup_proof import assess_post_merge_cleanup_proof
@@ -964,6 +965,69 @@ def _rule_reconcile_linked_issue_live(
return []
_PR_CLOSE_NEGATIVE = frozenset({"", "none", "n/a", "na", "0", "no", "not closed", "not performed"})
_ANCESTOR_AFFIRMATIVE_RE = re.compile(
r"ancestor|passed|true|verified|confirmed", re.IGNORECASE
)
def _reconciler_pr_close_performed(fields: dict[str, str], lock: dict) -> bool:
"""True when the report/session indicates a reconciler PR close happened."""
if lock.get("pr_closed") is True:
return True
value = (fields.get("prs closed", "") or "").strip().lower()
if value in _PR_CLOSE_NEGATIVE:
return False
# A closed PR is reported by number (e.g. "#99") or an affirmative result.
return bool(re.search(r"#\s*\d+|\b(?:closed|success|done)\b", value))
def _rule_reconcile_close_proof(
report_text: str,
*,
reconciler_close_lock: dict | None = None,
) -> list[dict[str, str]]:
"""#306: a reconciler PR close must carry exact proof fields.
Read-only/comment-only reconciliations are untouched. Once a PR close is
reported (via the ``PRs closed`` field or a session close lock), the
handoff must prove the close capability, ancestor landing, PR close
result, and the linked-issue result — narrative alone fails closed.
"""
fields = _handoff_fields(report_text)
lock = reconciler_close_lock or {}
if not _reconciler_pr_close_performed(fields, lock):
return []
missing: list[str] = []
capabilities = fields.get("capabilities proven", "")
if "gitea.pr.close" not in capabilities.lower():
missing.append("close capability proof (gitea.pr.close)")
ancestor = fields.get("ancestor proof", "")
if not _ANCESTOR_AFFIRMATIVE_RE.search(ancestor):
missing.append("ancestor proof")
prs_closed = (fields.get("prs closed", "") or "").strip().lower()
if prs_closed in _PR_CLOSE_NEGATIVE and lock.get("pr_closed") is True:
missing.append("PR close result")
linked = fields.get("linked issue live status", "") or fields.get("issues closed", "")
if not linked.strip():
missing.append("linked issue result")
if not missing:
return []
return [
validator_finding(
"reconcile.close_proof_fields",
"block",
"Reconciler close proof",
"reconciler PR close reported without required proof field(s): "
+ ", ".join(missing),
"include close capability proof, ancestor proof, PR close result, "
"and linked issue result in the handoff",
)
]
def _rule_reconcile_pagination_proof(report_text: str) -> list[dict[str, str]]:
text = report_text or ""
if not _INVENTORY_COMPLETE_RE.search(text):
@@ -1050,6 +1114,22 @@ def _rule_shared_manual_lock_pr_override(report_text: str) -> list[dict[str, str
)
def _rule_shared_issue_acceptance_gate(report_text: str) -> list[dict[str, str]]:
result = issue_acceptance_gate.validate_final_report_issue_acceptance(report_text)
if not result.get("applicable") or result.get("valid"):
return []
return _findings_from_reasons(
"shared.issue_acceptance_gate",
result.get("reasons") or [],
field="Controller acceptance",
severity="block",
safe_next_action=(
"add Controller Issue Acceptance proof or state that controller "
"acceptance is pending; do not claim issue complete from PR merge alone"
),
)
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"):
@@ -1251,6 +1331,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
_rule_reconcile_stale_author_fields,
_rule_reconcile_eligible_reviewed,
_rule_reconcile_linked_issue_live,
_rule_reconcile_close_proof,
_rule_reconcile_pagination_proof,
_rule_reviewer_git_fetch_readonly,
_rule_reviewer_legacy_workspace_mutations,
@@ -1267,6 +1348,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
_rule_shared_controller_handoff,
_rule_shared_email_disclosure,
*_SHARED_ISSUE_LOCK_RULES,
_rule_shared_issue_acceptance_gate,
_rule_reviewer_vague_mutations_none,
_rule_conflict_fix_push_proof,
_rule_worktree_cleanup_audit_proof,
@@ -1349,6 +1431,7 @@ def assess_final_report_validator(
issue_filing_lock: dict | None = None,
session_pr_opened: bool = False,
validation_session: dict | None = None,
reconciler_close_lock: dict | None = None,
) -> dict[str, Any]:
"""Validate final-report text against task-specific proof rules (#327).
@@ -1405,6 +1488,7 @@ def assess_final_report_validator(
"local_edits": local_edits,
"session_pr_opened": session_pr_opened,
"validation_session": validation_session,
"reconciler_close_lock": reconciler_close_lock,
}
for rule in _RULES_BY_TASK.get(normalized_kind, ()):
+233 -15
View File
@@ -696,6 +696,39 @@ def _verify_role_mutation_workspace(
task: str | None = None,
) -> str:
"""Bind reviewer/merger mutations to the active namespace workspace (#510)."""
# Check running runtimes to prevent stale mutations
try:
if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
config = gitea_config.load_config()
required_permission = task_capability_map.required_permission(task) if task else None
matching_profiles = []
if required_permission and config and "profiles" in config:
for p_name, p_data in config["profiles"].items():
p_allowed = p_data.get("allowed_operations") or []
p_forbidden = p_data.get("forbidden_operations") or []
p_allowed_n = []
for op in p_allowed:
try:
p_allowed_n.append(gitea_config.normalize_operation(op))
except Exception:
pass
p_forbidden_n = []
for op in p_forbidden:
try:
p_forbidden_n.append(gitea_config.normalize_operation(op))
except Exception:
pass
ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n)
if ok:
matching_profiles.append(p_name)
runtime_reasons = _check_mcp_runtimes_diagnostics(task or "unknown", matching_profiles)
if runtime_reasons:
raise RuntimeError("; ".join(runtime_reasons))
except Exception as exc:
if "stale-runtime:" in str(exc):
raise RuntimeError(str(exc))
pass
role = _effective_workspace_role()
git_state = issue_lock_worktree.read_worktree_git_state(
_resolve_preflight_workspace_path(worktree_path)
@@ -771,6 +804,7 @@ import task_capability_map # noqa: E402
import review_proofs # noqa: E402
import review_workflow_boundary # noqa: E402
import review_workflow_load # noqa: E402
import mcp_session_state # noqa: E402
import agent_temp_artifacts
import issue_lock_worktree # noqa: E402
import issue_lock_provenance # noqa: E402
@@ -2459,37 +2493,110 @@ _REVIEW_ACTIONS = {
_TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
# In-process only (#211): never persist to /tmp — host-global files are
# spoofable by any local process and go stale across sessions.
# Durable across MCP daemon process pools (#559), but never under /tmp (#211):
# host-global temp files are spoofable. Persistence uses the user-private
# cache directory from mcp_session_state (mode 0o700 / files 0o600), keyed by
# remote + profile identity with TTL.
_REVIEW_DECISION_LOCK: dict | None = None
def _decision_lock_binding(lock: dict | None = None) -> dict:
"""Resolve key fields for durable decision-lock storage."""
profile = get_profile()
profile_name = (profile.get("profile_name") or "").strip()
env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
stored_lock = ((lock or {}).get("session_profile_lock") or "").strip()
session_lock = env_lock or stored_lock or profile_name
remote = ((lock or {}).get("remote") or "").strip() or None
org = ((lock or {}).get("org") or (lock or {}).get("ready_org") or "").strip() or None
repo = ((lock or {}).get("repo") or (lock or {}).get("ready_repo") or "").strip() or None
return {
"session_profile": profile_name,
"session_profile_lock": session_lock,
"profile_identity": mcp_session_state.current_profile_identity(
profile_name=profile_name,
session_profile_lock=session_lock,
),
"remote": remote,
"org": org,
"repo": repo,
}
def _load_review_decision_lock():
"""Load decision lock from memory, falling back to durable session state."""
global _REVIEW_DECISION_LOCK
if _REVIEW_DECISION_LOCK is not None:
return _REVIEW_DECISION_LOCK
binding = _decision_lock_binding()
# Profile-keyed durable file; remote/org/repo validated from payload (#559).
durable = mcp_session_state.load_state(
kind=mcp_session_state.KIND_DECISION_LOCK,
profile_identity=binding.get("profile_identity"),
)
if durable is not None:
_REVIEW_DECISION_LOCK = dict(durable)
return _REVIEW_DECISION_LOCK
def _save_review_decision_lock(data):
"""Persist decision lock to memory + durable shared state (#559)."""
global _REVIEW_DECISION_LOCK
_REVIEW_DECISION_LOCK = data
if data is None:
binding = _decision_lock_binding(_REVIEW_DECISION_LOCK)
mcp_session_state.clear_state(
kind=mcp_session_state.KIND_DECISION_LOCK,
profile_identity=binding.get("profile_identity"),
)
_REVIEW_DECISION_LOCK = None
return
payload = dict(data)
binding = _decision_lock_binding(payload)
payload.setdefault("session_pid", os.getpid())
payload["writer_pid"] = os.getpid()
payload["session_profile"] = binding["session_profile"] or payload.get(
"session_profile"
)
payload["session_profile_lock"] = binding["session_profile_lock"]
payload["profile_identity"] = binding["profile_identity"]
if binding.get("remote") and not payload.get("remote"):
payload["remote"] = binding["remote"]
persisted = mcp_session_state.save_state(
kind=mcp_session_state.KIND_DECISION_LOCK,
payload=payload,
remote=payload.get("remote"),
org=payload.get("org") or payload.get("ready_org"),
repo=payload.get("repo") or payload.get("ready_repo"),
profile_identity=payload.get("profile_identity"),
)
_REVIEW_DECISION_LOCK = dict(persisted or payload)
def _review_decision_session_reasons(lock: dict | None) -> list[str]:
"""Reject locks that do not belong to this MCP process/session."""
"""Reject locks that do not belong to this MCP session identity (#559).
Different daemon PIDs in the same IDE session pool are allowed when the
profile identity and remote match. Spoofed/stale locks still fail closed.
"""
if lock is None:
return []
reasons = []
if lock.get("session_pid") != os.getpid():
reasons.append(
"review decision lock was created in a different process "
"(fail closed)"
)
env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
stored_lock = (lock.get("session_profile_lock") or "").strip()
stored_lock = (lock.get("session_profile_lock") or lock.get("profile_identity") or "").strip()
if env_lock and stored_lock and env_lock != stored_lock:
reasons.append(
"review decision lock session profile lock mismatch (fail closed)"
)
# TTL / identity checks from durable envelope fields.
for reason in mcp_session_state.identity_match_reasons(
lock,
remote=lock.get("remote"),
org=lock.get("org") or lock.get("ready_org"),
repo=lock.get("repo") or lock.get("ready_repo"),
profile_identity=env_lock or stored_lock,
):
if "profile identity mismatch" in reason or "expired" in reason or "future" in reason or "missing recorded_at" in reason:
reasons.append(reason)
return reasons
@@ -2500,11 +2607,12 @@ def init_review_decision_lock(remote: str | None, task: str | None, force: bool
if not force:
lock = _load_review_decision_lock()
if lock is not None:
if lock.get("remote") == remote and lock.get("session_pid") == os.getpid():
env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
stored_lock = (lock.get("session_profile_lock") or "").strip()
if not env_lock or not stored_lock or env_lock == stored_lock:
return
env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
stored_lock = (lock.get("session_profile_lock") or "").strip()
same_remote = lock.get("remote") == remote
same_profile = (not env_lock or not stored_lock or env_lock == stored_lock)
if same_remote and same_profile and not _review_decision_session_reasons(lock):
return
review_workflow_load.clear_review_workflow_load()
profile = get_profile()
profile_name = (profile.get("profile_name") or "").strip()
@@ -2519,6 +2627,10 @@ def init_review_decision_lock(remote: str | None, task: str | None, force: bool
"session_pid": os.getpid(),
"session_profile": profile_name,
"session_profile_lock": session_lock,
"profile_identity": mcp_session_state.current_profile_identity(
profile_name=profile_name,
session_profile_lock=session_lock,
),
"final_review_decision_ready": False,
"ready_pr_number": None,
"ready_action": None,
@@ -8635,6 +8747,102 @@ def gitea_route_task_session(
)
def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> list[str]:
"""Check running runtimes and return errors if they are missing or stale."""
import subprocess
import re
from datetime import datetime
reasons = []
code_path = os.path.join(PROJECT_ROOT, "gitea_mcp_server.py")
if not os.path.exists(code_path):
return reasons
code_mtime = datetime.fromtimestamp(os.path.getmtime(code_path))
try:
proc = subprocess.run(
["ps", "-o", "pid,lstart,command", "-ax"],
capture_output=True, text=True, check=True
)
except Exception as exc:
return [f"stale-runtime: failed to list running processes: {exc}"]
self_pid = os.getpid()
self_stale = False
running_profiles = {}
for line in proc.stdout.splitlines()[1:]:
line = line.strip()
if not line or "mcp_server.py" not in line:
continue
parts = line.split(None, 6)
if len(parts) < 7:
continue
pid_str = parts[0]
lstart_str = " ".join(parts[1:6])
try:
pid = int(pid_str)
start_time = datetime.strptime(lstart_str, "%a %b %d %H:%M:%S %Y")
except Exception:
continue
try:
env_proc = subprocess.run(
["ps", "eww", str(pid)],
capture_output=True, text=True, check=True
)
env_out = env_proc.stdout
except Exception:
continue
profile = "gitea-default"
match = re.search(r'\bGITEA_MCP_PROFILE=([^\s]+)', env_out)
if match:
profile = match.group(1)
is_stale = start_time < code_mtime
if pid == self_pid and is_stale:
self_stale = True
if profile not in running_profiles or start_time > running_profiles[profile]["start_time"]:
running_profiles[profile] = {
"pid": pid,
"start_time": start_time,
"is_stale": is_stale
}
if self_stale:
reasons.append(
"stale-runtime: The active Gitea MCP server process is stale (running code from before changes were merged). "
"Please fully restart the Gitea MCP server (e.g. touch /Users/jasonwalker/.gemini/config/mcp_config.json) and retry."
)
if matching_profiles:
any_running = False
any_fresh = False
for mp in matching_profiles:
if mp in running_profiles:
any_running = True
if not running_profiles[mp]["is_stale"]:
any_fresh = True
break
if not any_running:
reasons.append(
f"stale-runtime: None of the matching profiles for task '{task}' ({matching_profiles}) are running in the OS. "
"Please restart the MCP server to ensure they are spawned."
)
elif not any_fresh:
reasons.append(
f"stale-runtime: All matching profiles for task '{task}' ({matching_profiles}) are running but stale. "
"Please fully restart the Gitea MCP server and retry."
)
return reasons
@mcp.tool()
def gitea_resolve_task_capability(
task: str,
@@ -8745,6 +8953,16 @@ def gitea_resolve_task_capability(
configured = len(matching_profiles) > 0
available_in_session = allowed_in_current_session
if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
if runtime_reasons:
restart_required = True
reason_msg = "; ".join(runtime_reasons)
next_safe_action = (
"stale-runtime: Gitea MCP runtime conflict or missing process detected. "
"Please fully restart the Gitea MCP server and retry."
)
if not allowed_in_current_session:
if configured and switching:
restart_required = True
+300
View File
@@ -0,0 +1,300 @@
"""Controller issue-acceptance gate helpers (#500).
Pure validation for controller acceptance comments and final-report claims
that an issue is complete. Does not post comments or close issues.
"""
from __future__ import annotations
import re
ACCEPTANCE_HEADING = "controller issue acceptance"
REQUIRED_FIELDS = (
"STATE",
"WHO_IS_NEXT",
"NEXT_ACTION",
"NEXT_PROMPT",
"ISSUE",
"MERGED_PR",
"MERGE_COMMIT",
"ACCEPTANCE_CRITERIA_CHECKED",
"VALIDATION_REVIEWED",
"CONTROLLER_DECISION",
"WHY",
)
ACCEPTED_STATES = frozenset({"accepted"})
REJECTION_STATES = frozenset({
"more-work-required",
"more_work_required",
"needs-tests",
"needs_tests",
"needs-docs",
"needs_docs",
"needs-feature-enhancement",
"needs_feature_enhancement",
"needs-follow-up-issue",
"needs_follow_up_issue",
"blocked",
})
ALLOWED_NEXT_ACTORS = frozenset({
"controller",
"author",
"reviewer",
"merger",
"reconciler",
"user",
})
_FIELD_RE = re.compile(r"^\s*(?:[-*]\s*)?([A-Z][A-Z0-9_ ]+)\s*:\s*(.*)$")
_FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
_ISSUE_REF_RE = re.compile(r"#\d+")
_PR_REF_RE = re.compile(r"#\d+")
_CHECKED_ITEM_RE = re.compile(r"\[[xX]\]")
_UNCHECKED_ITEM_RE = re.compile(r"\[[\s]\]")
_CLAIMS_ISSUE_COMPLETE_RE = re.compile(
r"\bissue\s+(?:is\s+)?(?:complete|completed|accepted|closed\s+as\s+complete|fully\s+satisfied)\b|"
r"\bissue\s+acceptance\s*:\s*accepted\b|"
r"\bcontroller\s+acceptance\s*:\s*(?:accepted|complete)\b",
re.IGNORECASE,
)
_MERGE_ONLY_COMPLETE_RE = re.compile(
r"(?:pr\s+merged|merged\s+pr|merge\s+result\s*:\s*merged).{0,120}"
r"(?:issue\s+(?:is\s+)?(?:complete|closed|accepted)|issue\s+complete)",
re.IGNORECASE | re.DOTALL,
)
_CLAIMS_CONTROLLER_ACCEPTANCE_RE = re.compile(
r"controller\s+issue\s+acceptance|controller\s+acceptance\s+(?:posted|complete|pending)",
re.IGNORECASE,
)
_PENDING_ACCEPTANCE_RE = re.compile(
r"controller\s+acceptance\s+(?:pending|required|not\s+(?:yet\s+)?(?:performed|complete))",
re.IGNORECASE,
)
def render_controller_acceptance_template() -> str:
"""Return the canonical controller issue-acceptance comment template."""
return """## Controller Issue Acceptance
STATE:
<accepted | more-work-required | needs-tests | needs-docs | needs-feature-enhancement | needs-follow-up-issue | blocked>
WHO_IS_NEXT:
<author | reviewer | merger | reconciler | controller | user>
NEXT_ACTION:
<one sentence>
NEXT_PROMPT:
<paste-ready prompt for the next LLM>
ISSUE:
#...
MERGED_PR:
#...
MERGE_COMMIT:
<40-character SHA>
ACCEPTANCE_CRITERIA_CHECKED:
- [x] ...
- [ ] ...
VALIDATION_REVIEWED:
<tests/proofs reviewed>
CONTROLLER_DECISION:
<accepted or rejected>
WHY:
<reasoning>
MISSING_WORK:
<none, or exact missing work>
FOLLOW_UP_ISSUES:
<none, or issue list to create>
BLOCKERS:
<none, or exact blockers>
LAST_UPDATED_BY:
<identity/profile/date>
"""
def extract_acceptance_fields(text: str | None) -> dict[str, str]:
"""Return upper-case labeled fields from a controller acceptance block."""
fields: dict[str, str] = {}
current_key: str | None = None
for line in (text or "").splitlines():
match = _FIELD_RE.match(line)
if match:
current_key = match.group(1).strip().upper().replace(" ", "_")
fields[current_key] = match.group(2).strip()
continue
stripped = line.strip()
if current_key and stripped:
existing = fields.get(current_key, "")
fields[current_key] = (
f"{existing}\n{stripped}" if existing else stripped
)
return fields
def contains_acceptance_block(text: str | None) -> bool:
return ACCEPTANCE_HEADING in (text or "").lower()
def _empty_or_placeholder(value: str | None) -> bool:
value = (value or "").strip().lower()
return not value or value in {"none", "n/a", "unknown", "tbd", "<...>", "..."}
def _normalize_state(value: str | None) -> str:
return (value or "").strip().lower().replace(" ", "_").replace("-", "_")
def validate_controller_acceptance_comment(text: str | None) -> dict:
"""Validate a controller issue-acceptance comment."""
body = text or ""
if not contains_acceptance_block(body):
return {
"valid": False,
"fields": {},
"reasons": ["missing Controller Issue Acceptance heading"],
}
fields = extract_acceptance_fields(body)
reasons: list[str] = []
for field in REQUIRED_FIELDS:
if _empty_or_placeholder(fields.get(field)):
reasons.append(f"missing required controller acceptance field: {field}")
state = _normalize_state(fields.get("STATE"))
if state and state not in ACCEPTED_STATES and state not in REJECTION_STATES:
reasons.append(
"STATE must be accepted or a rejection path "
"(more-work-required, needs-tests, needs-docs, "
"needs-feature-enhancement, needs-follow-up-issue, blocked)"
)
actor = (fields.get("WHO_IS_NEXT") or "").strip().lower()
if actor and actor not in ALLOWED_NEXT_ACTORS:
reasons.append(
"WHO_IS_NEXT must be one of: "
+ ", ".join(sorted(ALLOWED_NEXT_ACTORS))
)
if not _ISSUE_REF_RE.search(fields.get("ISSUE") or ""):
reasons.append("ISSUE must cite an issue number (#N)")
if not _PR_REF_RE.search(fields.get("MERGED_PR") or ""):
reasons.append("MERGED_PR must cite a merged PR number (#N)")
if not _FULL_SHA_RE.search(fields.get("MERGE_COMMIT") or ""):
reasons.append("MERGE_COMMIT must include a full 40-character SHA")
criteria = fields.get("ACCEPTANCE_CRITERIA_CHECKED") or ""
if not _CHECKED_ITEM_RE.search(criteria) and not _UNCHECKED_ITEM_RE.search(criteria):
reasons.append(
"ACCEPTANCE_CRITERIA_CHECKED must list checked/unchecked criteria items"
)
decision = (fields.get("CONTROLLER_DECISION") or "").strip().lower()
if state in ACCEPTED_STATES:
if decision not in {"accepted", "accept"}:
reasons.append("accepted STATE requires CONTROLLER_DECISION: accepted")
if not _CHECKED_ITEM_RE.search(criteria):
reasons.append(
"accepted STATE requires at least one checked acceptance criterion"
)
if _empty_or_placeholder(fields.get("WHY")):
reasons.append("accepted STATE requires WHY with acceptance rationale")
elif state in REJECTION_STATES:
if decision not in {"rejected", "reject", "more_work_required"}:
reasons.append(
"rejection STATE requires CONTROLLER_DECISION: rejected"
)
if _empty_or_placeholder(fields.get("NEXT_PROMPT")):
reasons.append(
"rejection STATE requires a paste-ready NEXT_PROMPT for the next actor"
)
missing = fields.get("MISSING_WORK") or ""
if _empty_or_placeholder(missing):
reasons.append(
"rejection STATE requires MISSING_WORK describing what is still needed"
)
return {
"valid": not reasons,
"fields": fields,
"reasons": reasons,
}
def claims_issue_complete(text: str | None) -> bool:
"""Return True when text claims an issue is complete/accepted."""
return bool(_CLAIMS_ISSUE_COMPLETE_RE.search(text or ""))
def claims_merge_only_issue_complete(text: str | None) -> bool:
"""Return True when text treats PR merge as issue completion."""
return bool(_MERGE_ONLY_COMPLETE_RE.search(text or ""))
def claims_controller_acceptance_update(text: str | None) -> bool:
return bool(_CLAIMS_CONTROLLER_ACCEPTANCE_RE.search(text or ""))
def notes_controller_acceptance_pending(text: str | None) -> bool:
return bool(_PENDING_ACCEPTANCE_RE.search(text or ""))
def validate_final_report_issue_acceptance(report_text: str | None) -> dict:
"""Validate issue-completion and controller-acceptance claims in final reports."""
text = report_text or ""
reasons: list[str] = []
complete_claim = claims_issue_complete(text)
merge_only = claims_merge_only_issue_complete(text)
acceptance_claim = claims_controller_acceptance_update(text)
pending_noted = notes_controller_acceptance_pending(text)
has_block = contains_acceptance_block(text)
if merge_only and not (has_block and validate_controller_acceptance_comment(text)["valid"]):
reasons.append(
"final report treats PR merge as issue completion without controller acceptance proof"
)
if complete_claim and not pending_noted:
if not has_block:
reasons.append(
"final report claims issue complete but includes no Controller Issue Acceptance block"
)
else:
result = validate_controller_acceptance_comment(text)
if not result["valid"]:
reasons.extend(result["reasons"])
else:
state = _normalize_state(result["fields"].get("STATE"))
if state not in ACCEPTED_STATES:
reasons.append(
"final report claims issue complete but controller STATE is not accepted"
)
if acceptance_claim and has_block:
result = validate_controller_acceptance_comment(text)
if not result["valid"]:
reasons.extend(result["reasons"])
applicable = complete_claim or merge_only or acceptance_claim or pending_noted
return {
"applicable": applicable,
"valid": not reasons,
"reasons": reasons,
}
+391
View File
@@ -0,0 +1,391 @@
"""Durable MCP session validation state shared across daemon process pools (#559).
The IDE often routes sequential MCP tool calls to different daemon processes.
Session-scoped proofs (workflow load, review decision lock) must therefore
survive process boundaries while remaining fail-closed against spoofing.
Security notes (extends #211):
- Never store under host-global ``/tmp`` (world-writable, spoofable).
- Default root is ``~/.cache/gitea-tools/session-state`` (mode ``0o700``).
- Files are written atomically with mode ``0o600``.
- Records are keyed by remote + org + repo + profile identity, not by PID.
- TTL prevents indefinitely stale reuse across unrelated sessions.
"""
from __future__ import annotations
import fcntl
import json
import os
import re
import tempfile
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from typing import Any
STATE_DIR_ENV = "GITEA_MCP_SESSION_STATE_DIR"
DEFAULT_STATE_DIR = os.path.expanduser("~/.cache/gitea-tools/session-state")
TTL_HOURS_ENV = "GITEA_MCP_SESSION_STATE_TTL_HOURS"
DEFAULT_TTL_HOURS = 4.0
KIND_WORKFLOW_LOAD = "review_workflow_load"
KIND_DECISION_LOCK = "review_decision_lock"
_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"
def default_state_dir() -> str:
raw = (os.environ.get(STATE_DIR_ENV) or DEFAULT_STATE_DIR).strip()
return raw or DEFAULT_STATE_DIR
def ttl_hours() -> float:
raw = (os.environ.get(TTL_HOURS_ENV) or "").strip()
if not raw:
return DEFAULT_TTL_HOURS
try:
value = float(raw)
except ValueError:
return DEFAULT_TTL_HOURS
return value if value > 0 else DEFAULT_TTL_HOURS
def _sanitize_segment(value: str) -> str:
text = (value or "").strip()
if not text:
return "_"
return _SAFE_SEGMENT_RE.sub("_", text)
def current_profile_identity(
profile_name: str | None = None,
session_profile_lock: str | None = None,
profile_identity: str | None = None,
) -> str:
"""Resolve the session profile identity used as the durable key."""
env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
explicit = (profile_identity or session_profile_lock or "").strip()
lock = (explicit or env_lock or "").strip()
name = (profile_name or "").strip()
return lock or name or "unknown-profile"
def state_key(
*,
kind: str,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
profile_identity: str | None = None,
) -> str:
"""Build durable filename key.
Session proofs are one-active-per-profile (workflow load / decision lock),
so the primary key is kind + profile identity. Remote/org/repo are stored
inside the payload and validated on load (#559), which lets a later daemon
process recover state without already knowing the remote argument.
"""
# Keep remote/org/repo parameters for API stability / future kinds; they are
# intentionally not part of the filename for session-scoped proofs.
_ = (remote, org, repo)
return "-".join(
_sanitize_segment(part)
for part in (
kind,
profile_identity or "unknown-profile",
)
)
def state_file_path(
*,
kind: str,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
profile_identity: str | None = None,
state_dir: str | None = None,
) -> str:
root = (state_dir or default_state_dir()).strip()
name = state_key(
kind=kind,
remote=remote,
org=org,
repo=repo,
profile_identity=profile_identity,
)
return os.path.join(root, f"{name}.json")
def _ensure_state_dir(state_dir: str | None = None) -> str:
root = (state_dir or default_state_dir()).strip()
os.makedirs(root, mode=0o700, exist_ok=True)
return root
def _now_utc() -> datetime:
return datetime.now(timezone.utc)
def _parse_iso(value: str | None) -> datetime | None:
text = (value or "").strip()
if not text:
return None
if text.endswith("Z"):
text = text[:-1] + "+00:00"
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return None
if parsed.tzinfo is None:
return parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
@contextmanager
def _exclusive_file_lock(lock_path: str):
os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True)
fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
try:
fcntl.flock(fd, fcntl.LOCK_EX)
yield fd
finally:
try:
fcntl.flock(fd, fcntl.LOCK_UN)
finally:
os.close(fd)
def _read_json(path: str) -> dict[str, Any] | None:
if not path or not os.path.exists(path):
return None
try:
with open(path, encoding="utf-8") as handle:
data = json.load(handle)
except (OSError, json.JSONDecodeError):
return None
return data if isinstance(data, dict) else None
def _write_json(path: str, data: dict[str, Any]) -> None:
parent = os.path.dirname(path) or "."
os.makedirs(parent, mode=0o700, exist_ok=True)
payload = json.dumps(data, indent=2, sort_keys=True) + "\n"
fd, temp_path = tempfile.mkstemp(prefix=".session-", suffix=".json", dir=parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
os.chmod(temp_path, 0o600)
os.replace(temp_path, path)
try:
os.chmod(path, 0o600)
except OSError:
pass
finally:
if os.path.exists(temp_path):
try:
os.remove(temp_path)
except OSError:
pass
def identity_match_reasons(
record: dict[str, Any] | None,
*,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
profile_identity: str | None = None,
) -> list[str]:
"""Return fail-closed reasons when durable record identity does not match."""
if record is None:
return []
reasons: list[str] = []
expected_profile = current_profile_identity(profile_identity=profile_identity)
stored_profile = (
(record.get("session_profile_lock") or record.get("profile_identity") or "")
.strip()
)
if stored_profile and expected_profile and stored_profile != expected_profile:
if expected_profile != "unknown-profile":
reasons.append(
"session state profile identity mismatch "
f"(stored={stored_profile!r}, active={expected_profile!r}; fail closed)"
)
for field, expected in (
("remote", remote),
("org", org),
("repo", repo),
):
want = (expected or "").strip()
have = (str(record.get(field) or "")).strip()
if want and have and want != have:
reasons.append(
f"session state {field} mismatch "
f"(stored={have!r}, expected={want!r}; fail closed)"
)
recorded_at = _parse_iso(record.get("recorded_at") or record.get("updated_at"))
if recorded_at is None:
reasons.append("session state missing recorded_at timestamp (fail closed)")
else:
age = _now_utc() - recorded_at
if age > timedelta(hours=ttl_hours()):
reasons.append(
f"session state expired after {ttl_hours():g}h (fail closed)"
)
if age < timedelta(0):
reasons.append("session state recorded_at is in the future (fail closed)")
return reasons
def load_state(
*,
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] | None:
"""Load durable state payload when identity checks pass."""
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,
)
lock_path = f"{path}.lock"
with _exclusive_file_lock(lock_path):
envelope = _read_json(path)
if not envelope:
return None
payload = envelope.get("payload")
if not isinstance(payload, dict):
return None
# Identity fields live on both envelope and payload for convenience.
merged = dict(payload)
for key in (
"kind",
"remote",
"org",
"repo",
"profile_identity",
"session_profile_lock",
"recorded_at",
"updated_at",
"writer_pid",
):
if key in envelope and key not in merged:
merged[key] = envelope[key]
reasons = identity_match_reasons(
merged,
remote=remote,
org=org,
repo=repo,
profile_identity=profile,
)
if reasons:
return None
return merged
def save_state(
*,
kind: str,
payload: dict[str, Any] | None,
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] | None:
"""Persist or clear durable state for the given session identity key."""
profile = current_profile_identity(
profile_name=payload.get("session_profile") if payload else None,
session_profile_lock=(
(payload or {}).get("session_profile_lock") or profile_identity
),
)
# Prefer explicit args over payload fields for key location.
key_remote = remote if remote is not None else (payload or {}).get("remote")
key_org = org if org is not None else (payload or {}).get("org")
key_repo = repo if repo is not None else (payload or {}).get("repo")
root = _ensure_state_dir(state_dir)
path = state_file_path(
kind=kind,
remote=key_remote,
org=key_org,
repo=key_repo,
profile_identity=profile,
state_dir=root,
)
lock_path = f"{path}.lock"
with _exclusive_file_lock(lock_path):
if payload is None:
for candidate in (path, lock_path):
if os.path.exists(candidate):
try:
os.remove(candidate)
except OSError:
pass
return None
now = _now_utc().isoformat().replace("+00:00", "Z")
body = dict(payload)
body.setdefault("session_pid", os.getpid())
body["writer_pid"] = os.getpid()
body["profile_identity"] = profile
if not (body.get("session_profile_lock") or "").strip():
body["session_profile_lock"] = profile
body["recorded_at"] = body.get("recorded_at") or now
body["updated_at"] = now
if key_remote is not None:
body["remote"] = key_remote
if key_org is not None:
body["org"] = key_org
if key_repo is not None:
body["repo"] = key_repo
envelope = {
"kind": kind,
"remote": key_remote,
"org": key_org,
"repo": key_repo,
"profile_identity": profile,
"session_profile_lock": body.get("session_profile_lock"),
"recorded_at": body["recorded_at"],
"updated_at": body["updated_at"],
"writer_pid": body["writer_pid"],
"payload": body,
}
_write_json(path, envelope)
return dict(body)
def clear_state(
*,
kind: str,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
profile_identity: str | None = None,
state_dir: str | None = None,
) -> None:
save_state(
kind=kind,
payload=None,
remote=remote,
org=org,
repo=repo,
profile_identity=profile_identity,
state_dir=state_dir,
)
+113 -9
View File
@@ -1,4 +1,4 @@
"""Canonical review-merge workflow load proof for reviewer mutations (#389, #403)."""
"""Canonical review-merge workflow load proof for reviewer mutations (#389, #403, #559)."""
from __future__ import annotations
@@ -7,6 +7,7 @@ import os
import re
from pathlib import Path
import mcp_session_state
import review_workflow_boundary as boundary
WORKFLOW_REL_PATH = (
@@ -88,17 +89,79 @@ def assess_prompt_conflict(prompt_text: str | None) -> tuple[bool, list[str]]:
return bool(reasons), reasons
def _session_binding_fields() -> dict:
"""Capture profile identity used to share state across daemon processes."""
env_lock = (os.environ.get(mcp_session_state.SESSION_PROFILE_LOCK_ENV) or "").strip()
profile_name = (os.environ.get("GITEA_MCP_PROFILE") or "").strip()
remote = (os.environ.get("GITEA_MCP_REMOTE") or "").strip() or None
identity = mcp_session_state.current_profile_identity(
profile_name=profile_name,
session_profile_lock=env_lock,
)
return {
"session_profile": profile_name or identity,
"session_profile_lock": env_lock or identity,
"profile_identity": identity,
"remote": remote,
}
def _persist_workflow_load(record: dict | None) -> dict | None:
"""Write durable workflow-load proof (or clear it)."""
binding = _session_binding_fields()
if record is None:
mcp_session_state.clear_state(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
remote=binding.get("remote"),
profile_identity=binding.get("profile_identity"),
)
return None
payload = dict(record)
payload.update({
k: v for k, v in binding.items() if v is not None
})
return mcp_session_state.save_state(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
payload=payload,
remote=payload.get("remote"),
org=payload.get("org"),
repo=payload.get("repo"),
profile_identity=payload.get("profile_identity"),
)
def _load_durable_workflow_load() -> dict | None:
binding = _session_binding_fields()
return mcp_session_state.load_state(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
remote=binding.get("remote"),
profile_identity=binding.get("profile_identity"),
)
def _active_workflow_load() -> dict | None:
"""Prefer in-process cache; fall back to durable shared state (#559)."""
global _REVIEW_WORKFLOW_LOAD
if _REVIEW_WORKFLOW_LOAD is not None:
return _REVIEW_WORKFLOW_LOAD
durable = _load_durable_workflow_load()
if durable is not None:
_REVIEW_WORKFLOW_LOAD = dict(durable)
return _REVIEW_WORKFLOW_LOAD
def record_review_workflow_load(
project_root: str,
*,
prompt_text: str | None = None,
) -> dict:
"""Record in-process workflow load proof for the current MCP session."""
"""Record workflow load proof for the current MCP session (durable + memory)."""
global _REVIEW_WORKFLOW_LOAD
meta = build_canonical_workflow_metadata(
project_root, prompt_text=prompt_text)
boundary_state = boundary.assess_boundary_status(project_root)
_REVIEW_WORKFLOW_LOAD = {
binding = _session_binding_fields()
record = {
**meta,
"session_pid": os.getpid(),
"loaded": True,
@@ -107,7 +170,10 @@ def record_review_workflow_load(
"pre_review_command_count": boundary_state.get("pre_review_command_count"),
"boundary_violation_count": boundary_state.get("boundary_violation_count"),
"boundary_reasons": list(boundary_state.get("reasons") or []),
**{k: v for k, v in binding.items() if v is not None},
}
persisted = _persist_workflow_load(record)
_REVIEW_WORKFLOW_LOAD = dict(persisted or record)
return dict(_REVIEW_WORKFLOW_LOAD)
@@ -115,12 +181,13 @@ def clear_review_workflow_load() -> None:
"""Test helper and review_pr session reset."""
global _REVIEW_WORKFLOW_LOAD
_REVIEW_WORKFLOW_LOAD = None
_persist_workflow_load(None)
boundary.clear_pre_review_commands()
def workflow_load_status(project_root: str | None = None) -> dict:
"""Non-throwing status for capability/runtime reports."""
load = _REVIEW_WORKFLOW_LOAD
load = _active_workflow_load()
if load is None:
return {
"workflow_load_proof_present": False,
@@ -148,6 +215,8 @@ def workflow_load_status(project_root: str | None = None) -> dict:
"prompt_conflicts_with_workflow": load.get(
"prompt_conflicts_with_workflow"),
"session_pid": load.get("session_pid"),
"writer_pid": load.get("writer_pid"),
"profile_identity": load.get("profile_identity"),
"boundary_status": load.get("boundary_status"),
"boundary_clean": load.get("boundary_clean"),
"workflow_load_helper_result": boundary.workflow_load_helper_result(
@@ -160,13 +229,47 @@ def _session_validation_reasons(
load: dict,
project_root: str | None,
) -> list[str]:
"""Validate durable/in-memory load proof for this session identity (#559)."""
reasons: list[str] = []
if load.get("session_pid") != os.getpid():
# Cross-process daemon pools are allowed when profile identity matches.
# Reject only when the stored profile identity conflicts with this process.
binding = _session_binding_fields()
stored_identity = (
load.get("session_profile_lock")
or load.get("profile_identity")
or ""
).strip()
active_identity = (binding.get("profile_identity") or "").strip()
if (
stored_identity
and active_identity
and stored_identity != active_identity
and active_identity != "unknown-profile"
and stored_identity != "unknown-profile"
):
reasons.append(
"workflow load proof was recorded in a different process "
"(fail closed)"
"workflow load proof profile identity mismatch "
f"(stored={stored_identity!r}, active={active_identity!r}; fail closed)"
)
return reasons
# Expired durable records are treated as absent.
identity_reasons = mcp_session_state.identity_match_reasons(
load,
remote=binding.get("remote") or load.get("remote"),
org=load.get("org"),
repo=load.get("repo"),
profile_identity=active_identity or stored_identity,
)
# Filter out remote-mismatch noise when remote was not bound at load time.
for reason in identity_reasons:
if "missing recorded_at" in reason or "expired" in reason or "future" in reason:
reasons.append(reason)
elif "profile identity mismatch" in reason:
reasons.append(reason)
if reasons:
return reasons
if load.get("prompt_conflicts_with_workflow"):
reasons.extend(load.get("prompt_conflict_reasons") or [
"active prompt conflicts with loaded review-merge workflow"
@@ -196,7 +299,8 @@ def review_workflow_load_blockers(
) -> list[str]:
"""Reasons reviewer mutations must fail closed."""
boundary_reasons = boundary.boundary_blockers(project_root)
if boundary_reasons and _REVIEW_WORKFLOW_LOAD is None:
load = _active_workflow_load()
if boundary_reasons and load is None:
return boundary_reasons
status = workflow_load_status(project_root)
if not status.get("workflow_load_proof_present"):
@@ -214,4 +318,4 @@ def recovery_handoff_without_replay() -> list[str]:
"Do not call gitea_submit_pr_review, gitea_mark_final_review_decision, "
"or gitea_merge_pr until workflow-load proof is present.",
"Do not include approve/merge replay commands in the recovery handoff.",
]
]
@@ -0,0 +1,68 @@
# Controller issue-acceptance prompt
Use after a PR merges when auditing whether the linked issue is truly complete.
```text
Audit issue #<N> against its acceptance criteria after merged PR #<PR>.
Post a Controller Issue Acceptance comment with checked criteria, validation
reviewed, controller decision, next actor, and paste-ready next prompt.
Do not mark the issue accepted unless every required criterion is satisfied.
```
## Comment template
```text
## Controller Issue Acceptance
STATE:
<accepted | more-work-required | needs-tests | needs-docs | needs-feature-enhancement | needs-follow-up-issue | blocked>
WHO_IS_NEXT:
<author | reviewer | merger | reconciler | controller | user>
NEXT_ACTION:
<one sentence>
NEXT_PROMPT:
<paste-ready prompt for the next LLM>
ISSUE:
#...
MERGED_PR:
#...
MERGE_COMMIT:
<40-character SHA>
ACCEPTANCE_CRITERIA_CHECKED:
- [x] ...
- [ ] ...
VALIDATION_REVIEWED:
<tests/proofs reviewed>
CONTROLLER_DECISION:
<accepted or rejected>
WHY:
<reasoning>
MISSING_WORK:
<none, or exact missing work>
FOLLOW_UP_ISSUES:
<none, or issue list to create>
BLOCKERS:
<none, or exact blockers>
LAST_UPDATED_BY:
<identity/profile/date>
```
## Rejection paths
When rejecting completion, `STATE` must name the gap (`needs-tests`,
`needs-docs`, `more-work-required`, etc.), `MISSING_WORK` must be explicit,
and `NEXT_PROMPT` must be ready for the next author session.
@@ -375,6 +375,24 @@ Include:
* confirmation that no normal review, approval, request-changes, or merge was
performed
## 18A. Reconciler close proof is enforced (#306)
When a reconciler run closes a PR, the final-report validator
(`final_report_validator` rule `reconcile.close_proof_fields`) **blocks** the
handoff unless it carries all four close proofs. The prompt is guidance; the
MCP validator is the authority.
A close is detected from `PRs closed: #<n>` (or a session close lock). Once a
close is reported, the handoff must include:
* `Capabilities proven:` naming `gitea.pr.close` — the exact close capability
* `Ancestor proof:` — the landed/ancestor proof for the closed PR
* `PRs closed:` — the PR close result (the closed PR number)
* `Linked issue live status:` (or `Issues closed:`) — the linked-issue result
Comment-only and blocked reconciliations (no PR close) are unaffected: the rule
returns no finding when nothing was closed.
## 19. Local artifact and report consistency rule
Do not create local walkthrough, notes, markdown, JSON, or report artifacts
+129
View File
@@ -0,0 +1,129 @@
#!/usr/bin/env python3
"""Live health-check script to verify Gitea MCP namespace connections.
Spawns the MCP server processes as defined in the IDE's global config,
performs the JSON-RPC handshake, and queries the tools list to verify
that the connection is fully operational and doesn't return EOF.
"""
import json
import os
import subprocess
import sys
def test_connection(name, config):
print(f"Testing MCP connection for '{name}'...")
command = config.get("command")
args = config.get("args", [])
env = config.get("env", {})
# Merge current environment
run_env = os.environ.copy()
run_env.update(env)
# Spawn subprocess
try:
proc = subprocess.Popen(
[command] + args,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
env=run_env
)
except Exception as e:
print(f" [FAIL] Failed to spawn process: {e}")
return False
# Send initialize request
init_req = {
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "healthcheck", "version": "1.0"}
},
"id": 1
}
try:
proc.stdin.write(json.dumps(init_req) + "\n")
proc.stdin.flush()
# Read response
line = proc.stdout.readline()
if not line:
stderr_content = proc.stderr.read()
print(f" [FAIL] Received EOF from process. Stderr:\n{stderr_content}")
proc.terminate()
return False
print(f" [OK] Received initialize response: {line.strip()[:150]}...")
# Send initialized notification
init_notif = {
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
proc.stdin.write(json.dumps(init_notif) + "\n")
proc.stdin.flush()
# Send tools/list request
list_req = {
"jsonrpc": "2.0",
"method": "tools/list",
"params": {},
"id": 2
}
proc.stdin.write(json.dumps(list_req) + "\n")
proc.stdin.flush()
line = proc.stdout.readline()
if not line:
print(" [FAIL] Received EOF on tools/list request.")
proc.terminate()
return False
res = json.loads(line)
if "error" in res:
print(f" [FAIL] Server returned error: {res['error']}")
proc.terminate()
return False
tools = res.get("result", {}).get("tools", [])
tool_names = [t.get("name") for t in tools]
print(f" [OK] Successfully retrieved {len(tool_names)} tools: {tool_names[:5]}...")
proc.terminate()
return True
except Exception as e:
print(f" [FAIL] Error during handshake: {e}")
proc.terminate()
return False
def main():
config_path = "/Users/jasonwalker/.gemini/config/mcp_config.json"
try:
with open(config_path) as f:
mcp_config = json.load(f)
except Exception as e:
print(f"Failed to load mcp_config.json: {e}")
sys.exit(1)
servers = mcp_config.get("mcpServers", {})
failed = False
for name in ["gitea-author", "gitea-reviewer"]:
if name in servers:
if not test_connection(name, servers[name]):
failed = True
else:
print(f"Server '{name}' not found in mcp_config.json")
if failed:
sys.exit(1)
else:
print("All Gitea MCP connection tests passed!")
if __name__ == "__main__":
main()
+21
View File
@@ -18,14 +18,25 @@ def _reset_mutation_authority(monkeypatch):
explicitly.
"""
monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
# Isolate durable session-state files so tests never share host cache (#559).
import tempfile
_state_tmp = tempfile.TemporaryDirectory(prefix="gitea-session-state-")
monkeypatch.setenv("GITEA_MCP_SESSION_STATE_DIR", _state_tmp.name)
try:
import mcp_server
except Exception:
_state_tmp.cleanup()
yield
return
monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
try:
import review_workflow_load
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
except Exception:
pass
try:
import capability_stop_terminal
capability_stop_terminal.clear()
@@ -37,3 +48,13 @@ def _reset_mutation_authority(monkeypatch):
capability_stop_terminal.clear()
except Exception:
pass
try:
import review_workflow_load
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
except Exception:
pass
try:
mcp_server._REVIEW_DECISION_LOCK = None
except Exception:
pass
_state_tmp.cleanup()
+75
View File
@@ -498,6 +498,81 @@ class TestCanonicalReconcileSchema(unittest.TestCase):
)
class TestReconcilerCloseProof(unittest.TestCase):
"""Issue #306: a reconciler PR close must carry its proof fields."""
def _closed_report(self, **kwargs):
# A report that claims PR #99 was closed via the reconciler path.
report = (
_reconcile_handoff(**kwargs)
.replace(
"- Capabilities proven: gitea.read, gitea.pr.comment",
"- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close",
)
.replace("- Missing capabilities: gitea.pr.close", "- Missing capabilities: none")
.replace("- PRs closed: none", "- PRs closed: #99")
.replace(
"- Blocker: missing gitea.pr.close capability", "- Blocker: none"
)
)
return report
def test_full_proof_close_passes(self):
result = assess_final_report_validator(
self._closed_report(), "reconcile_already_landed"
)
self.assertEqual(result["grade"], "A")
self.assertFalse(
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
)
def test_close_without_capability_proof_blocks(self):
# Claims PRs closed: #99 but never proves gitea.pr.close capability.
report = _reconcile_handoff().replace("- PRs closed: none", "- PRs closed: #99")
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertTrue(result["blocked"])
self.assertTrue(
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
)
def test_close_without_ancestor_proof_blocks(self):
report = self._closed_report(drop=("Ancestor proof",))
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertTrue(result["blocked"])
self.assertTrue(
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
)
def test_close_without_linked_issue_result_blocks(self):
report = self._closed_report(drop=("Linked issue live status", "Issues closed"))
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertTrue(result["blocked"])
self.assertTrue(
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
)
def test_close_lock_requires_proof_even_if_text_says_none(self):
# Session lock proves a PR was closed; the report omits close proof.
result = assess_final_report_validator(
_reconcile_handoff(),
"reconcile_already_landed",
reconciler_close_lock={"pr_closed": True},
)
self.assertTrue(result["blocked"])
self.assertTrue(
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
)
def test_comment_only_reconcile_needs_no_close_proof(self):
result = assess_final_report_validator(
_reconcile_handoff(), "reconcile_already_landed"
)
self.assertEqual(result["grade"], "A")
self.assertFalse(
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
)
class TestEntryPoint(unittest.TestCase):
def test_unknown_task_kind_blocks(self):
result = assess_final_report_validator("report", "unknown_mode")
+183
View File
@@ -0,0 +1,183 @@
"""Tests for controller issue-acceptance gate (#500)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_acceptance_gate # noqa: E402
from final_report_validator import assess_final_report_validator # noqa: E402
def _accepted_comment(**overrides):
lines = [
"## Controller Issue Acceptance",
"",
"STATE:",
"accepted",
"",
"WHO_IS_NEXT:",
"controller",
"",
"NEXT_ACTION:",
"Close the tracker follow-up after verification.",
"",
"NEXT_PROMPT:",
"Verify deployment checklist for issue #500 and post acceptance.",
"",
"ISSUE:",
"#500",
"",
"MERGED_PR:",
"#503",
"",
"MERGE_COMMIT:",
"0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"",
"ACCEPTANCE_CRITERIA_CHECKED:",
"- [x] documentation added",
"- [x] validator tests added",
"",
"VALIDATION_REVIEWED:",
"pytest tests/test_issue_acceptance_gate.py -q",
"",
"CONTROLLER_DECISION:",
"accepted",
"",
"WHY:",
"All acceptance criteria satisfied with proof.",
"",
"MISSING_WORK:",
"none",
"",
"FOLLOW_UP_ISSUES:",
"none",
"",
"BLOCKERS:",
"none",
"",
"LAST_UPDATED_BY:",
"jcwalker3 / prgs-controller / 2026-07-08",
]
text = "\n".join(lines)
for key, value in overrides.items():
text = text.replace(f"{key}:\n", f"{key}:\n{value}\n", 1)
return text
def _rejection_comment(state="needs-tests"):
text = _accepted_comment()
text = text.replace("STATE:\naccepted", f"STATE:\n{state}")
text = text.replace(
"CONTROLLER_DECISION:\naccepted",
"CONTROLLER_DECISION:\nrejected",
)
text = text.replace(
"ACCEPTANCE_CRITERIA_CHECKED:\n- [x] documentation added\n- [x] validator tests added",
"ACCEPTANCE_CRITERIA_CHECKED:\n- [ ] regression tests for rejection paths",
)
text = text.replace(
"MISSING_WORK:\nnone",
"MISSING_WORK:\nAdd regression tests for controller rejection paths.",
)
text = text.replace(
"NEXT_PROMPT:\nVerify deployment checklist for issue #500 and post acceptance.",
"NEXT_PROMPT:\nImplement the missing tests for issue #500 and reopen the PR.",
)
return text
class TestControllerAcceptanceComment(unittest.TestCase):
def test_accepted_comment_valid(self):
result = issue_acceptance_gate.validate_controller_acceptance_comment(
_accepted_comment()
)
self.assertTrue(result["valid"], result["reasons"])
def test_missing_who_is_next_rejected(self):
text = _accepted_comment().replace("WHO_IS_NEXT:\ncontroller\n", "WHO_IS_NEXT:\n\n")
result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
self.assertFalse(result["valid"])
self.assertTrue(any("WHO_IS_NEXT" in r for r in result["reasons"]))
def test_missing_next_prompt_on_rejection(self):
text = _rejection_comment()
text = text.replace(
"NEXT_PROMPT:\nImplement the missing tests for issue #500 and reopen the PR.\n",
"NEXT_PROMPT:\n\n",
)
result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
self.assertFalse(result["valid"])
self.assertTrue(any("NEXT_PROMPT" in r for r in result["reasons"]))
def test_vague_merge_only_completion_detected(self):
text = "PR merged. Issue complete."
self.assertTrue(issue_acceptance_gate.claims_merge_only_issue_complete(text))
def test_rejection_paths_require_missing_work(self):
for state in (
"more-work-required",
"needs-tests",
"needs-docs",
"needs-feature-enhancement",
"needs-follow-up-issue",
):
text = _rejection_comment(state=state).replace(
"MISSING_WORK:\nAdd regression tests for controller rejection paths.\n",
"MISSING_WORK:\n\n",
)
result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
self.assertFalse(result["valid"], state)
self.assertTrue(any("MISSING_WORK" in r for r in result["reasons"]))
def test_accepted_requires_checked_criteria(self):
text = _accepted_comment().replace(
"ACCEPTANCE_CRITERIA_CHECKED:\n- [x] documentation added\n- [x] validator tests added\n",
"ACCEPTANCE_CRITERIA_CHECKED:\n- [ ] documentation added\n",
)
result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
self.assertFalse(result["valid"])
self.assertTrue(any("checked acceptance criterion" in r for r in result["reasons"]))
class TestFinalReportIntegration(unittest.TestCase):
def test_merge_only_complete_blocks_work_issue_report(self):
report = (
"## Controller Handoff\n\n"
"- Task: work issue #500\n"
"- Merge result: merged\n"
"- Current status: PR merged; issue complete\n"
)
result = assess_final_report_validator(report, "work_issue")
self.assertTrue(
any(
f["rule_id"] == "shared.issue_acceptance_gate"
for f in result["findings"]
),
result["findings"],
)
def test_pending_acceptance_allowed(self):
report = (
"## Controller Handoff\n\n"
"- Task: work issue #500\n"
"- Merge result: merged\n"
"- Current status: controller acceptance pending\n"
)
result = assess_final_report_validator(report, "work_issue")
self.assertFalse(
any(
f["rule_id"] == "shared.issue_acceptance_gate"
for f in result["findings"]
),
result["findings"],
)
def test_accepted_block_allows_complete_claim(self):
report = _accepted_comment() + "\n\nIssue is complete."
gate = issue_acceptance_gate.validate_final_report_issue_acceptance(report)
self.assertTrue(gate["valid"], gate["reasons"])
if __name__ == "__main__":
unittest.main()
+207
View File
@@ -0,0 +1,207 @@
"""Tests for durable MCP session state shared across daemon processes (#559)."""
from __future__ import annotations
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
sys_path_root = str(Path(__file__).resolve().parent.parent)
import sys
if sys_path_root not in sys.path:
sys.path.insert(0, sys_path_root)
import mcp_session_state
import review_workflow_load
import mcp_server
class TestMcpSessionStateStore(unittest.TestCase):
def setUp(self):
self._tmpdir = tempfile.TemporaryDirectory()
self.state_dir = self._tmpdir.name
self._env = patch.dict(
os.environ,
{
mcp_session_state.STATE_DIR_ENV: self.state_dir,
mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
"GITEA_MCP_PROFILE": "prgs-reviewer",
},
clear=False,
)
self._env.start()
def tearDown(self):
self._env.stop()
self._tmpdir.cleanup()
def test_round_trip_same_identity(self):
saved = mcp_session_state.save_state(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
payload={"loaded": True, "workflow_hash": "abc123def456"},
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertIsNotNone(saved)
self.assertEqual(saved["workflow_hash"], "abc123def456")
self.assertIn("recorded_at", saved)
loaded = mcp_session_state.load_state(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertIsNotNone(loaded)
self.assertEqual(loaded["workflow_hash"], "abc123def456")
self.assertEqual(loaded["profile_identity"], "prgs-reviewer")
def test_profile_mismatch_returns_none(self):
mcp_session_state.save_state(
kind=mcp_session_state.KIND_DECISION_LOCK,
payload={"final_review_decision_ready": True, "remote": "prgs"},
remote="prgs",
)
with patch.dict(
os.environ,
{mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"},
clear=False,
):
loaded = mcp_session_state.load_state(
kind=mcp_session_state.KIND_DECISION_LOCK,
remote="prgs",
)
self.assertIsNone(loaded)
def test_clear_removes_file(self):
mcp_session_state.save_state(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
payload={"loaded": True},
remote="prgs",
)
path = mcp_session_state.state_file_path(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
remote="prgs",
profile_identity="prgs-reviewer",
state_dir=self.state_dir,
)
self.assertTrue(os.path.exists(path))
mcp_session_state.clear_state(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
remote="prgs",
profile_identity="prgs-reviewer",
)
self.assertFalse(os.path.exists(path))
def test_files_are_private_mode(self):
mcp_session_state.save_state(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
payload={"loaded": True},
remote="prgs",
)
path = mcp_session_state.state_file_path(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
remote="prgs",
profile_identity="prgs-reviewer",
state_dir=self.state_dir,
)
mode = os.stat(path).st_mode & 0o777
self.assertEqual(mode, 0o600)
class TestWorkflowLoadCrossProcess(unittest.TestCase):
def setUp(self):
self._tmpdir = tempfile.TemporaryDirectory()
self._env = patch.dict(
os.environ,
{
mcp_session_state.STATE_DIR_ENV: self._tmpdir.name,
mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
"GITEA_MCP_PROFILE": "prgs-reviewer",
"GITEA_MCP_REMOTE": "prgs",
},
clear=False,
)
self._env.start()
review_workflow_load.clear_review_workflow_load()
def tearDown(self):
review_workflow_load.clear_review_workflow_load()
self._env.stop()
self._tmpdir.cleanup()
def test_different_pid_same_profile_accepted(self):
root = str(Path(__file__).resolve().parent.parent)
recorded = review_workflow_load.record_review_workflow_load(root)
self.assertTrue(recorded["loaded"])
# Simulate another daemon process: clear memory, keep durable file,
# and change apparent PID identity only (profile remains the same).
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
status = review_workflow_load.workflow_load_status(root)
self.assertTrue(status["workflow_load_proof_present"])
self.assertTrue(
status["workflow_load_valid"],
msg=status.get("reasons"),
)
def test_profile_mismatch_blocks(self):
root = str(Path(__file__).resolve().parent.parent)
review_workflow_load.record_review_workflow_load(root)
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
with patch.dict(
os.environ,
{mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"},
clear=False,
):
# Durable load uses active identity; mismatched profile key misses.
status = review_workflow_load.workflow_load_status(root)
self.assertFalse(status["workflow_load_proof_present"])
class TestDecisionLockCrossProcess(unittest.TestCase):
def setUp(self):
self._tmpdir = tempfile.TemporaryDirectory()
self._env = patch.dict(
os.environ,
{
mcp_session_state.STATE_DIR_ENV: self._tmpdir.name,
mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
},
clear=False,
)
self._env.start()
mcp_server._save_review_decision_lock(None)
review_workflow_load.clear_review_workflow_load()
def tearDown(self):
mcp_server._save_review_decision_lock(None)
review_workflow_load.clear_review_workflow_load()
self._env.stop()
self._tmpdir.cleanup()
def test_decision_lock_survives_memory_clear(self):
with patch.object(mcp_server, "get_profile", return_value={
"profile_name": "prgs-reviewer",
}):
mcp_server.init_review_decision_lock("prgs", "review_pr", force=True)
lock = mcp_server._load_review_decision_lock()
self.assertIsNotNone(lock)
self.assertEqual(lock["remote"], "prgs")
self.assertFalse(lock["final_review_decision_ready"])
# New process: memory empty, durable state remains.
mcp_server._REVIEW_DECISION_LOCK = None
restored = mcp_server._load_review_decision_lock()
self.assertIsNotNone(restored)
self.assertEqual(restored["remote"], "prgs")
reasons = mcp_server._review_decision_session_reasons(restored)
self.assertEqual(reasons, [])
if __name__ == "__main__":
unittest.main()
+71
View File
@@ -0,0 +1,71 @@
import os
import unittest
from unittest.mock import patch, MagicMock
from datetime import datetime
import gitea_mcp_server
class TestMcpStaleRuntime(unittest.TestCase):
@patch("subprocess.run")
@patch("os.path.getmtime")
@patch("os.path.exists")
@patch("os.getpid")
@patch.dict("os.environ", {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"})
def test_stale_and_missing_runtimes(self, mock_getpid, mock_exists, mock_getmtime, mock_run):
# Setup mocks
mock_getpid.return_value = 12345
mock_exists.return_value = True
# Code modification time: Jul 8 2026, 14:00:00
code_time = datetime(2026, 7, 8, 14, 0, 0)
mock_getmtime.return_value = code_time.timestamp()
# Mock ps -ax output
# PID 12345 is self (started at 13:00:00 - stale)
# PID 54321 is prgs-author (started at 15:00:00 - fresh)
# prgs-reviewer is missing
ps_output = (
" PID LSTART COMMAND\n"
"12345 Wed Jul 8 13:00:00 2026 /path/to/python mcp_server.py\n"
"54321 Wed Jul 8 15:00:00 2026 /path/to/python mcp_server.py\n"
)
mock_run_ps = MagicMock()
mock_run_ps.stdout = ps_output
# Mock env output for ps eww
mock_run_env12345 = MagicMock()
mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler"
mock_run_env54321 = MagicMock()
mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author"
def side_effect(args, **kwargs):
if args[0] == "ps" and "eww" in args:
pid = args[2]
if pid == "12345":
return mock_run_env12345
elif pid == "54321":
return mock_run_env54321
elif args[0] == "ps":
return mock_run_ps
raise ValueError(f"Unexpected subprocess run args: {args}")
mock_run.side_effect = side_effect
# Test 1: required reviewer role matching prgs-reviewer (which is missing)
reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("review_pr", ["prgs-reviewer"])
self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons))
self.assertTrue(any("stale-runtime: None of the matching profiles for task" in r for r in reasons))
# Test 2: required author role matching prgs-author (which is fresh)
reasons_author = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
# Still contains self stale error
self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons_author))
# But does NOT contain missing author profile error
self.assertFalse(any("None of the matching profiles for task" in r for r in reasons_author))
if __name__ == "__main__":
unittest.main()
+31 -5
View File
@@ -33,12 +33,38 @@ class TestReviewWorkflowLoadModule(unittest.TestCase):
self.assertTrue(status["workflow_load_proof_present"])
self.assertTrue(status["workflow_load_valid"])
def test_stale_session_pid_blocks(self):
def test_profile_identity_mismatch_blocks(self):
"""#559: different daemon PIDs are OK; profile identity mismatch is not."""
import tempfile
from unittest.mock import patch
import mcp_session_state
root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
review_workflow_load.record_review_workflow_load(root)
review_workflow_load._REVIEW_WORKFLOW_LOAD["session_pid"] = 0
blockers = review_workflow_load.review_workflow_load_blockers(root)
self.assertTrue(any("different process" in b for b in blockers))
with tempfile.TemporaryDirectory() as tmp:
with patch.dict(
os.environ,
{
mcp_session_state.STATE_DIR_ENV: tmp,
mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
},
clear=False,
):
review_workflow_load.clear_review_workflow_load()
review_workflow_load.record_review_workflow_load(root)
# Corrupt the in-memory profile identity while keeping PID.
review_workflow_load._REVIEW_WORKFLOW_LOAD[
"session_profile_lock"
] = "other-profile"
review_workflow_load._REVIEW_WORKFLOW_LOAD[
"profile_identity"
] = "other-profile"
blockers = review_workflow_load.review_workflow_load_blockers(root)
self.assertTrue(
any("profile identity mismatch" in b for b in blockers),
msg=blockers,
)
review_workflow_load.clear_review_workflow_load()
def test_prompt_conflict_detected(self):
conflict, reasons = review_workflow_load.assess_prompt_conflict(
+94
View File
@@ -0,0 +1,94 @@
"""Tests for web UI lease visibility (#433)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
from webui.app import create_app
from webui.lease_loader import (
_duplicate_branch_warnings,
_duplicate_pr_warnings,
load_lease_snapshot,
parse_reviewer_lease_comment,
snapshot_to_dict,
)
class TestLeaseLoader(unittest.TestCase):
def test_parse_reviewer_lease_comment(self):
body = (
"<!-- mcp-review-lease:v1 -->\n"
"pr: #42\n"
"reviewer_identity: sysadmin\n"
"profile: prgs-reviewer\n"
"phase: validating\n"
"expires_at: 2026-07-07T20:00:00Z\n"
)
parsed = parse_reviewer_lease_comment(body)
self.assertIsNotNone(parsed)
assert parsed is not None
self.assertEqual(parsed["pr_number"], 42)
self.assertEqual(parsed["phase"], "validating")
def test_duplicate_pr_warnings(self):
raw_prs = [
{"number": 1, "title": "Closes #99", "body": ""},
{"number": 2, "title": "fixes #99", "body": ""},
]
warnings = _duplicate_pr_warnings(raw_prs)
self.assertEqual(len(warnings), 1)
self.assertEqual(warnings[0].issue_number, 99)
self.assertEqual(warnings[0].pr_numbers, (1, 2))
def test_duplicate_branch_warnings(self):
warnings = _duplicate_branch_warnings([
"feat/issue-12-a",
"feat/issue-12-b",
"master",
])
self.assertEqual(len(warnings), 1)
self.assertEqual(warnings[0].issue_number, 12)
def test_load_snapshot_with_injected_fetch(self):
def fetch_prs(_h, _o, _r, _a):
return ([], None)
def fetch_issues(_h, _o, _r, _a):
return ([], None)
snapshot = load_lease_snapshot(
fetch_prs=fetch_prs,
fetch_issues=fetch_issues,
fetch_comments=lambda *_a, **_k: [],
issue_lock_path=str(Path("/nonexistent/lock.json")),
)
data = snapshot_to_dict(snapshot)
self.assertIn("claim_inventory", data)
self.assertIn("collision_history", data)
class TestLeaseRoutes(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
def test_leases_page_renders(self):
response = self.client.get("/leases")
self.assertEqual(response.status_code, 200)
self.assertIn("Leases", response.text)
self.assertIn("Collision warnings", response.text)
self.assertNotIn("child issue", response.text.lower())
def test_api_leases_json(self):
response = self.client.get("/api/leases")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIn("claim_inventory", data)
self.assertIn("duplicate_prs", data)
self.assertIn("reviewer_leases", data)
if __name__ == "__main__":
unittest.main()
+9 -5
View File
@@ -46,10 +46,14 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertEqual(response.status_code, 200)
self.assertIn("Gitea-Tools", response.text)
def test_leases_is_implemented(self):
response = self.client.get("/leases")
self.assertEqual(response.status_code, 200)
self.assertIn("Leases", response.text)
self.assertIn("Collision warnings", response.text)
def test_extra_stub_routes(self):
for path in ("/worktrees", "/leases"):
with self.subTest(path=path):
self.assertEqual(self.client.get(path).status_code, 200)
self.assertEqual(self.client.get("/worktrees").status_code, 200)
def test_post_is_rejected(self):
response = self.client.post("/health")
@@ -62,10 +66,10 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertIn("Live queue", response.text)
def test_nav_links_on_all_pages(self):
for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit"):
for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
with self.subTest(path=path):
text = self.client.get(path).text
for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"):
for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
self.assertIn(f'href="{href}"', text)
+9 -4
View File
@@ -14,6 +14,8 @@ from webui.project_registry import find_project, load_registry, registry_to_dict
from webui.project_views import render_project_detail, render_projects_list
from webui.prompt_library import find_prompt, library_to_dict
from webui.prompt_views import render_prompt_detail, render_prompts_page
from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
from webui.lease_views import render_leases_page
from webui.queue_loader import load_queue_snapshot, snapshot_to_dict
from webui.queue_views import render_queue_page
@@ -141,10 +143,12 @@ async def worktrees(_request: Request) -> HTMLResponse:
async def leases(_request: Request) -> HTMLResponse:
return _stub_page(
"Leases",
"Lease visibility will show active issue and reviewer PR leases.",
)
snapshot = load_lease_snapshot()
return HTMLResponse(render_leases_page(snapshot))
async def api_leases(_request: Request) -> JSONResponse:
return JSONResponse(lease_snapshot_to_dict(load_lease_snapshot()))
async def method_not_allowed(request: Request, _exc: Exception) -> Response:
@@ -175,6 +179,7 @@ def create_app() -> Starlette:
Route("/audit", audit, methods=["GET"]),
Route("/worktrees", worktrees, methods=["GET"]),
Route("/leases", leases, methods=["GET"]),
Route("/api/leases", api_leases, methods=["GET"]),
],
exception_handlers={405: method_not_allowed},
)
+330
View File
@@ -0,0 +1,330 @@
"""Lease and collision visibility for the web UI (#433)."""
from __future__ import annotations
import os
import re
import subprocess
from dataclasses import dataclass
from typing import Any, Callable
from urllib.parse import urlparse
from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
from issue_claim_heartbeat import build_claim_inventory
from merged_cleanup_reconcile import ISSUE_LOCK_FILE, read_issue_lock
from webui.project_registry import ProjectRecord, load_registry
from webui.queue_loader import _extract_linked_issue, _fetch_issues, _fetch_prs
_REVIEWER_LEASE_MARKER = "<!-- mcp-review-lease:v1 -->"
_REVIEWER_FIELD_RE = re.compile(
r"^\s*([a-z_]+)\s*:\s*(.+?)\s*$",
re.IGNORECASE | re.MULTILINE,
)
_ISSUE_BRANCH_RE = re.compile(r"issue-(\d+)", re.IGNORECASE)
_COLLISION_HISTORY = (
{"number": 267, "title": "Author work leases"},
{"number": 268, "title": "Issue claim heartbeat leases"},
{"number": 400, "title": "Early duplicate-work detection"},
{"number": 407, "title": "Per-PR reviewer leases"},
)
@dataclass(frozen=True)
class CollisionWarning:
kind: str
message: str
issue_number: int | None = None
pr_numbers: tuple[int, ...] = ()
@dataclass(frozen=True)
class LeaseSnapshot:
project_id: str
repo_label: str
issue_lock: dict[str, Any] | None
claim_inventory: dict[str, Any]
reviewer_leases: tuple[dict[str, Any], ...]
duplicate_prs: tuple[CollisionWarning, ...]
duplicate_branches: tuple[CollisionWarning, ...]
collision_history: tuple[dict[str, Any], ...]
fetch_error: str | None = None
def _repo_root() -> str:
override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
if override:
return os.path.realpath(override)
return os.path.realpath(os.path.join(os.path.dirname(__file__), ".."))
def _host_from_url(remote_host: str) -> str:
parsed = urlparse(remote_host.strip())
return parsed.netloc or remote_host.strip().rstrip("/")
def _parse_pr_ref(value: str | None) -> int | None:
digits = re.sub(r"[^\d]", "", value or "")
return int(digits) if digits.isdigit() else None
def parse_reviewer_lease_comment(body: str) -> dict[str, Any] | None:
text = body or ""
if _REVIEWER_LEASE_MARKER not in text:
return None
fields: dict[str, str] = {}
for match in _REVIEWER_FIELD_RE.finditer(text):
fields[match.group(1).strip().lower()] = match.group(2).strip()
if not fields:
return None
return {
"pr_number": _parse_pr_ref(fields.get("pr")),
"issue_number": _parse_pr_ref(fields.get("issue")),
"reviewer_identity": fields.get("reviewer_identity"),
"profile": fields.get("profile"),
"phase": (fields.get("phase") or "").strip().lower() or None,
"expires_at": fields.get("expires_at"),
"blocker": fields.get("blocker"),
}
def _fetch_comments(
host: str,
org: str,
repo: str,
auth: str,
*,
issue_number: int,
) -> list[dict]:
url = f"{repo_api_url(host, org, repo)}/issues/{issue_number}/comments"
comments: list[dict] = []
page = 1
while page <= 10:
raw_page, meta = api_fetch_page(url, auth, page=page, limit=50)
comments.extend(raw_page)
if meta.get("is_final_page"):
break
page += 1
return comments
def _list_local_branch_names(project_root: str) -> list[str]:
result = subprocess.run(
["git", "-C", project_root, "branch", "--list", "--format=%(refname:short)"],
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
return []
return [line.strip() for line in (result.stdout or "").splitlines() if line.strip()]
def _duplicate_pr_warnings(raw_prs: list[dict]) -> list[CollisionWarning]:
issue_to_prs: dict[int, list[int]] = {}
for pr in raw_prs:
linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
if linked is None:
continue
issue_to_prs.setdefault(linked, []).append(int(pr["number"]))
warnings: list[CollisionWarning] = []
for issue_number, pr_numbers in sorted(issue_to_prs.items()):
if len(pr_numbers) > 1:
warnings.append(
CollisionWarning(
kind="duplicate-pr",
issue_number=issue_number,
pr_numbers=tuple(sorted(pr_numbers)),
message=(
f"Issue #{issue_number} has {len(pr_numbers)} open PRs: "
f"{', '.join(f'#{n}' for n in sorted(pr_numbers))} (#400)"
),
)
)
return warnings
def _duplicate_branch_warnings(branch_names: list[str]) -> list[CollisionWarning]:
issue_to_branches: dict[int, list[str]] = {}
for name in branch_names:
match = _ISSUE_BRANCH_RE.search(name)
if not match:
continue
issue_num = int(match.group(1))
issue_to_branches.setdefault(issue_num, []).append(name)
warnings: list[CollisionWarning] = []
for issue_number, branches in sorted(issue_to_branches.items()):
if len(branches) > 1:
warnings.append(
CollisionWarning(
kind="duplicate-branch",
issue_number=issue_number,
message=(
f"Issue #{issue_number} has {len(branches)} local branches: "
f"{', '.join(branches)}"
),
)
)
return warnings
def _extract_reviewer_leases(
raw_prs: list[dict],
*,
fetch_comments: Callable[[int], list[dict]],
) -> list[dict[str, Any]]:
leases: list[dict[str, Any]] = []
for pr in raw_prs:
pr_number = int(pr["number"])
for comment in fetch_comments(pr_number):
parsed = parse_reviewer_lease_comment(comment.get("body") or "")
if not parsed:
continue
leases.append(
{
**parsed,
"pr_number": parsed.get("pr_number") or pr_number,
"comment_id": comment.get("id"),
"author": (comment.get("user") or {}).get("login"),
"created_at": comment.get("created_at"),
}
)
active_phases = {"claimed", "validating", "approved", "request-changes", "merging"}
return [lease for lease in leases if (lease.get("phase") or "") in active_phases]
def load_lease_snapshot(
*,
project_id: str | None = None,
project_root: str | None = None,
issue_lock_path: str | None = None,
fetch_prs: Callable | None = None,
fetch_issues: Callable | None = None,
fetch_comments: Callable[[str, str, str, str, int], list[dict]] | None = None,
) -> LeaseSnapshot:
registry = load_registry()
project: ProjectRecord | None = None
if project_id:
project = next((p for p in registry.projects if p.id == project_id), None)
else:
project = registry.projects[0] if registry.projects else None
root = project_root or _repo_root()
lock = read_issue_lock(issue_lock_path if issue_lock_path is not None else ISSUE_LOCK_FILE)
if project is None:
return LeaseSnapshot(
project_id=project_id or "",
repo_label="",
issue_lock=lock,
claim_inventory={"entries": [], "counts": {}},
reviewer_leases=(),
duplicate_prs=(),
duplicate_branches=(),
collision_history=_COLLISION_HISTORY,
fetch_error="project not found in registry",
)
host = _host_from_url(project.remote_host)
pr_fetch = fetch_prs or _fetch_prs
issue_fetch = fetch_issues or _fetch_issues
comment_fetch = fetch_comments or _fetch_comments
using_live = fetch_prs is None or fetch_issues is None
auth = get_auth_header(host) if using_live else "test-auth"
if using_live and not auth:
return LeaseSnapshot(
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
issue_lock=lock,
claim_inventory={"entries": [], "counts": {}},
reviewer_leases=(),
duplicate_prs=(),
duplicate_branches=_duplicate_branch_warnings(_list_local_branch_names(root)),
collision_history=_COLLISION_HISTORY,
fetch_error=(
f"Gitea credentials unavailable for {host}; "
"remote lease artifacts cannot be loaded"
),
)
try:
raw_prs, _ = pr_fetch(host, project.gitea_owner, project.repo_name, auth)
raw_issues, _ = issue_fetch(host, project.gitea_owner, project.repo_name, auth)
except Exception as exc: # noqa: BLE001
return LeaseSnapshot(
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
issue_lock=lock,
claim_inventory={"entries": [], "counts": {}},
reviewer_leases=(),
duplicate_prs=(),
duplicate_branches=_duplicate_branch_warnings(_list_local_branch_names(root)),
collision_history=_COLLISION_HISTORY,
fetch_error=f"Gitea fetch failed: {exc}",
)
comments_by_issue: dict[int, list[dict]] = {}
for issue in raw_issues:
if not any(lb.get("name") == "status:in-progress" for lb in issue.get("labels", [])):
continue
number = int(issue["number"])
comments_by_issue[number] = comment_fetch(
host, project.gitea_owner, project.repo_name, auth, issue_number=number
)
branch_names = _list_local_branch_names(root)
inventory = build_claim_inventory(
issues=raw_issues,
comments_by_issue=comments_by_issue,
open_prs=raw_prs,
branch_names=branch_names,
)
def _pr_comments(pr_number: int) -> list[dict]:
return comment_fetch(
host, project.gitea_owner, project.repo_name, auth, issue_number=pr_number
)
reviewer_leases = tuple(_extract_reviewer_leases(raw_prs, fetch_comments=_pr_comments))
return LeaseSnapshot(
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
issue_lock=lock,
claim_inventory=inventory,
reviewer_leases=reviewer_leases,
duplicate_prs=tuple(_duplicate_pr_warnings(raw_prs)),
duplicate_branches=tuple(_duplicate_branch_warnings(branch_names)),
collision_history=_COLLISION_HISTORY,
)
def snapshot_to_dict(snapshot: LeaseSnapshot) -> dict[str, Any]:
return {
"project_id": snapshot.project_id,
"repo_label": snapshot.repo_label,
"issue_lock": snapshot.issue_lock,
"claim_inventory": snapshot.claim_inventory,
"reviewer_leases": list(snapshot.reviewer_leases),
"duplicate_prs": [
{
"kind": w.kind,
"message": w.message,
"issue_number": w.issue_number,
"pr_numbers": list(w.pr_numbers),
}
for w in snapshot.duplicate_prs
],
"duplicate_branches": [
{
"kind": w.kind,
"message": w.message,
"issue_number": w.issue_number,
}
for w in snapshot.duplicate_branches
],
"collision_history": list(snapshot.collision_history),
"fetch_error": snapshot.fetch_error,
}
+130
View File
@@ -0,0 +1,130 @@
"""HTML views for lease and collision visibility (#433)."""
from __future__ import annotations
import html
import json
from webui.layout import render_page
from webui.lease_loader import LeaseSnapshot
def _escape(text: str) -> str:
return html.escape(text, quote=True)
def _warnings_block(snapshot: LeaseSnapshot) -> str:
warnings = list(snapshot.duplicate_prs) + list(snapshot.duplicate_branches)
if not warnings:
return "<p class='muted'>No duplicate PR/branch collisions detected in current inventory.</p>"
items = "".join(f"<li>{_escape(w.message)}</li>" for w in warnings)
return f"<ul class='collision-warnings'>{items}</ul>"
def _lock_block(snapshot: LeaseSnapshot) -> str:
lock = snapshot.issue_lock
if not lock:
return "<p class='muted'>No active local issue lock file.</p>"
return (
"<pre class='prompt-text'>"
f"{_escape(json.dumps(lock, indent=2, sort_keys=True))}"
"</pre>"
)
def _claims_table(snapshot: LeaseSnapshot) -> str:
entries = snapshot.claim_inventory.get("entries") or []
if not entries:
return "<p class='muted'>No in-progress issue claims in fetched inventory.</p>"
rows = []
for entry in entries:
rows.append(
"<tr>"
f"<td>#{entry.get('issue_number')}</td>"
f"<td><span class='badge badge-{_escape(str(entry.get('status')))}'>"
f"{_escape(str(entry.get('status')))}</span></td>"
f"<td>{_escape(str(entry.get('linked_open_pr') or ''))}</td>"
f"<td>{entry.get('heartbeat_count', 0)}</td>"
f"<td>{_escape(', '.join(entry.get('matching_branches') or []) or '')}</td>"
f"<td class='muted'>{_escape('; '.join(entry.get('reasons') or []))}</td>"
"</tr>"
)
return (
"<table class='registry leases'>"
"<thead><tr><th>Issue</th><th>Claim status</th><th>Open PR</th>"
"<th>Heartbeats</th><th>Branches</th><th>Notes</th></tr></thead>"
f"<tbody>{''.join(rows)}</tbody></table>"
)
def _reviewer_leases_table(snapshot: LeaseSnapshot) -> str:
if not snapshot.reviewer_leases:
return (
"<p class='muted'>No active reviewer PR lease comments found "
"(marker <code>&lt;!-- mcp-review-lease:v1 --&gt;</code>; see #407).</p>"
)
rows = []
for lease in snapshot.reviewer_leases:
rows.append(
"<tr>"
f"<td>#{lease.get('pr_number')}</td>"
f"<td>{_escape(str(lease.get('reviewer_identity') or ''))}</td>"
f"<td><code>{_escape(str(lease.get('profile') or ''))}</code></td>"
f"<td>{_escape(str(lease.get('phase') or ''))}</td>"
f"<td><code>{_escape(str(lease.get('expires_at') or ''))}</code></td>"
"</tr>"
)
return (
"<table class='registry leases'>"
"<thead><tr><th>PR</th><th>Reviewer</th><th>Profile</th>"
"<th>Phase</th><th>Expires</th></tr></thead>"
f"<tbody>{''.join(rows)}</tbody></table>"
)
def _history_links(snapshot: LeaseSnapshot) -> str:
items = "".join(
f"<li><a href=\"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/"
f"{item['number']}\">#{item['number']}</a> — {_escape(item['title'])}</li>"
for item in snapshot.collision_history
)
return f"<ul>{items}</ul>"
LEASE_PAGE_STYLES = """
<style>
.collision-warnings li { color: #f0c4c4; margin: 0.35rem 0; }
table.leases td:nth-child(6) { font-size: 0.85rem; }
.badge-active, .badge-awaiting_review { color: #8fd19e; border-color: #3d6b4a; }
.badge-stale, .badge-phantom, .badge-reclaimable { color: #f0a8a8; border-color: #7a3b3b; }
</style>
"""
def render_leases_page(snapshot: LeaseSnapshot) -> str:
error_block = ""
if snapshot.fetch_error:
error_block = (
'<div class="stub"><p><strong>Partial load:</strong> '
f"{_escape(snapshot.fetch_error)}</p></div>"
)
body = (
"<h2>Leases &amp; collisions</h2>"
"<p>Read-only visibility for issue claims, reviewer PR leases, and "
"duplicate-work risks. Does not acquire or release leases.</p>"
f"{error_block}"
f"<p class='meta'>Project: <code>{_escape(snapshot.repo_label)}</code></p>"
"<h3>Collision warnings</h3>"
f"{_warnings_block(snapshot)}"
"<h3>Local issue lock</h3>"
f"{_lock_block(snapshot)}"
"<h3>In-progress issue claims (#268)</h3>"
f"{_claims_table(snapshot)}"
"<h3>Reviewer PR leases (#407)</h3>"
f"{_reviewer_leases_table(snapshot)}"
"<h3>Collision / lease history issues</h3>"
f"{_history_links(snapshot)}"
"<p><a href=\"/api/leases\">JSON API</a></p>"
f"{LEASE_PAGE_STYLES}"
)
return render_page(title="Leases", body_html=body)