Compare commits

...
Author SHA1 Message Date
sysadmin 36fe4785ec Merge pull request 'feat(mcp): post-restart reconciliation and completion proof (Closes #662)' (#879) from fix/issue-662-post-restart-reconcile into master 2026-07-24 09:04:23 -05:00
jcwalker3 1232789b41 Merge branch 'master' into fix/issue-662-post-restart-reconcile 2026-07-24 09:01:48 -05:00
sysadmin 2976c21ee6 Merge pull request 'test(#878): #628 building-block regression coverage (child of #628)' (#795) from feat/issue-628-autonomous-handoffs-orchestration into master 2026-07-24 08:51:56 -05:00
jcwalker3 2602605c83 Merge branch 'master' into feat/issue-628-autonomous-handoffs-orchestration 2026-07-24 08:50:34 -05:00
jcwalker3 ae31e1e852 Merge pull request 'feat(webui): model usage, token cost, latency, and performance analytics (Closes #651)' (#876) from feat/issue-651-usage-cost-analytics into master
Reviewed-on: #876
Reviewed-by: sysadmin <[email protected]>
2026-07-24 08:46:02 -05:00
jcwalker3 572a3cf8b8 Merge branch 'master' into feat/issue-628-autonomous-handoffs-orchestration 2026-07-24 08:43:12 -05:00
jcwalker3 22f9547fd8 Merge branch 'master' into feat/issue-651-usage-cost-analytics 2026-07-24 08:36:55 -05:00
sysadmin 2e4ed38c51 Merge pull request 'feat(lease): make the author task heartbeat load-bearing (Closes #790, Slice A)' (#794) from fix/issue-790-slice-a-heartbeat-policy into master 2026-07-24 08:26:09 -05:00
sysadminandClaude Opus 4.8 a7a283f449 feat(mcp): post-restart reconciliation and completion proof (Closes #662)
Add pure post_restart_reconcile.reconcile_after_restart classifier with a
machine-readable completion proof covering service health, sessions, leases,
capabilities, worktrees, interrupted mutations (never auto-resumed),
duplicates, and queue state. Soft-depends on #660 checkpoints (skipped with
reason when the schema module is absent).

Wire read-only MCP tool gitea_reconcile_after_restart, boot-once hook via
gitea_assess_master_parity, log_only/enforce modes (mutation_hold), and docs.

Closes #662
Related: #655 #652 #653 #660 #661

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-24 08:42:20 -04:00
jcwalker3 6f74552617 Merge branch 'master' into feat/issue-628-autonomous-handoffs-orchestration 2026-07-24 07:37:06 -05:00
sysadmin 38506a753f Merge remote-tracking branch 'prgs/master' into feat/issue-651-usage-cost-analytics
# Conflicts:
#	docs/webui-authz-audit.md
#	webui/console_authz.py
2026-07-24 08:33:35 -04:00
sysadmin 0987f93c67 Merge branch 'master' into fix/issue-790-slice-a-heartbeat-policy
Keep PR #794 current after master landed #863 (sanctioned restart controls).
Preserves Slice A heartbeat conflict-gate remediation for review #544.
2026-07-24 08:22:51 -04:00
sysadmin 103d0df289 Merge pull request 'feat(webui): sanctioned restart and graceful reload controls (Closes #642)' (#863) from feat/issue-642-sanctioned-restart-controls into master 2026-07-24 07:21:17 -05:00
sysadminandClaude Opus 4.8 b179610e7f fix(webui): remediate PR #876 REQUEST_CHANGES for analytics (#651)
Address review blockers and medium/low findings:

- F1: HTML-escape all dynamic analytics fields (html.escape quote=True)
- F2: Gate POST /api/v1/analytics/usage through console_authz
  record_analytics_usage (fail closed for unauthenticated / Phase 1)
- F3: Enforce usage_events retention (max rows + max age)
- F4: Coerce None remote/org/repo to empty strings in load_analytics

Add tests for XSS escaping, unauthorized ingest deny, retention, and
None scope coercion. Document authz + retention in analytics guide.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-24 08:11:12 -04:00
jcwalker3 4b436ea7d0 Merge branch 'master' into feat/issue-628-autonomous-handoffs-orchestration 2026-07-24 07:08:09 -05:00
jcwalker3 8ad3641fd3 Merge branch 'master' into fix/issue-790-slice-a-heartbeat-policy 2026-07-24 07:07:53 -05:00
jcwalker3 205207abb0 Merge branch 'master' into feat/issue-642-sanctioned-restart-controls 2026-07-24 07:03:50 -05:00
jcwalker3 0041542fe2 Merge branch 'master' into feat/issue-651-usage-cost-analytics 2026-07-24 07:03:24 -05:00
sysadmin a87a7d1da2 Merge pull request 'Implement native author issue worktree bootstrap (#850)' (#853) from fix/issue-850-native-mcp-bootstrap into master 2026-07-24 07:02:48 -05:00
sysadmin bc54effbd0 Merge branch 'master' into fix/issue-790-slice-a-heartbeat-policy 2026-07-24 08:01:27 -04:00
jcwalker3 9504fa8bbd Merge branch 'master' into fix/issue-850-native-mcp-bootstrap 2026-07-24 06:55:44 -05:00
jcwalker3 18bca47977 Merge branch 'master' into feat/issue-642-sanctioned-restart-controls 2026-07-24 06:55:15 -05:00
jcwalker3 1f144705e2 Merge branch 'master' into feat/issue-651-usage-cost-analytics 2026-07-24 06:55:03 -05:00
sysadmin e1d844bfed fix(bootstrap): path-shaped branches ancestry without isdir (#850 review #551)
resolve_canonical_repo_root fallback now uses commonpath only — no
string split and no os.path.isdir gate — so MCP project_root =
branches/<wt> still resolves to the repo root when the path is not yet
on disk (#274 / review #551 regression).
2026-07-24 07:54:07 -04:00
sysadmin 06e95254f0 fix(bootstrap): address PR #853 REQUEST_CHANGES findings (#850)
- Remove /branches/ string-split fallback in resolve_canonical_repo_root;
  recover roots via commonpath ancestry only (review #531 F2).
- Refuse existing branches that do not contain live master; no weak
  merge-base acceptance (F3).
- Verify caller-supplied assignment_id/lease_id against the control plane
  or fail closed (F4).
- Compensating recovery releases bound workflow leases via lease_lifecycle (F5).
- Regression tests for each finding.
2026-07-24 07:46:34 -04:00
sysadminandClaude Opus 4.8 b6a8989c77 fix(lease): honor heartbeat-lifecycle non-live reclaim bands in same-issue conflict gate (#790 review #502/#516)
assess_same_issue_lease_conflict entered the reclaim branch only under is_lease_expired, so a heartbeat-lifecycle lease that is non-live but whose expires_at is still in the future (stale_absolute_cap under default policy; stale_missed_heartbeat under TTL>grace) fell through to the foreign active-lease block and never reached assess_expired_lock_reclaim, which already permits those bands. The gate now also enters reclaim/renewal when a heartbeat-lifecycle lease is not is_lease_live even with a future expires_at; legacy locks are excluded and keep their absolute expires_at clock (AC-N8). Adds tests/test_issue_794_conflict_gate_reclaim.py.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-24 07:38:57 -04:00
sysadmin 3acfb039a9 fix(pr-795): honest scope for #628 building-block regression tests
Address REQUEST_CHANGES on PR #795: stop claiming all 21 umbrella ACs
from a single unit-test module. Scope docs and cases to child issue #878
(CTH/format, ownership classify_skip, dependency edges). Does not close
umbrella #628.
2026-07-24 07:35:26 -04:00
sysadmin a58b5d6e69 Merge remote-tracking branch 'prgs/master' into feat/issue-628-autonomous-handoffs-orchestration 2026-07-24 07:32:47 -04:00
sysadmin 5e935dffb4 Merge pull request 'feat(webui): system-health dashboard (Closes #639)' (#862) from feat/issue-639-webui-system-health-dashboard into master 2026-07-24 06:28:46 -05:00
jcwalker3 0ae05cb9bc Merge branch 'master' into fix/issue-850-native-mcp-bootstrap 2026-07-24 06:16:32 -05:00
jcwalker3 82464f4054 Merge branch 'master' into feat/issue-639-webui-system-health-dashboard 2026-07-24 06:16:12 -05:00
jcwalker3 2a5d6571ec Merge branch 'master' into feat/issue-642-sanctioned-restart-controls 2026-07-24 06:16:03 -05:00
jcwalker3 37c3e5dc39 Merge branch 'master' into feat/issue-642-sanctioned-restart-controls 2026-07-24 06:11:00 -05:00
sysadminandClaude Opus 4.8 2baf726ee6 merge(master): sync PR #853 with master; keep bootstrap and #860 transitions
Resolve task_capability_map.py by unioning #850 bootstrap_author_issue_worktree
preflight transitions with master's #860 dirty-orphan recovery and work_issue
commit transitions. No feature logic discarded.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-24 07:04:20 -04:00
jcwalker3 d2eaca4949 Merge branch 'master' into feat/issue-642-sanctioned-restart-controls 2026-07-24 05:54:59 -05:00
jcwalker3 d7e69fbe77 Merge branch 'master' into feat/issue-639-webui-system-health-dashboard 2026-07-24 02:07:53 -05:00
jcwalker3 4ff4d2acc9 Merge branch 'master' into feat/issue-642-sanctioned-restart-controls 2026-07-24 02:07:44 -05:00
jcwalker3 e0536d344f Merge branch 'master' into fix/issue-850-native-mcp-bootstrap 2026-07-24 02:07:03 -05:00
jcwalker3 bf5c72e3ba Merge branch 'master' into feat/issue-628-autonomous-handoffs-orchestration 2026-07-24 02:06:27 -05:00
jcwalker3 fc8fe329d2 Merge branch 'master' into feat/issue-642-sanctioned-restart-controls 2026-07-23 22:22:36 -05:00
sysadminandClaude Opus 4.8 e593444eea feat(webui): sanctioned restart and graceful reload controls (Closes #642)
Adds a gated `system.restart_namespace` action so operators and workers can
restart or gracefully reload MCP namespaces through an authorized, audited
path instead of manual host process killing (#630).

- webui/sanctioned_restart.py: restart/reload operation model, dry-run
  intent preview, confirmation-string enforcement, audit emission, and
  post-restart health verification. Fails closed on unknown auth, missing
  capability, or ambiguous target namespace.
- webui/console_authz.py: RBAC entries for the restart capability with
  secret redaction preserved.
- webui/gated_actions.py: registers the restart action in the gated action
  framework so it cannot be invoked without capability + confirmation.
- task_capability_map.py: capability mapping for the restart operation.
- docs/sanctioned-restart-controls.md: operator documentation for the
  sanctioned path and the explicit prohibition on pkill recovery.
- docs/webui-authz-audit.md: audit model updated for restart events.
- tests/test_webui_sanctioned_restart.py: authorized preview, unauthorized
  deny, confirmation enforcement, contamination classification, and audit
  emission coverage.

No unrestricted kill path is exposed; manual pkill remains classified as
contamination and continues to block clean claims.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 23:19:21 -04:00
jcwalker3 67cd2da561 fix: remediate PR #853 review #528 findings for native MCP bootstrap (#850)
- Fix module reloading bug in task capability router (F-1)
- Harden journal persistence and pending creations crash window (F-3)
- Implement dirty worktree and author commit recovery preservation (F-4)
- Fail closed on missing identity, profile, or session parameters (F-5)
- Fix branches root path traversal and symlink validation (F-6)
- Enforce O_NOFOLLOW and symlink checking on transition locks (F-7)
- Support common ancestor merge-base verification for base SHA (F-8)
- Release transition lock on compensating recovery (F-10)
- Thread journal_dir through recovery and fix guidance strings (F-11, F-12)
- Fix unittest mock import in bootstrap test suite (F-13)
2026-07-23 20:40:09 -05:00
sysadmin f80e3b33b0 Merge remote-tracking branch 'prgs/master' into feat/issue-639-webui-system-health-dashboard 2026-07-23 21:14:50 -04:00
jcwalker3 e9f6d68bd7 Merge branch 'master' into fix/issue-850-native-mcp-bootstrap 2026-07-23 19:53:00 -05:00
jcwalker3 5adc328a2b Merge branch 'master' into feat/issue-628-autonomous-handoffs-orchestration 2026-07-23 19:52:36 -05:00
jcwalker3 6a636c58e7 Merge branch 'master' into fix/issue-790-slice-a-heartbeat-policy 2026-07-23 19:52:27 -05:00
jcwalker3 b3859f6dad Merge branch 'master' into fix/issue-850-native-mcp-bootstrap 2026-07-23 19:13:38 -05:00
jcwalker3 9cca5f3dd2 Merge branch 'master' into feat/issue-628-autonomous-handoffs-orchestration 2026-07-23 19:13:09 -05:00
jcwalker3 d2fe0110a0 Merge branch 'master' into fix/issue-790-slice-a-heartbeat-policy 2026-07-23 19:13:01 -05:00
sysadminandClaude Opus 4.8 edd5f813b2 Merge master into feat/issue-639-webui-system-health-dashboard
Resolve the #638 shell landing against the #639 dashboard:

- webui/layout.py: drop the flat NAV_ITEMS tuple in favor of master's
  grouped NAV_GROUPS nav-config module.
- webui/nav.py: register /system-health as a live item in the Health
  group, satisfying issue #639 AC5 through the canonical nav source.
- docs/webui-local-dev.md: keep both additive sections (#638 shell and
  #639 dashboard).
- tests/test_webui_system_health_dashboard.py: assert the nav entry via
  iter_nav_items() instead of the removed NAV_ITEMS tuple.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 19:56:57 -04:00
jcwalker3 3b2b4e1dca Remediate PR #853 in response to review #525 for Issue #850 2026-07-23 17:49:31 -05:00
sysadmin a942afe6c4 Implement native author issue worktree bootstrap (#850) 2026-07-23 17:28:23 -04:00
jcwalker3 0254a99336 Merge branch 'master' into feat/issue-628-autonomous-handoffs-orchestration 2026-07-23 16:21:56 -05:00
sysadminandClaude Opus 4.8 ecda200180 feat(webui): system-health dashboard (Closes #639)
Phase 1 child of the Web Console epic #631. Adds the operator-facing
system-health dashboard on top of the read-only system-health API landed
by #634, so runtime problems are visible on a surface instead of being
discovered late through failed LLM sessions.

- webui/system_health_views.py (new): renders the SystemHealthSnapshot as
  readiness, stale-runtime parity, version/uptime, dependency, MCP
  namespace, probe-error, and recovery cards.
- webui/app.py: GET /system-health, sharing load_system_health() with the
  JSON API so page and API cannot disagree. ?deep=1 behaves as on the API.
- webui/layout.py: nav entry and health card/badge styles.
- tests/test_webui_system_health_dashboard.py (new, 26 cases).
- docs/webui-local-dev.md: route, field authority, and redaction split.

Readiness honesty is preserved from the API: ready and readiness_complete
render separately, a probe that did not run is listed under "Not probed"
rather than counted healthy, and mutation safety is never claimed when the
runtime is stale or parity is indeterminate.

Redaction is split by field kind. Free text (probe details, reasons, probe
errors) passes through system_health.redact. Structured fields (commit
SHAs, probe names, statuses, timestamps) are HTML-escaped only: redact's
opaque-token rule matches any run of 32 or more characters, so routing a
40-character git SHA through it rendered "[redacted]" and blanked the
parity evidence the page exists to show.

Non-goals honored: no restart or reload controls (Phase 2, #642), no
manual process-kill guidance (#630). Read-only throughout.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 16:08:36 -04:00
jcwalker3 f1e4809930 Merge branch 'master' into fix/issue-790-slice-a-heartbeat-policy 2026-07-23 12:31:44 -05:00
jcwalker3 0752b5d242 Merge branch 'master' into feat/issue-628-autonomous-handoffs-orchestration 2026-07-23 12:31:35 -05:00
jcwalker3 dc1d0e045f Merge branch 'master' into fix/issue-790-slice-a-heartbeat-policy 2026-07-23 01:13:11 -05:00
jcwalker3 6d4a0d12ec Merge branch 'master' into feat/issue-628-autonomous-handoffs-orchestration 2026-07-23 01:12:59 -05:00
jcwalker3 badc4e636b Merge branch 'master' into fix/issue-790-slice-a-heartbeat-policy 2026-07-23 00:06:18 -05:00
jcwalker3 c1d2bad901 feat(issue-628): Add unit and integration tests for autonomous canonical handoffs and task orchestration 2026-07-22 04:08:43 -05:00
sysadminandClaude Opus 4.8 243f52dc79 feat(lease): make the author task heartbeat load-bearing (#790 Slice A)
Slice A of Issue #790, per the controller reassessment in comment 13958. Does
not close the issue: terminal retirement (Slice B) and the read-side generation
check plus the #760 renewal re-scope (Slice C) are deliberately not implemented.

The defect. `issue_lock_store.assess_lock_freshness` parsed `last_heartbeat_at`
and then never consulted it. Liveness was decided by an absolute four-hour
`expires_at` and by PID liveness, and the recorded PID is the long-lived MCP
daemon rather than the authoring task, so an abandoned claim stayed live for the
full four hours. A tree-wide search found the field written in exactly one place
and advanced by nothing. Issue #787 / PR #789 hit this; Issue #760 / PR #791 hit
it again, blocking reconciliation for over five hours after its work had landed.

A1 — central policy. New `lease_policy` declares every duration for every task
class in one place: author initial/sliding TTL 10 minutes, heartbeat cadence 2,
stale warning 5, missed-heartbeat grace 10, absolute cap 8 hours, recovery grace
10, terminal race-drain 2. It ships first so the first heartbeat and TTL
behavior to run reads from it (AC-N7). The duplicated four-hour literal is gone
from both `issue_lock_store` and `gitea_mcp_server`. Reviewer, merger, and
conflict-fix classes are declared but not rewired — Slice C moves those call
sites — and a test asserts the declaration still equals the constants #747 and
`pr_work_lease` own, so the two cannot drift apart unnoticed.

A2 — load-bearing freshness, with two deliberate asymmetries. An alive PID never
establishes freshness anywhere (AC-N2); it is recorded as evidence and no branch
returns live because of it. A dead PID still marks a lease stale, and that band
still precedes every heartbeat evaluation, so #753 dead-session recovery keys on
exactly the classification it always did. New bands `stale_missed_heartbeat` and
`stale_absolute_cap` are classified in `branch_cleanup_guard` rather than
falling through to unknown-status, and still block unless the ownership record
proves `reclaim_allowed is True`. A heartbeat lease carrying no heartbeat is
contradictory and fails closed. `assess_expired_lock_reclaim` accepts a lapsed
heartbeat as reclaim grounds for heartbeat-lifecycle leases only: under this
lifecycle the heartbeat is the liveness proof, and also requiring a dead PID
would reinstate the original defect.

A3/A4 — task-session identity and the writer. `mint_task_session_id` produces an
ownership key containing no process identifier, since the daemon PID is reused
by every task it serves and identifies none of them. `heartbeat_session_lock`
writes inside the existing per-issue flock under the #772 generation
compare-and-swap, verifying exact issue, branch, realpath-normalized worktree,
claimant username, claimant profile, and recorded session identifier. It cannot
acquire, take over, or revive: a lease past its grace is refused and must use
the reclaim path, so a session that stopped proving liveness cannot restore
ownership retroactively. New `gitea_heartbeat_issue_lock` gates on the same
authority as `lock_issue`, being strictly narrower.

A5 — legacy compatibility (AC-N8). The explicit `lifecycle_version` marker, never
a timestamp comparison, discriminates legacy from heartbeat leases: a legacy lock
has `last_heartbeat_at == created_at` forever precisely because nothing advanced
it, and a freshly minted heartbeat lease has them equal too, so the equality
carries no information in either direction. Legacy locks keep their recorded
absolute expiry and are never evaluated against the short grace, so deployment
cannot make an existing claim instantly reclaimable. They leave that state only
by terminal retirement (Slice B) or by `rebind_legacy_lock`, which re-verifies
the exact owner and mints a genuine identifier and first heartbeat while
preserving the original claim under `legacy_origin`. Rebinding a lapsed legacy
lease is refused; that belongs to #760 renewal or #601 reclaim.

A6 — native coverage. Review #499 proved assessor-level tests miss discard
points, so `tests/test_issue_790_heartbeat_mcp_path.py` drives the real tools
against a real git repository and a real durable lock: lock creation and
read-back, policy window, freshness, survival of `verify_lock_for_mutation`,
invariance of the duplicate-work and linked-open-PR gates, CAS rejection,
foreign-session and foreign-claimant refusal, alive-PID-only refusal, missed
heartbeat, legacy protection on deployment, and legacy rebinding.

Tests. New suites 55 passed. Lock and lease regression set (issue_lock_store,
lease_lifecycle, #753, #755, #760 x2, #768, #772, lock registration, worktree,
adoption, duplicate gate, branch cleanup guard, capability invariants, claim
heartbeat, worktrees) 383 passed with 98 subtests. Full suite 4295 passed, 11
failed, 6 skipped, 499 subtests passed, against a clean master baseline worktree
at 620ed6e9 that reports 11 failed and 4240 passed — the same eleven node IDs.
The 55-test delta is exactly the new suites; no new failures.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_011u6GKSJwwrrYjguPjs1aK5
2026-07-22 04:19:48 -04:00
38 changed files with 8159 additions and 296 deletions
File diff suppressed because it is too large Load Diff
+80 -37
View File
@@ -40,22 +40,88 @@ def _normalize_path(path: str) -> str:
return (path or "").replace("\\", "/").rstrip("/") return (path or "").replace("\\", "/").rstrip("/")
def get_canonical_branches_root(project_root: str | None = None) -> str:
"""Return the absolute path of the canonical branches directory for *project_root*."""
root = os.path.realpath(project_root) if project_root else os.path.realpath(os.getcwd())
canonical_repo_root = resolve_canonical_repo_root(root, root)
return os.path.realpath(os.path.join(canonical_repo_root, "branches"))
def is_path_under_branches(path: str, project_root: str | None = None) -> bool: def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
"""True when *path* resolves inside ``<project_root>/branches/``.""" """True when *path* resolves inside a canonical ``branches/`` directory."""
normalized = _normalize_path(path) if not path or not str(path).strip():
if not normalized:
return False return False
if "/branches/" in f"{normalized}/": try:
return True real_path = os.path.realpath(os.path.abspath(str(path).strip()))
if normalized.endswith("/branches"): except Exception:
return True return False
if project_root:
root = _normalize_path(os.path.realpath(project_root)) branches_root = get_canonical_branches_root(project_root or real_path)
real = _normalize_path(os.path.realpath(path)) try:
if real.startswith(f"{root}/"): common = os.path.commonpath([branches_root, real_path])
rel = real[len(root) + 1 :] except Exception:
return rel == "branches" or rel.startswith("branches/") return False
return False
if common != branches_root:
return False
rel = os.path.relpath(real_path, branches_root)
return rel != "." and not rel.startswith("..")
def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str:
"""Return the stable repository root for *workspace_path* via git metadata (#460)."""
p = (workspace_path or "").strip()
if p:
try:
res = subprocess.run(
["git", "-C", p, "rev-parse", "--git-common-dir"],
capture_output=True,
text=True,
check=True,
)
common = _realpath_git_common_dir(p, res.stdout)
if common.endswith(f"{os.sep}.git") or os.path.basename(common) == ".git":
candidate_root = os.path.dirname(common)
real_p = os.path.realpath(p)
try:
if os.path.commonpath([candidate_root, real_p]) == candidate_root:
return candidate_root
except Exception:
pass
except Exception:
pass
# Fallback when git metadata is unavailable. Never string-split on
# "/branches/" (review #531 F2 / #551): recover the repo root only via
# resolved-path commonpath ancestry. Do **not** require on-disk isdir —
# MCP may launch with project_root = branches/<wt> before that path
# exists, and #274 path-shaped worktree-as-project-root must still resolve.
fallback = os.path.realpath(fallback_project_root or workspace_path or ".")
cur = fallback
for _ in range(64):
parent = os.path.dirname(cur)
if parent == cur:
break
branches_dir = os.path.realpath(os.path.join(parent, "branches"))
try:
# Path-shaped: fallback is under parent/branches/ (commonpath).
if os.path.commonpath([branches_dir, fallback]) == branches_dir:
return parent
except ValueError:
pass
# Fallback path itself is the branches directory.
if os.path.basename(os.path.realpath(cur)) == "branches":
try:
if os.path.commonpath([os.path.realpath(cur), fallback]) == os.path.realpath(
cur
):
return parent
except ValueError:
pass
cur = parent
return fallback
def resolve_mutation_workspace( def resolve_mutation_workspace(
@@ -87,29 +153,6 @@ def _realpath_git_common_dir(workspace_path: str, common_dir: str) -> str:
return os.path.realpath(os.path.join(workspace_path, raw)) return os.path.realpath(os.path.join(workspace_path, raw))
def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str:
"""Return the stable repository root for *workspace_path* via git metadata (#460)."""
path = (workspace_path or "").strip()
fallback = os.path.realpath(fallback_project_root)
if not path:
return fallback
try:
res = subprocess.run(
["git", "-C", path, "rev-parse", "--git-common-dir"],
capture_output=True,
text=True,
check=True,
)
common = _realpath_git_common_dir(path, res.stdout)
except Exception:
return fallback
if common.endswith(f"{os.sep}.git"):
return os.path.dirname(common)
if os.path.basename(common) == ".git":
return os.path.dirname(common)
return fallback
def resolve_author_mutation_context( def resolve_author_mutation_context(
worktree_path: str | None, worktree_path: str | None,
process_project_root: str, process_project_root: str,
+13 -1
View File
@@ -163,7 +163,19 @@ _TERMINAL_OWNERSHIP_STATUSES = frozenset(
{"released", "abandoned", "done", "blocked", "terminal", "closed"} {"released", "abandoned", "done", "blocked", "terminal", "closed"}
) )
_EXPIRED_STATUSES = frozenset({"expired"}) _EXPIRED_STATUSES = frozenset({"expired"})
_STALE_STATUSES = frozenset({"stale", "stale_dead_process", "stale_missing_worktree"}) _STALE_STATUSES = frozenset(
{
"stale",
"stale_dead_process",
"stale_missing_worktree",
# #790 Slice A heartbeat-lifecycle bands. Listed here so they are
# *classified* rather than falling through to the unknown-status branch;
# they still block unless the ownership record proves
# ``reclaim_allowed is True``, so the O2 fail-closed rule is unchanged.
"stale_missed_heartbeat",
"stale_absolute_cap",
}
)
def _norm_str(value: Any) -> str: def _norm_str(value: Any) -> str:
+34 -1
View File
@@ -577,6 +577,11 @@ class ControlPlaneDB:
conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_role_model ON usage_events(role, model);") conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_role_model ON usage_events(role, model);")
conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_stage ON usage_events(stage);") conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_stage ON usage_events(stage);")
# #651 retention: cap growth so unauthenticated or high-volume ingest
# cannot DoS the control-plane DB (PR #876 F3). Applied after every write.
USAGE_EVENTS_MAX_ROWS = 10_000
USAGE_EVENTS_RETENTION_DAYS = 90
def record_usage_event( def record_usage_event(
self, self,
*, *,
@@ -649,7 +654,35 @@ class ControlPlaneDB:
ts, ts,
), ),
) )
return cursor.lastrowid usage_id = cursor.lastrowid
self._enforce_usage_events_retention(conn)
return usage_id
def _enforce_usage_events_retention(self, conn: sqlite3.Connection) -> None:
"""Drop aged and excess usage_events rows (PR #876 F3)."""
# Age-based: ISO-8601 UTC timestamps compare lexicographically.
cutoff = (
datetime.now(timezone.utc)
- timedelta(days=int(self.USAGE_EVENTS_RETENTION_DAYS))
).strftime("%Y-%m-%dT%H:%M:%SZ")
conn.execute(
"DELETE FROM usage_events WHERE created_at < ?",
(cutoff,),
)
# Count-based: keep the newest USAGE_EVENTS_MAX_ROWS by usage_id.
max_rows = int(self.USAGE_EVENTS_MAX_ROWS)
if max_rows > 0:
conn.execute(
"""
DELETE FROM usage_events
WHERE usage_id NOT IN (
SELECT usage_id FROM usage_events
ORDER BY usage_id DESC
LIMIT ?
)
""",
(max_rows,),
)
def query_usage_events( def query_usage_events(
self, self,
+2 -1
View File
@@ -247,7 +247,8 @@ def bootstrap_permits_control_checkout(
""" """
if not isinstance(assessment, dict): if not isinstance(assessment, dict):
return False return False
if not is_create_issue_task(task): import author_issue_bootstrap
if not is_create_issue_task(task) and not author_issue_bootstrap.is_author_issue_bootstrap_task(task):
return False return False
# Positive proof: the assessment must affirmatively allow, with no # Positive proof: the assessment must affirmatively allow, with no
+2
View File
@@ -69,6 +69,7 @@ that gates each call, not which tools exist.
- `gitea_audit_worktree_cleanup` - `gitea_audit_worktree_cleanup`
- `gitea_authorize_reconciliation_cleanup_phase` - `gitea_authorize_reconciliation_cleanup_phase`
- `gitea_authorize_review_correction` - `gitea_authorize_review_correction`
- `gitea_bootstrap_author_issue_worktree`
- `gitea_capability_stop_terminal_report` - `gitea_capability_stop_terminal_report`
- `gitea_capture_branches_worktree_snapshot` - `gitea_capture_branches_worktree_snapshot`
- `gitea_check_pr_eligibility` - `gitea_check_pr_eligibility`
@@ -100,6 +101,7 @@ that gates each call, not which tools exist.
- `gitea_get_profile` - `gitea_get_profile`
- `gitea_get_runtime_context` - `gitea_get_runtime_context`
- `gitea_get_shell_health` - `gitea_get_shell_health`
- `gitea_heartbeat_issue_lock`
- `gitea_heartbeat_reviewer_pr_lease` - `gitea_heartbeat_reviewer_pr_lease`
- `gitea_inspect_workflow_lease` - `gitea_inspect_workflow_lease`
- `gitea_issue_irrecoverable_provenance_authorization` - `gitea_issue_irrecoverable_provenance_authorization`
@@ -89,11 +89,17 @@ record_usage(
) )
``` ```
### HTTP Ingestion API ### HTTP Ingestion API (authorized write)
`POST /api/v1/analytics/usage` is a **gated write**. It runs through
`console_authz` action `record_analytics_usage` (operator+, Phase 2 execution).
Unauthenticated or phase-inactive requests receive **403** and do not write.
Prefer in-process `record_usage` for MCP / session instrumentation.
```http ```http
POST /api/v1/analytics/usage HTTP/1.1 POST /api/v1/analytics/usage HTTP/1.1
Content-Type: application/json Content-Type: application/json
# Requires authenticated principal with record_analytics_usage execution enabled
{ {
"remote": "dadeschools", "remote": "dadeschools",
@@ -114,6 +120,18 @@ Content-Type: application/json
} }
``` ```
### Retention
`usage_events` is retained with hard caps applied on every write:
| Limit | Default |
|---|---|
| Max rows | 10,000 (`ControlPlaneDB.USAGE_EVENTS_MAX_ROWS`) |
| Max age | 90 days (`ControlPlaneDB.USAGE_EVENTS_RETENTION_DAYS`) |
Older rows (by `created_at`) and excess oldest rows (by `usage_id`) are deleted
after each insert so unbounded growth / DoS-by-volume cannot fill the DB.
--- ---
## 6. Querying Analytics API ## 6. Querying Analytics API
+56
View File
@@ -0,0 +1,56 @@
# Post-restart MCP reconciliation (#662)
After an MCP process restart, sessions, leases, capabilities, worktrees, and
interrupted mutations must be reconciled before operators claim a clean runtime.
This document describes the #662 completion-proof path.
## Components
| Piece | Where | Responsibility |
|-------|-------|----------------|
| `post_restart_reconcile.reconcile_after_restart` | `post_restart_reconcile.py` | Pure classification: inventory → completion proof DTO. No I/O. |
| `RestartCompletionProof` | `post_restart_reconcile.py` | Machine-readable proof (`.as_dict()` is JSON-serializable). |
| `gitea_reconcile_after_restart` | `gitea_mcp_server.py` | MCP tool: gathers inventory from the #613 control-plane DB + master-parity, classifies, returns the proof. Read-only. |
| Boot hook | `gitea_assess_master_parity` | First post-restart parity probe also runs reconcile once (log-only by default). |
## Dimensions
The assessor classifies:
- **service_health** — process healthy / parity mutation-safe
- **clients** — connected client descriptors (optional inventory)
- **sessions** — active session rows with dead owner pids are unresolved
- **checkpoints** — soft-depends on #660; skipped with reason when schema absent
- **leases** — live control-plane leases after restart
- **capabilities** — master-parity / stale-runtime (#610)
- **worktrees** — lease-bound paths missing on disk
- **interrupted_mutations** — mutating lease phases or explicit pending inventory; **never auto-resumed**
- **duplicates** — multiple live claims on the same work item
- **queue** — allocator resume safety
## Modes
| Mode | Env / arg | Behavior |
|------|-----------|----------|
| `log_only` (default) | unset or `GITEA_POST_RESTART_RECONCILE_MODE=log_only` | Proof only; `mutation_hold=false` |
| `enforce` | `GITEA_POST_RESTART_RECONCILE_MODE=enforce` or `mode=enforce` | Sets `mutation_hold=true` when overall status is degraded/failed or interrupted mutations remain |
## Follow-up issues
Unresolved dimensions produce `proposed_follow_ups` entries suitable for durable
Gitea issues. The MCP tool **does not create** those issues in v1 (rollout is
log-only first). Controllers may file them from the proof payload.
## Links
- Umbrella: #655
- Vision: #652 · Roadmap: #653
- Checkpoint schema: #660 (soft dependency)
- Drain proof: #661 (soft)
- This issue: #662
## Non-goals
- HA multi-instance failover
- Automatic silent mutation replay
- Implementing the #660 checkpoint schema itself
+122
View File
@@ -0,0 +1,122 @@
# Sanctioned restart and graceful reload controls (#642)
Sessions used to recover MCP connectivity by killing the host daemon
(`pkill -f mcp_server.py`, #630). That is forbidden and stays forbidden: it
kills every namespace on the host, contaminates whichever session survives, and
leaves no audit trail. This document describes the sanctioned replacement,
implemented in `webui/sanctioned_restart.py`.
## What the console will and will not do
The console **never** restarts anything. It authorizes an intent, records it,
and hands off to a host supervisor. There is no code path in which the console
sends a signal, spawns a process, or renders a kill command — a regression test
asserts the module contains no `subprocess`, `signal`, `os.kill`, `os.system`,
or `popen` reference, and that no returned payload contains a kill command.
## Operations
| Mode | Action | Minimum role | Behaviour |
|------|--------|--------------|-----------|
| `reload` | `system.reload_namespace` | controller | Host supervisor reloads the namespace in place, draining in-flight requests. |
| `restart` | `system.restart_namespace` | admin | Host supervisor restarts the namespace. In-flight requests are lost. |
Scope is always exactly one namespace. A fleet-wide restart is an explicit
non-goal: `all`, `*`, `fleet`, and an empty scope are refused with
`fleet_scope_not_permitted`, because that is precisely the blast radius the
forbidden kill already had. An unrecognised namespace is refused rather than
passed through to the host.
## The gate sequence
`assess_restart_request()` applies every gate in order and reports the first
failure with a stable reason code:
| Order | Gate | Reason code on failure |
|-------|------|------------------------|
| 1 | Mode is `restart` or `reload` | `unknown_mode` |
| 2 | Scope is a single known namespace | `fleet_scope_not_permitted`, `unknown_namespace` |
| 3 | Principal holds the required console role | `unauthorized` |
| 4 | Confirmation phrase supplied | `confirmation_required` |
| 5 | Confirmation names this namespace and mode | `confirmation_mismatch` |
| 6 | Out-of-band operator authorization present | `operator_authorization_missing` |
| 7 | Runtime is not contaminated | `contaminated_runtime` |
| 8 | Host restart hook configured | `restart_hook_not_configured` |
Passing every gate yields `host_action_required`, never "restarted".
### Confirmation binds the namespace
The required phrase is `"<mode> <namespace>"` — for example
`restart gitea-author`. Binding the namespace into the phrase is the point: a
confirmation typed for one namespace cannot be replayed against another.
### Operator authorization is not self-assertable
Host daemon maintenance is authorized out of band through
`GITEA_OPERATOR_DAEMON_MAINTENANCE_AUTHORIZATION`, read from the process
environment and nowhere else (#630; #710 finding F1). A worker session cannot
set an environment variable for an already-running daemon, so this cannot be
faked the way a tool argument could.
### The host hook
`GITEA_SANCTIONED_RESTART_HOOK` holds an opaque reference the *host* resolves —
a supervisor label such as a launchd job name, never a command line. With no
hook configured the request is refused; the console does not fall back to a
process kill. The value is read server-side and never rendered to a client.
## Manual kill remains contamination
`classify_restart_command()` classifies an operator-proposed recovery command.
A manual `pkill`/`kill`/`killall` of the MCP daemon is contamination, not a
restart: it returns `clean_claim_allowed: false` and builds a durable
contamination marker (redacted command only, never secrets) naming
`system.restart_namespace` as the sanctioned alternative.
A live, uncleared contamination marker also blocks a restart. This is stricter
than #630's task-scoped gate, which deliberately lets a contaminated worker keep
commenting and handing off: restarting a contaminated runtime would launder the
contamination rather than resolve it. Clear the marker through the reconciler
path first.
## Post-restart health verification
After the host supervisor acts, `verify_post_restart_health()` decides whether
the session may claim to be clean:
| Status | Meaning | Clean claim |
|--------|---------|-------------|
| `clean` | Required tool callable, proven through the live client namespace | Allowed |
| `unproven` | Reported healthy without live client-namespace evidence | Refused |
| `unhealthy` | Probe failed | Refused |
Only `probe_source=client_namespace` evidence clears a session. Static tool
registration is not proof, and neither is an offline subprocess probe — an IDE
client can hold a registered tool list while live calls fail with
`client is closing: EOF` (see
[`mcp-namespace-health.md`](mcp-namespace-health.md)).
## Audit
Every attempt — allowed or denied — is recorded through
`webui.console_audit` with actor, target namespace, mode, result, and reason
code, and is redacted before it is persisted. `system.restart_namespace` is
break-glass, so its records are retained for 730 days. Records carry
`process_kill_executed: false`, which is a fact about the code path rather than
a claim: no such path exists.
## Environment variables
| Variable | Purpose |
|----------|---------|
| `GITEA_SANCTIONED_RESTART_HOOK` | Host supervisor reference; absent means restart is refused. |
| `GITEA_OPERATOR_DAEMON_MAINTENANCE_AUTHORIZATION` | Out-of-band operator authorization reference. |
| `WEBUI_AUDIT_LOG` | Console audit sink; absent means records are built but not persisted. |
## Non-goals
* No unrestricted `kill` from the UI, in any role, in any phase.
* No fleet-wide restart.
* No silent auto-restart loop: every attempt is confirmed and audited.
* This does not implement the Phase 1 health API (#634).
+12 -2
View File
@@ -91,6 +91,9 @@ already define, and a regression test asserts each mapping matches.
| `close_pr` | controller | privileged | `gitea.pr.close` | Yes | No | No | 3 | | `close_pr` | controller | privileged | `gitea.pr.close` | Yes | No | No | 3 |
| `merge_pr` | controller | privileged | `gitea.pr.merge` | Yes | **Yes** | **Yes** | 3 | | `merge_pr` | controller | privileged | `gitea.pr.merge` | Yes | **Yes** | **Yes** | 3 |
| `delete_branch` | admin | destructive | `gitea.branch.delete` | Yes | **Yes** | **Yes** | 3 | | `delete_branch` | admin | destructive | `gitea.branch.delete` | Yes | **Yes** | **Yes** | 3 |
| `record_analytics_usage` | operator | gated_write | `runtime.record_analytics_usage` | Yes | No | No | 2 |
| `system.reload_namespace` | controller | privileged | `runtime.reload_namespace` | Yes | No | No | 2 |
| `system.restart_namespace` | admin | destructive | `runtime.restart_namespace` | Yes | **Yes** | **Yes** | 2 |
**Dual control** means the acting principal may not be the sole authority: a **Dual control** means the acting principal may not be the sole authority: a
second distinct principal must confirm. **Break-glass** means the action is second distinct principal must confirm. **Break-glass** means the action is
@@ -102,6 +105,13 @@ honouring it.
`delete_branch` is admin-only rather than controller because it is the one `delete_branch` is admin-only rather than controller because it is the one
irreversible action in the set. irreversible action in the set.
`system.restart_namespace` is admin-only for the same reason: restarting a
namespace drops every in-flight request on it. `system.reload_namespace` drains
first, so it is privileged but not destructive. Neither action is ever executed
by the console — both hand off to a host supervisor, and neither exposes a raw
process kill. See
[`sanctioned-restart-controls.md`](sanctioned-restart-controls.md) (#642).
### Authorization decision ### Authorization decision
`authorize(action_id, principal, for_execution=False)` returns a decision `authorize(action_id, principal, for_execution=False)` returns a decision
@@ -199,8 +209,8 @@ breaking the request it describes.
| Class | Applies to | Default | | Class | Applies to | Default |
|-------|-----------|---------| |-------|-----------|---------|
| `standard` | Routine gated writes | 90 days | | `standard` | Routine gated writes | 90 days |
| `privileged` | `review_pr`, `close_pr`, and any unclassifiable action | 365 days | | `privileged` | `review_pr`, `close_pr`, `system.reload_namespace`, and any unclassifiable action | 365 days |
| `break_glass` | `merge_pr`, `delete_branch` | 730 days | | `break_glass` | `merge_pr`, `delete_branch`, `system.restart_namespace` | 730 days |
Each record carries its own class, day count, and computed `expires_at`, so Each record carries its own class, day count, and computed `expires_at`, so
retention is auditable per record rather than inferred from file age. An retention is auditable per record rather than inferred from file age. An
+32
View File
@@ -54,6 +54,7 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
| `/` | Home / operator overview | | `/` | Home / operator overview |
| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`, `uptime_seconds`) | | `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`, `uptime_seconds`) |
| `/api/v1/system/health` | Structured read-only system health (#634) | | `/api/v1/system/health` | Structured read-only system health (#634) |
| `/system-health` | System-health dashboard — readiness, version/uptime, dependencies, MCP namespaces, stale-runtime parity (#639) |
| `/queue` | Live PR and issue queue dashboard (#429) | | `/queue` | Live PR and issue queue dashboard (#429) |
| `/api/queue` | JSON queue export with pagination metadata | | `/api/queue` | JSON queue export with pagination metadata |
| `/projects` | Project registry list with status and onboarding progress (#427, #635) | | `/projects` | Project registry list with status and onboarding progress (#427, #635) |
@@ -258,6 +259,37 @@ Not-yet-implemented surfaces (`/sessions`, `/inventory`, `/timeline`,
surfaces are backed by #636). Mutating methods on stub routes still fail closed surfaces are backed by #636). Mutating methods on stub routes still fail closed
with `read-only-mvp`. with `read-only-mvp`.
## System-health dashboard (#639)
`/system-health` renders the same snapshot the `/api/v1/system/health` API
returns, so the page and the API can never disagree. Cards: overall readiness,
stale-runtime parity, version and uptime, dependency probes, MCP namespaces,
probe errors (only when present), and recovery pointers. `?deep=1` opts into
the network probe exactly as the API does; the plain page load stays cheap.
Field authority and honesty rules:
* `ready` and `readiness_complete` are shown separately. A snapshot whose
required probes never ran is not the same as one that ran them and passed,
and the page never collapses the two into an unproven green.
* A probe that did not run appears under **Not probed**, never as healthy.
* `stale_runtime.mutation_safe` is displayed verbatim from the API. When the
runtime is stale, or when parity is indeterminate, the page warns and does
not claim mutation safety.
* MCP namespaces are reported `unproven`: the web process runs outside the
IDE-managed MCP client and cannot prove that path (#543).
Redaction is split by field kind. Free text — probe details, readiness and
parity reasons, probe errors — passes through `system_health.redact`.
Structured fields — commit SHAs, probe names, statuses, timestamps — are
HTML-escaped only, because `redact`'s opaque-token rule matches any run of 32
or more characters and would otherwise blank every 40-character git SHA, which
is precisely the evidence the parity view exists to show.
The dashboard is read-only: no restart, reload, or process-kill control. Those
arrive in Phase 2 (#642). Recovery guidance points at the sanctioned client
reconnect / operator restart path — never a manual daemon kill (#630).
## Deployment boundary (#435) ## Deployment boundary (#435)
MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused** MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**
+668 -145
View File
@@ -977,6 +977,32 @@ def _create_issue_bootstrap_assessment(
""" """
import create_issue_bootstrap as _cib import create_issue_bootstrap as _cib
import author_issue_bootstrap as _aib
if _aib.is_author_issue_bootstrap_task(task):
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
remote_master_sha_error: str | None = None
try:
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
ctx["canonical_repo_root"]
)
except Exception as exc:
remote_master_sha = None
remote_master_sha_error = (
f"{type(exc).__name__}: {exc}".strip() or "resolver failed"
)
return _aib.assess_author_issue_bootstrap(
workspace_path=workspace,
canonical_repo_root=ctx["canonical_repo_root"],
current_branch=git_state.get("current_branch"),
head_sha=git_state.get("head_sha"),
porcelain_status=git_state.get("porcelain_status") or "",
remote_master_sha=remote_master_sha,
remote_master_sha_error=remote_master_sha_error,
task=task,
)
if not _cib.is_create_issue_task(task): if not _cib.is_create_issue_task(task):
return None return None
@@ -2039,6 +2065,7 @@ import allocator_dependencies # noqa: E402
import dependency_graph # noqa: E402 # #784 durable dependency edges import dependency_graph # noqa: E402 # #784 durable dependency edges
import control_plane_db # noqa: E402 import control_plane_db # noqa: E402
import lease_lifecycle # noqa: E402 import lease_lifecycle # noqa: E402
import lease_policy # noqa: E402
import workflow_dashboard # noqa: E402 # #605 live queue/lease dashboard import workflow_dashboard # noqa: E402 # #605 live queue/lease dashboard
import restart_coordinator # noqa: E402 # #658 MCP restart coordinator/impact import restart_coordinator # noqa: E402 # #658 MCP restart coordinator/impact
import incident_bridge # noqa: E402 import incident_bridge # noqa: E402
@@ -2284,7 +2311,6 @@ import canonical_comment_validator as ccv # noqa: E402
# GITEA_ISSUE_LOCK_DIR, bound to the current MCP session via a per-PID pointer. # GITEA_ISSUE_LOCK_DIR, bound to the current MCP session via a per-PID pointer.
# Legacy global path retained only for test/doc references — do not seed manually. # Legacy global path retained only for test/doc references — do not seed manually.
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json" ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
WORK_LEASE_TTL_HOURS = 4
AUTHOR_ISSUE_WORK_LEASE = "author_issue_work" AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
VALID_WORK_LEASE_OPERATIONS = frozenset({ VALID_WORK_LEASE_OPERATIONS = frozenset({
AUTHOR_ISSUE_WORK_LEASE, AUTHOR_ISSUE_WORK_LEASE,
@@ -2599,7 +2625,12 @@ def _build_author_issue_work_lease(
host: str | None, host: str | None,
) -> dict: ) -> dict:
created = _work_lease_now() created = _work_lease_now()
expires = created + timedelta(hours=WORK_LEASE_TTL_HOURS) # #790 Slice A: the window comes from the central policy, not a literal here.
# It is also now a *sliding* window — the lease lives ``initial_ttl_minutes``
# past its last valid heartbeat rather than a fixed four hours past its
# creation, so an abandoned task stops holding the claim within one TTL.
policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
expires = created + timedelta(minutes=policy.initial_ttl_minutes)
return { return {
"operation_type": AUTHOR_ISSUE_WORK_LEASE, "operation_type": AUTHOR_ISSUE_WORK_LEASE,
"issue_number": issue_number, "issue_number": issue_number,
@@ -2610,6 +2641,15 @@ def _build_author_issue_work_lease(
"created_at": _work_lease_timestamp(created), "created_at": _work_lease_timestamp(created),
"expires_at": _work_lease_timestamp(expires), "expires_at": _work_lease_timestamp(expires),
"last_heartbeat_at": _work_lease_timestamp(created), "last_heartbeat_at": _work_lease_timestamp(created),
# #790 AC-N1: the ownership key for this task. Distinct from the recorded
# PID, which is the shared daemon and identifies no individual task.
"task_session_id": issue_lock_store.mint_task_session_id(
AUTHOR_ISSUE_WORK_LEASE
),
# #790 AC-N8: the explicit lifecycle marker. Its absence — never a
# timestamp comparison — is what makes a lock legacy.
"lifecycle_version": lease_policy.LIFECYCLE_HEARTBEAT_V1,
"heartbeat_count": 1,
} }
@@ -4380,6 +4420,135 @@ def gitea_lock_issue(
@mcp.tool() @mcp.tool()
def gitea_heartbeat_issue_lock(
issue_number: int,
branch_name: str,
task_session_id: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
expected_generation: int | None = None,
) -> dict:
"""Prove an owned author issue lease is still active (#790 Slice A).
The task-liveness signal the lifecycle was missing. Before this, an author
lease carried a fixed four-hour expiry that nothing could shorten, and the
only liveness evidence was the recorded PID the long-lived MCP daemon,
which stays alive across every task it serves and so proved nothing about
whether the authoring task still held the work.
Each successful call slides the lease ``initial_ttl_minutes`` past *now*
from the central policy, so an actively heartbeating session is never
evicted while an abandoned one releases its claim within one TTL.
What this tool cannot do, by construction:
* **Acquire.** It refuses when no durable lock exists.
* **Take over.** Exact issue, branch, realpath-normalized worktree,
claimant username, claimant profile, and recorded task-session identifier
must all match; a superseded session holding an older identifier is
refused.
* **Revive.** A lease already past its grace is not heartbeatable that
would let a session restore ownership it had stopped proving. It must use
the sanctioned reclaim path, which mints a new generation.
A lock predating the heartbeat lifecycle is rebound rather than heartbeated:
its exact owner is re-verified and a genuine task-session identifier and
first heartbeat are minted (#790 AC-N8). The rebind is decided server-side
from the durable lifecycle marker; there is no caller-facing switch.
Args:
issue_number: The locked issue number.
branch_name: The branch recorded on the lock.
task_session_id: The identifier this session received when it acquired
or rebound the lock. It is a fencing token, not an ownership
assertion: it is compared against durable state and can only ever
cause a refusal, never grant anything. Omitted only when rebinding a
legacy lock, which has no identifier yet and mints one.
remote: Known instance 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Author worktree recorded on the lock.
expected_generation: Optional fencing value. The per-issue flock already
serializes the read and the write, so this is for a caller that
wants to pin the generation it last observed across calls; a moved
generation fails closed.
Returns:
dict with 'success', 'performed', the sliding 'expires_at',
'last_heartbeat_at', 'lock_generation', 'task_session_id', the applied
'policy', and post-write 'freshness'; on refusal 'success'/'performed'
False with 'reasons' naming exactly what did not match.
"""
blocked = _profile_permission_block(
task_capability_map.required_permission("heartbeat_issue_lock"),
issue_number=issue_number,
remote=remote,
host=host,
org=org,
repo=repo,
org_explicit=org is not None,
repo_explicit=repo is not None,
)
if blocked:
return blocked
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
worktree_path, _canonical_local_git_root()
)
h, o, r = _resolve(remote, host, org, repo)
claimant = _work_lease_claimant(h)
identity = claimant.get("username")
profile = claimant.get("profile")
existing = _load_existing_issue_lock(
remote=remote, org=o, repo=r, issue_number=issue_number
)
if not existing:
return {
"success": False,
"performed": False,
"issue_number": issue_number,
"reasons": [
f"no durable lock for issue #{issue_number}; heartbeat cannot "
"acquire a claim (fail closed)"
],
}
if issue_lock_store.is_legacy_lease(existing):
outcome = issue_lock_store.rebind_legacy_lock(
remote=remote,
org=o,
repo=r,
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
identity=identity,
profile=profile,
expected_generation=expected_generation,
)
outcome["operation"] = "legacy_rebind"
return outcome
outcome = issue_lock_store.heartbeat_session_lock(
remote=remote,
org=o,
repo=r,
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
identity=identity,
profile=profile,
task_session_id=str(task_session_id or ""),
expected_generation=expected_generation,
)
outcome["operation"] = "heartbeat"
return outcome
@mcp.tool() @mcp.tool()
def gitea_recover_dirty_orphaned_issue_worktree( def gitea_recover_dirty_orphaned_issue_worktree(
issue_number: int, issue_number: int,
@@ -4653,6 +4822,7 @@ def gitea_recover_dirty_orphaned_issue_worktree(
) )
return result return result
@mcp.tool() @mcp.tool()
def gitea_rebind_dirty_same_claimant_author_session( def gitea_rebind_dirty_same_claimant_author_session(
issue_number: int, issue_number: int,
@@ -4909,8 +5079,6 @@ def gitea_rebind_dirty_same_claimant_author_session(
} }
return result return result
return result
@mcp.tool() @mcp.tool()
def gitea_assess_work_issue_duplicate( def gitea_assess_work_issue_duplicate(
@@ -9580,6 +9748,7 @@ def gitea_commit_files(
host: str | None = None, host: str | None = None,
org: str | None = None, org: str | None = None,
repo: str | None = None, repo: str | None = None,
worktree_path: str | None = None,
) -> dict: ) -> dict:
"""Commit changes to multiple files in a Gitea repository in a single atomic commit. """Commit changes to multiple files in a Gitea repository in a single atomic commit.
@@ -9592,10 +9761,46 @@ def gitea_commit_files(
host: Override the Gitea host. host: Override the Gitea host.
org: Override the owner/organization. org: Override the owner/organization.
repo: Override the repository name. repo: Override the repository name.
worktree_path: Optional worktree path for author mutation context.
Returns: Returns:
dict with success status and commit/branch information. dict with success status and commit/branch information.
""" """
if worktree_path is None:
lock_data = issue_lock_store.read_session_issue_lock() or {}
worktree_path = lock_data.get("worktree_path")
if not worktree_path:
try:
prof = get_profile()
uname = prof.get("username") or prof.get("profile_name")
for path in issue_lock_store.iter_lock_files():
lk = issue_lock_store.read_lock_file(path) or {}
claimant = lk.get("claimant") or {}
if lk.get("remote") == remote and (claimant.get("username") == uname or lk.get("profile") == prof.get("profile_name")):
issue_lock_store.bind_session_lock(lk, renewal_sanctioned=True)
worktree_path = lk.get("worktree_path")
break
except Exception:
pass
if worktree_path is None and files:
for f in files:
p = f.get("workspace_path") or f.get("local_path") or ""
if p and os.path.isabs(p):
real_p = os.path.realpath(p)
real_root = os.path.realpath(PROJECT_ROOT)
branches_dir = os.path.join(real_root, "branches")
if real_p.startswith(branches_dir + os.sep):
rel_sub = os.path.relpath(real_p, branches_dir)
wt_folder = rel_sub.split(os.sep)[0]
if wt_folder and wt_folder != "..":
worktree_path = os.path.join(branches_dir, wt_folder)
break
if worktree_path:
os.environ["GITEA_AUTHOR_WORKTREE"] = worktree_path
os.environ["GITEA_ACTIVE_WORKTREE"] = worktree_path
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop( ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
"commit_files" "commit_files"
) )
@@ -9608,7 +9813,7 @@ def gitea_commit_files(
"reasons": block_reasons, "reasons": block_reasons,
} }
blocked = _namespace_mutation_block( blocked = _namespace_mutation_block(
"commit_files", commit="", branch="", remote=remote "commit_files", commit="", branch="", remote=remote, worktree_path=worktree_path
) )
if blocked: if blocked:
return blocked return blocked
@@ -9636,7 +9841,7 @@ def gitea_commit_files(
) )
# #735: forward explicit org/repo into shared anti-stomp preflight. # #735: forward explicit org/repo into shared anti-stomp preflight.
verify_preflight_purity(remote, task="commit_files", org=org, repo=repo) verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo)
processed_files, source_proofs = _prepare_commit_payload_files(files) processed_files, source_proofs = _prepare_commit_payload_files(files)
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo)
@@ -9913,6 +10118,96 @@ def gitea_publish_unpublished_issue_branch(
} }
@mcp.tool()
def gitea_bootstrap_author_issue_worktree(
issue_number: int,
assignment_id: str | None = None,
lease_id: str | None = None,
expected_base_sha: str | None = None,
branch_name: str | None = None,
worktree_path: str | None = None,
idempotency_key: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
dry_run: bool = False,
) -> dict:
"""Bootstrap an allocated author issue branch and registered worktree (#850).
Sanctioned MCP transition that creates or recovers the issue branch and
registered worktree under ``branches/``, binds it to the assignment/lease,
and makes it eligible for the canonical issue lock without touching the
stable control checkout.
Args:
issue_number: Allocated issue number to bootstrap.
assignment_id: Optional allocation assignment ID.
lease_id: Optional workflow lease ID.
expected_base_sha: Authoritative expected base SHA / concurrency pin.
branch_name: Optional custom branch name (must match issue-<N> pattern).
worktree_path: Optional custom worktree path under branches/.
idempotency_key: Optional key for idempotent replay/resume.
remote: Known instance 'dadeschools' or 'prgs'.
host: Override Gitea host.
org: Override Org.
repo: Override Repo.
dry_run: Report planned transition without mutating repository.
"""
task = "bootstrap_author_issue_worktree"
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
task
)
if not ok:
return _author_mutation_block(block_reasons)
blocked = _namespace_mutation_block(task, remote=remote)
if blocked:
return blocked
blocked = _profile_permission_block(
task_capability_map.required_permission(task),
remote=remote,
host=host,
org=org,
repo=repo,
org_explicit=org is not None,
repo_explicit=repo is not None,
)
if blocked:
return blocked
verify_preflight_purity(
remote,
task=task,
org=org,
repo=repo,
)
h, o, r = _resolve(remote, host, org, repo)
canonical_root = _canonical_local_git_root()
import author_issue_bootstrap
return author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=issue_number,
canonical_repo_root=canonical_root,
assignment_id=assignment_id,
lease_id=lease_id,
expected_base_sha=expected_base_sha,
branch_name=branch_name,
worktree_path=worktree_path,
idempotency_key=idempotency_key,
remote=remote,
host=h,
org=o,
repo=r,
active_identity=_active_username(),
active_profile=_active_profile_name(),
owner_session=_current_session_id(),
dry_run=dry_run,
)
# Merge methods supported by the Gitea merge API. # Merge methods supported by the Gitea merge API.
_MERGE_METHODS = ("merge", "squash", "rebase") _MERGE_METHODS = ("merge", "squash", "rebase")
@@ -12014,163 +12309,127 @@ def gitea_reconcile_merged_cleanups(
if dry_run: if dry_run:
report["dry_run"] = True report["dry_run"] = True
report["executed"] = False report["executed"] = False
# #851: surface planned lifecycle order so dry-run matches execute.
report["planned_execution_orders"] = {
str(entry.get("pr_number")): entry.get("planned_execution_order") or []
for entry in (report.get("entries") or [])
}
return {"success": True, "performed": False, **report} return {"success": True, "performed": False, **report}
verify_preflight_purity( verify_preflight_purity(
remote, task="reconcile_merged_cleanups", org=org, repo=repo remote, task="reconcile_merged_cleanups", org=org, repo=repo
) )
actions: list[dict] = [] actions: list[dict] = []
project_root = _canonical_local_git_root()
def _ownership_records_for_branch(
head_branch: str, pr_num_int: int | None
) -> list[dict]:
ownership_bundle = _collect_branch_ownership_records(
remote=remote,
host=h,
org=o,
repo=r,
branch=head_branch,
pr_number=pr_num_int,
project_root=project_root,
auth=auth,
base_api=base,
)
ownership_records = list(ownership_bundle.get("records") or [])
if ownership_bundle.get("inventory_error"):
ownership_records.append(
{
"category": (
branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR
),
"status": "unknown",
"remote": remote,
"host": h,
"org": o,
"repo": r,
"branch": head_branch,
"reclaim_allowed": False,
"role": "inventory",
}
)
return ownership_records
def _attempt_owned_remote_delete(
*,
head_branch: str,
pr_num_int: int | None,
after_worktree_removal: bool = False,
) -> dict:
"""Fail-closed remote delete with live ownership reassessment (#851)."""
import urllib.parse
ownership_records = _ownership_records_for_branch(head_branch, pr_num_int)
ownership = branch_cleanup_guard.assess_active_branch_ownership(
remote=remote,
org=o,
repo=r,
branch=head_branch,
host=h,
records=ownership_records,
)
if ownership.get("block"):
return {
"action": "delete_remote_branch",
"branch": head_branch,
"success": False,
"performed": False,
"delete_acknowledged": False,
"verified_absent": False,
"blocker_kind": "active_branch_ownership",
"reasons": ownership.get("reasons") or [],
"blocking_categories": ownership.get("blocking_categories") or [],
"after_worktree_removal": after_worktree_removal,
"ownership_reassessed": after_worktree_removal,
}
encoded = urllib.parse.quote(head_branch, safe="")
url = f"{base}/branches/{encoded}"
with _audited(
"delete_branch",
host=h,
remote=remote,
org=o,
repo=r,
target_branch=head_branch,
request_metadata={
"branch": head_branch,
"source": "reconcile_merged_cleanups",
"ownership_checked": True,
"after_worktree_removal": after_worktree_removal,
},
):
api_request("DELETE", url, auth)
readback = _probe_remote_branch(h, o, r, auth, head_branch)
readback_assessment = branch_cleanup_guard.assess_post_delete_readback(
readback
)
verified = bool(readback_assessment.get("verified_absent"))
return {
"action": "delete_remote_branch",
"branch": head_branch,
"success": bool(readback_assessment.get("ok")),
"performed": True,
"delete_acknowledged": True,
"verified_absent": verified,
"readback": readback_assessment.get("readback"),
"reasons": readback_assessment.get("reasons") or [],
"after_worktree_removal": after_worktree_removal,
"ownership_reassessed": after_worktree_removal,
}
for entry in report.get("entries") or []: for entry in report.get("entries") or []:
head_branch = entry.get("head_branch") or "" head_branch = entry.get("head_branch") or ""
remote_assessment = entry.get("remote_branch") or {} remote_assessment = entry.get("remote_branch") or {}
local_assessment = entry.get("local_worktree") or {} local_assessment = entry.get("local_worktree") or {}
pr_num = entry.get("pr_number")
try:
pr_num_int = int(pr_num) if pr_num is not None else None
except (TypeError, ValueError):
pr_num_int = None
# #851 lifecycle: when the target worktree is independently safe, remove if remote_assessment.get("safe_to_delete_remote"):
# it first so worktree_binding ownership does not permanently strand import urllib.parse
# both the worktree and the remote branch. Never skip worktree removal
# merely because remote delete would be blocked by that binding. pr_num = entry.get("pr_number")
# Ownership protection for remote delete remains fail-closed below. try:
worktree_removed = False pr_num_int = int(pr_num) if pr_num is not None else None
except (TypeError, ValueError):
pr_num_int = None
ownership_bundle = _collect_branch_ownership_records(
remote=remote,
host=h,
org=o,
repo=r,
branch=head_branch,
pr_number=pr_num_int,
project_root=_canonical_local_git_root(),
auth=auth,
base_api=base,
)
ownership_records = list(ownership_bundle.get("records") or [])
if ownership_bundle.get("inventory_error"):
ownership_records.append(
{
"category": (
branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR
),
"status": "unknown",
"remote": remote,
"host": h,
"org": o,
"repo": r,
"branch": head_branch,
"reclaim_allowed": False,
"role": "inventory",
}
)
ownership = branch_cleanup_guard.assess_active_branch_ownership(
remote=remote,
org=o,
repo=r,
branch=head_branch,
host=h,
records=ownership_records,
)
if ownership.get("block"):
actions.append(
{
"action": "delete_remote_branch",
"branch": head_branch,
"success": False,
"performed": False,
"delete_acknowledged": False,
"verified_absent": False,
"blocker_kind": "active_branch_ownership",
"reasons": ownership.get("reasons") or [],
"blocking_categories": ownership.get(
"blocking_categories"
)
or [],
}
)
continue
encoded = urllib.parse.quote(head_branch, safe="")
url = f"{base}/branches/{encoded}"
with _audited(
"delete_branch",
host=h,
remote=remote,
org=o,
repo=r,
target_branch=head_branch,
request_metadata={
"branch": head_branch,
"source": "reconcile_merged_cleanups",
"ownership_checked": True,
},
):
api_request("DELETE", url, auth)
readback = _probe_remote_branch(h, o, r, auth, head_branch)
readback_assessment = branch_cleanup_guard.assess_post_delete_readback(
readback
)
verified = bool(readback_assessment.get("verified_absent"))
actions.append(
{
"action": "delete_remote_branch",
"branch": head_branch,
"success": bool(readback_assessment.get("ok")),
"performed": True,
"delete_acknowledged": True,
"verified_absent": verified,
"readback": readback_assessment.get("readback"),
"reasons": readback_assessment.get("reasons") or [],
}
)
if local_assessment.get("safe_to_remove_worktree"): if local_assessment.get("safe_to_remove_worktree"):
result = merged_cleanup_reconcile.remove_local_worktree( result = merged_cleanup_reconcile.remove_local_worktree(
project_root, _canonical_local_git_root(),
head_branch, head_branch,
worktree_path=local_assessment.get("worktree_path"), worktree_path=local_assessment.get("worktree_path"),
) )
actions.append({"action": "remove_local_worktree", **result}) actions.append({"action": "remove_local_worktree", **result})
# Idempotent resume: absent worktree is already gone.
msg = (result.get("message") or "").lower()
worktree_removed = bool(result.get("success")) or (
"not found" in msg
)
if remote_assessment.get("safe_to_delete_remote"):
actions.append(
_attempt_owned_remote_delete(
head_branch=head_branch,
pr_num_int=pr_num_int,
after_worktree_removal=worktree_removed,
)
)
for scratch in report.get("reviewer_scratch_entries") or []: for scratch in report.get("reviewer_scratch_entries") or []:
if not scratch.get("safe_to_remove_worktree"): if not scratch.get("safe_to_remove_worktree"):
continue continue
result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree( result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree(
project_root, scratch.get("worktree_path") or "" _canonical_local_git_root(), scratch.get("worktree_path") or ""
) )
actions.append({"action": "remove_reviewer_scratch_worktree", **result}) actions.append({"action": "remove_reviewer_scratch_worktree", **result})
@@ -17763,6 +18022,17 @@ def gitea_assess_master_parity(
} }
if parity["restart_required"] and enforced: if parity["restart_required"] and enforced:
out["report"] = master_parity_gate.parity_report(parity) out["report"] = master_parity_gate.parity_report(parity)
# #662 AC1: first post-restart master-parity probe also runs the boot
# reconcile once (log-only by default; never raises).
boot_proof = _ensure_boot_post_restart_reconcile()
if boot_proof is not None:
out["post_restart_reconcile"] = {
"reconcile_id": boot_proof.get("reconcile_id"),
"overall_status": boot_proof.get("overall_status"),
"mutation_hold": boot_proof.get("mutation_hold"),
"unresolved_count": boot_proof.get("unresolved_count"),
"mode": boot_proof.get("mode"),
}
return out return out
@@ -20150,6 +20420,12 @@ def gitea_resolve_task_capability(
task: str, task: str,
remote: str = "dadeschools", remote: str = "dadeschools",
host: str | None = None, host: str | None = None,
org: str | None = None,
repo: str | None = None,
issue_number: int | None = None,
worktree_path: str | None = None,
pr_number: int | None = None,
**kwargs: Any,
) -> dict: ) -> dict:
"""Read-only / side-effect free: resolve capability, profile, and namespace for a task. """Read-only / side-effect free: resolve capability, profile, and namespace for a task.
@@ -20166,13 +20442,14 @@ def gitea_resolve_task_capability(
remote: Known remote instance name. remote: Known remote instance name.
host: Optional override for the Gitea host. host: Optional override for the Gitea host.
""" """
task_key = task_capability_map._canonical_preflight_task(task)
TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP
# Every fresh attempt invalidates the previous task/role stamp before any # Every fresh attempt invalidates the previous task/role stamp before any
# fallible resolver work. Unknown/malformed tasks and unexpected failures # fallible resolver work. Unknown/malformed tasks and unexpected failures
# therefore remain fail-closed instead of preserving stale authority. # therefore remain fail-closed instead of preserving stale authority.
_clear_resolved_capability_stamp() _clear_resolved_capability_stamp()
if task not in TASK_MAP: if task_key not in TASK_MAP:
# #723: structured fail-closed unknown_task (never raise into internal_error). # #723: structured fail-closed unknown_task (never raise into internal_error).
profile = get_profile() profile = get_profile()
h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
@@ -20218,8 +20495,8 @@ def gitea_resolve_task_capability(
result["cleared_stale_denial"] = True result["cleared_stale_denial"] = True
return result return result
required_permission = task_capability_map.required_permission(task) required_permission = task_capability_map.required_permission(task_key)
required_role = task_capability_map.required_role(task) required_role = task_capability_map.required_role(task_key)
role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS
infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT) infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
@@ -20405,7 +20682,10 @@ def gitea_resolve_task_capability(
available_in_session = allowed_in_current_session available_in_session = allowed_in_current_session
runtime_stale_blocker = False runtime_stale_blocker = False
if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ: if (
"PYTEST_CURRENT_TEST" not in os.environ
or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ
) and os.environ.get("GITEA_ALLOW_STALE_RUNTIME") != "1":
runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles) runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
if runtime_reasons: if runtime_reasons:
restart_required = True restart_required = True
@@ -22201,6 +22481,249 @@ def gitea_request_mcp_restart(
return payload return payload
# --- #662 post-restart reconciliation ---------------------------------------
_POST_RESTART_LAST_PROOF: dict | None = None
_POST_RESTART_BOOT_RAN = False
def _post_restart_reconcile_mode() -> str:
"""Return log_only (default) or enforce from process environment."""
raw = (os.environ.get("GITEA_POST_RESTART_RECONCILE_MODE") or "").strip().lower()
if raw in {"enforce", "enforced", "hold"}:
return "enforce"
return "log_only"
def _gather_post_restart_inventory(
*,
remote: str,
org: str,
repo: str,
limit: int = 200,
) -> dict:
"""Gather live control-plane facts for post-restart reconcile (#662).
Best-effort and fail-closed: any inventory section that cannot be read is
recorded on ``incomplete_reasons`` and ``inventory_complete`` is cleared.
Never mutates Gitea or the control-plane DB.
"""
import master_parity_gate
import post_restart_reconcile as prr
inventory_complete = True
incomplete_reasons: list[str] = []
sessions: list[dict] = []
leases: list[dict] = []
worktree_bindings: list[dict] = []
db, db_errs = _control_plane_db_or_error()
if db is None:
inventory_complete = False
incomplete_reasons.extend(
db_errs or ["control-plane DB unavailable; cannot reconcile after restart"]
)
else:
try:
sessions = db.list_sessions(statuses=("active",), limit=max(1, int(limit)))
except Exception as exc: # noqa: BLE001
inventory_complete = False
incomplete_reasons.append(
f"session inventory failed: {_redact(str(exc))}"
)
try:
lease_result = lease_lifecycle.list_active_leases(
db,
remote=remote if remote in REMOTES else remote,
org=org,
repo=repo,
role=None,
include_non_active=True,
limit=max(1, int(limit)),
)
leases = list(lease_result.get("leases") or [])
except Exception as exc: # noqa: BLE001
inventory_complete = False
incomplete_reasons.append(
f"lease inventory failed: {_redact(str(exc))}"
)
leases = []
# Derive worktree bindings from live leases that carry a path.
for row in leases:
wt = (row.get("worktree_path") or "").strip()
if not wt:
continue
exists = bool(lease_lifecycle.worktree_exists(wt))
worktree_bindings.append(
{
"path": wt,
"exists": exists,
"missing": not exists,
"lease_id": row.get("lease_id"),
"session_id": row.get("session_id"),
"work_kind": row.get("work_kind"),
"work_number": row.get("work_number"),
}
)
# Capability / master-parity dimension (code parity after restart).
try:
parity = _current_master_parity()
capabilities = {
"stale": bool(parity.get("stale")),
"startup_head": parity.get("startup_head"),
"current_head": parity.get("current_head"),
"in_parity": parity.get("in_parity"),
"mutation_safe": parity.get("mutation_safe"),
}
service_health = {
"healthy": bool(parity.get("mutation_safe", not parity.get("stale"))),
"parity_summary": master_parity_gate.format_parity(parity),
}
boot_head = parity.get("startup_head") or _process_boot_head_sha
current_head = parity.get("current_head")
except Exception as exc: # noqa: BLE001
inventory_complete = False
incomplete_reasons.append(
f"master-parity inventory failed: {_redact(str(exc))}"
)
capabilities = {"stale": None}
service_health = {"healthy": None, "error": "parity assessment failed"}
boot_head = _process_boot_head_sha
current_head = None
# #660 soft dependency: checkpoint schema not landed → skip dimension.
checkpoints_available = False
try:
import importlib
importlib.import_module("session_checkpoint_schema")
checkpoints_available = True
except Exception: # noqa: BLE001
checkpoints_available = False
return {
"inventory_complete": inventory_complete,
"incomplete_reasons": incomplete_reasons,
"service_health": service_health,
"clients": [], # client transport inventory is host/IDE-owned
"sessions": sessions,
"leases": leases,
"checkpoints_available": checkpoints_available,
"checkpoints": [] if checkpoints_available else None,
"worktree_bindings": worktree_bindings,
"pending_mutations": [],
"capabilities": capabilities,
"boot_head_sha": boot_head,
"current_head_sha": current_head,
"queue_state": {"safe_to_resume": inventory_complete},
"reconcile_version": prr.RECONCILE_VERSION,
}
def _run_post_restart_reconcile(
*,
remote: str = "prgs",
org: str | None = None,
repo: str | None = None,
mode: str | None = None,
limit: int = 200,
) -> dict:
"""Gather + classify post-restart state; cache the latest proof (#662)."""
global _POST_RESTART_LAST_PROOF, _POST_RESTART_BOOT_RAN
import post_restart_reconcile as prr
try:
_h, o, r = _resolve(remote, None, org, repo)
except ValueError as exc:
return {
"success": False,
"read_only": True,
"reasons": [str(exc)],
}
inventory = _gather_post_restart_inventory(
remote=remote, org=o, repo=r, limit=limit
)
proof = prr.reconcile_after_restart(
inventory,
mode=mode or _post_restart_reconcile_mode(),
)
payload = proof.as_dict()
payload["success"] = True
payload["read_only"] = True
payload["remote"] = remote
payload["org"] = o
payload["repo"] = r
payload["follow_up_create_supported"] = False
payload["follow_up_create_note"] = (
"proposed_follow_ups lists durable issues the apply path may create; "
"this tool never creates them (log-only by default, #662 rollout)"
)
_POST_RESTART_LAST_PROOF = payload
_POST_RESTART_BOOT_RAN = True
return payload
def _ensure_boot_post_restart_reconcile() -> dict | None:
"""Run post-restart reconcile once per process (boot hook, #662 AC1)."""
global _POST_RESTART_BOOT_RAN
if _POST_RESTART_BOOT_RAN:
return _POST_RESTART_LAST_PROOF
# Best-effort: never raise from the boot hook.
try:
return _run_post_restart_reconcile()
except Exception: # noqa: BLE001
_POST_RESTART_BOOT_RAN = True
return None
@mcp.tool()
def gitea_reconcile_after_restart(
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
mode: str | None = None,
limit: int = 200,
) -> dict:
"""Run post-restart MCP reconciliation and return a completion proof (#662).
Gathers live control-plane sessions, leases, worktree bindings, and
master-parity evidence, then classifies them with the pure
``post_restart_reconcile.reconcile_after_restart`` assessor. Returns a
machine-readable completion proof listing resolved / unresolved dimensions
and proposed durable follow-up issues.
Read-only by design: never restarts MCP, never auto-resumes write
mutations, and never creates Gitea issues (those are a separate apply
path). Default mode is ``log_only``; set
``GITEA_POST_RESTART_RECONCILE_MODE=enforce`` (or pass ``mode='enforce'``)
to set ``mutation_hold`` when anything remains unresolved.
Soft-depends on #660 for session checkpoints: when the checkpoint schema
module is absent the checkpoints dimension is ``skipped`` with an explicit
reason rather than inventing a schema.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"read_only": True,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
return _run_post_restart_reconcile(
remote=remote,
org=org,
repo=repo,
mode=mode,
limit=limit,
)
@mcp.tool() @mcp.tool()
def gitea_inspect_workflow_lease( def gitea_inspect_workflow_lease(
lease_id: str, lease_id: str,
+579 -54
View File
@@ -15,15 +15,27 @@ import json
import os import os
import re import re
import tempfile import tempfile
import uuid
from contextlib import contextmanager from contextlib import contextmanager
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from typing import Any from typing import Any
import lease_policy
LOCK_DIR_ENV = "GITEA_ISSUE_LOCK_DIR" LOCK_DIR_ENV = "GITEA_ISSUE_LOCK_DIR"
DEFAULT_LOCK_DIR = os.path.expanduser("~/.cache/gitea-tools/issue-locks") DEFAULT_LOCK_DIR = os.path.expanduser("~/.cache/gitea-tools/issue-locks")
WORK_LEASE_TTL_HOURS = 4
AUTHOR_ISSUE_WORK_LEASE = "author_issue_work" AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
# Freshness classifications. ``STATUS_STALE`` remains the dead-PID band that
# #753 recovery keys on; the two bands below are new in #790 Slice A and apply
# only to leases minted under the heartbeat lifecycle.
STATUS_LIVE = "live"
STATUS_EXPIRED = "expired"
STATUS_ABSENT = "absent"
STATUS_STALE = "stale"
STATUS_STALE_MISSED_HEARTBEAT = "stale_missed_heartbeat"
STATUS_STALE_ABSOLUTE_CAP = "stale_absolute_cap"
_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+") _SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
@@ -257,6 +269,331 @@ def bind_session_lock(
return path return path
def _ownership_refusals(
lock: dict[str, Any],
*,
issue_number: int,
branch_name: str,
worktree_path: str,
identity: str | None,
profile: str | None,
) -> list[str]:
"""Exact-ownership mismatches between a durable lock and a live caller.
Shared by the heartbeat writer and the legacy rebind path so the two cannot
disagree about what "the same owner" means. Every field is compared against
durable state; nothing is taken on the caller's word beyond the identity the
server itself resolved.
"""
reasons: list[str] = []
if lock.get("issue_number") != issue_number:
reasons.append(
f"lock targets issue #{lock.get('issue_number')}, not #{issue_number}"
)
if str(lock.get("branch_name") or "") != str(branch_name or ""):
reasons.append(
f"lock branch '{lock.get('branch_name')}' does not match '{branch_name}'"
)
if not _same_realpath(str(lock.get("worktree_path") or ""), worktree_path):
reasons.append(
f"lock worktree '{lock.get('worktree_path')}' does not match "
f"'{worktree_path}'"
)
lease = lock.get("work_lease") if isinstance(lock, dict) else None
claimant = lease.get("claimant") if isinstance(lease, dict) else None
claimant = claimant if isinstance(claimant, dict) else {}
recorded_identity = str(claimant.get("username") or "").strip()
recorded_profile = str(claimant.get("profile") or "").strip()
if not recorded_identity or not recorded_profile:
reasons.append("lock does not record both a claimant username and profile")
if recorded_identity and recorded_identity != str(identity or "").strip():
reasons.append(
f"lock claimant '{recorded_identity}' does not match active identity "
f"'{str(identity or '').strip() or 'unknown'}'"
)
if recorded_profile and recorded_profile != str(profile or "").strip():
reasons.append(
f"lock profile '{recorded_profile}' does not match active profile "
f"'{str(profile or '').strip() or 'unknown'}'"
)
return reasons
def _refusal(reasons: list[str], **extra: Any) -> dict[str, Any]:
return {"success": False, "performed": False, "reasons": reasons, **extra}
def heartbeat_session_lock(
*,
remote: str,
org: str,
repo: str,
issue_number: int,
branch_name: str,
worktree_path: str,
identity: str | None,
profile: str | None,
task_session_id: str,
expected_generation: int | None = None,
lock_dir: str | None = None,
now: datetime | None = None,
) -> dict[str, Any]:
"""Slide a heartbeat-lifecycle lease forward (#790 Slice A, A4).
The write happens inside the same per-issue ``flock`` that serializes
acquisition, and under the #772 generation compare-and-swap, so a heartbeat
can never race a concurrent reclaim: whichever lands first moves the
generation and the other fails closed.
Refuses — never revives — in every ambiguous case. A lease that has already
lapsed past its grace is *not* heartbeatable: allowing that would let a
session that stopped proving liveness restore ownership retroactively, which
is precisely the revival AC-N5 forbids. Such a session must go through the
sanctioned reclaim path, which mints a fresh generation.
"""
current = _lease_now(now)
root = _ensure_lock_dir(lock_dir)
path = lock_file_path(
remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=root
)
declared_session = str(task_session_id or "").strip()
if not declared_session:
return _refusal(["no task_session_id supplied (fail closed)"])
sentinel = flock_path(path)
try:
with _exclusive_file_lock(sentinel):
lock = read_lock_file(path)
if not lock:
return _refusal([f"no durable lock for issue #{issue_number}"])
if is_legacy_lease(lock):
return _refusal(
[
"lock predates the heartbeat lifecycle; it must be rebound "
"by its exact owner before it can be heartbeated"
],
lifecycle=lease_lifecycle_version(lock),
legacy_lease=True,
)
reasons = _ownership_refusals(
lock,
issue_number=issue_number,
branch_name=branch_name,
worktree_path=worktree_path,
identity=identity,
profile=profile,
)
recorded_session = lease_task_session_id(lock)
if not recorded_session:
reasons.append(
"lock declares the heartbeat lifecycle but records no "
"task_session_id (fail closed)"
)
elif recorded_session != declared_session:
# A superseded session holding an old identifier cannot heartbeat
# over the session that replaced it.
reasons.append(
"task_session_id does not match the session recorded on the lock"
)
if reasons:
return _refusal(reasons)
current_generation = lock_generation(lock)
if (
expected_generation is not None
and current_generation != expected_generation
):
return _refusal(
[
f"lock generation changed: expected {expected_generation}, "
f"found {current_generation}; another session reclaimed or "
"replaced this claim (fail closed)"
],
lock_generation=current_generation,
)
freshness = assess_lock_freshness(lock, now=current)
if not freshness.get("live"):
return _refusal(
[
f"lease is not live ({freshness.get('status')}): "
f"{freshness.get('reason')}; a lapsed lease must be "
"reclaimed, not heartbeated"
],
freshness=freshness,
)
policy = lease_policy.policy_for(lease_task_class(lock))
expires = current + timedelta(minutes=policy.initial_ttl_minutes)
record = dict(lock)
lease = dict(record.get("work_lease") or {})
prior_heartbeat = lease.get("last_heartbeat_at")
lease["last_heartbeat_at"] = _format_lease_timestamp(current)
lease["expires_at"] = _format_lease_timestamp(expires)
try:
lease["heartbeat_count"] = int(lease.get("heartbeat_count") or 0) + 1
except (TypeError, ValueError):
lease["heartbeat_count"] = 1
record["work_lease"] = lease
record["lock_generation"] = current_generation + 1
save_lock_file(path, record)
except LockContentionError as exc:
return _refusal([f"issue #{issue_number} lock contention: {exc} (fail closed)"])
return {
"success": True,
"performed": True,
"issue_number": issue_number,
"branch_name": branch_name,
"worktree_path": worktree_path,
"task_session_id": declared_session,
"lock_generation": record["lock_generation"],
"prior_generation": current_generation,
"prior_heartbeat_at": prior_heartbeat,
"last_heartbeat_at": lease["last_heartbeat_at"],
"expires_at": lease["expires_at"],
"heartbeat_count": lease["heartbeat_count"],
"lock_file_path": path,
"policy": lease_policy.describe(lease_task_class(record)),
"freshness": assess_lock_freshness(record, now=current),
}
def rebind_legacy_lock(
*,
remote: str,
org: str,
repo: str,
issue_number: int,
branch_name: str,
worktree_path: str,
identity: str | None,
profile: str | None,
expected_generation: int | None = None,
lock_dir: str | None = None,
now: datetime | None = None,
) -> dict[str, Any]:
"""Move a legacy lock into the heartbeat lifecycle (#790 AC-N8).
One of the two sanctioned exits from the preserved-expiry legacy state; the
other is terminal retirement, which is Slice B. Only the exact recorded
owner may rebind, and only while the legacy lock is still live under its
original absolute expiry — an already-expired legacy lease belongs to the
#760 renewal path or #601 reclaim, and this must not become a second, weaker
way to revive one.
The rebind mints a genuine task-session identifier and a genuine first
heartbeat. It does not fabricate history: the original creation and expiry
are preserved under ``legacy_origin`` for audit, and the new lifecycle's
absolute cap runs from the rebind, not from the legacy claim.
"""
current = _lease_now(now)
root = _ensure_lock_dir(lock_dir)
path = lock_file_path(
remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=root
)
sentinel = flock_path(path)
try:
with _exclusive_file_lock(sentinel):
lock = read_lock_file(path)
if not lock:
return _refusal([f"no durable lock for issue #{issue_number}"])
if not is_legacy_lease(lock):
return _refusal(
[
"lock is already on the heartbeat lifecycle; use the "
"heartbeat path"
],
lifecycle=lease_lifecycle_version(lock),
legacy_lease=False,
)
reasons = _ownership_refusals(
lock,
issue_number=issue_number,
branch_name=branch_name,
worktree_path=worktree_path,
identity=identity,
profile=profile,
)
if reasons:
return _refusal(reasons)
current_generation = lock_generation(lock)
if (
expected_generation is not None
and current_generation != expected_generation
):
return _refusal(
[
f"lock generation changed: expected {expected_generation}, "
f"found {current_generation} (fail closed)"
],
lock_generation=current_generation,
)
freshness = assess_lock_freshness(lock, now=current)
if not freshness.get("live"):
return _refusal(
[
f"legacy lease is not live ({freshness.get('status')}): "
f"{freshness.get('reason')}; rebinding is not a recovery "
"path for a lapsed lease"
],
freshness=freshness,
)
policy = lease_policy.policy_for(lease_task_class(lock))
expires = current + timedelta(minutes=policy.initial_ttl_minutes)
session_id = mint_task_session_id(lease_task_class(lock))
record = dict(lock)
lease = dict(record.get("work_lease") or {})
legacy_origin = {
"created_at": lease.get("created_at"),
"expires_at": lease.get("expires_at"),
"last_heartbeat_at": lease.get("last_heartbeat_at"),
"lifecycle": lease_policy.LIFECYCLE_LEGACY,
}
lease["lifecycle_version"] = lease_policy.LIFECYCLE_HEARTBEAT_V1
lease["task_session_id"] = session_id
lease["created_at"] = _format_lease_timestamp(current)
lease["last_heartbeat_at"] = _format_lease_timestamp(current)
lease["expires_at"] = _format_lease_timestamp(expires)
lease["heartbeat_count"] = 1
record["work_lease"] = lease
record["legacy_rebind"] = {
"rebound_at": _format_lease_timestamp(current),
"task_session_id": session_id,
"prior_generation": current_generation,
"legacy_origin": legacy_origin,
"reason": (
"legacy lock rebound into the heartbeat lifecycle by its exact "
"recorded owner"
),
}
record["lock_generation"] = current_generation + 1
save_lock_file(path, record)
except LockContentionError as exc:
return _refusal([f"issue #{issue_number} lock contention: {exc} (fail closed)"])
return {
"success": True,
"performed": True,
"issue_number": issue_number,
"task_session_id": session_id,
"lock_generation": record["lock_generation"],
"prior_generation": current_generation,
"lifecycle": lease_policy.LIFECYCLE_HEARTBEAT_V1,
"legacy_rebind": record["legacy_rebind"],
"expires_at": lease["expires_at"],
"last_heartbeat_at": lease["last_heartbeat_at"],
"lock_file_path": path,
"freshness": assess_lock_freshness(record, now=current),
}
def read_session_issue_lock(lock_dir: str | None = None) -> dict[str, Any] | None: def read_session_issue_lock(lock_dir: str | None = None) -> dict[str, Any] | None:
root = (lock_dir or default_lock_dir()).strip() root = (lock_dir or default_lock_dir()).strip()
pointer = read_lock_file(session_pointer_path(root)) pointer = read_lock_file(session_pointer_path(root))
@@ -340,6 +677,16 @@ def _parse_lease_timestamp(value: str | None) -> datetime | None:
return None return None
def _format_lease_timestamp(value: datetime) -> str:
"""Serialize a lease timestamp in the durable ``...Z`` form already on disk."""
return (
value.astimezone(timezone.utc)
.replace(microsecond=0)
.isoformat()
.replace("+00:00", "Z")
)
def lease_expires_at(lock: dict[str, Any] | None) -> datetime | None: def lease_expires_at(lock: dict[str, Any] | None) -> datetime | None:
if not lock: if not lock:
return None return None
@@ -360,26 +707,113 @@ def is_lease_live(lock: dict[str, Any] | None, *, now: datetime | None = None) -
return assess_lock_freshness(lock, now=now)["live"] return assess_lock_freshness(lock, now=now)["live"]
def lease_task_class(lock_data: dict[str, Any] | None) -> str:
"""Policy task class for a durable lock; author work when unrecorded."""
lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
if isinstance(lease, dict):
recorded = str(lease.get("operation_type") or "").strip()
if recorded:
return recorded
return AUTHOR_ISSUE_WORK_LEASE
def lease_lifecycle_version(lock_data: dict[str, Any] | None) -> str:
"""Read the durable lifecycle marker (#790 AC-N8).
The marker is the *only* discriminator between a heartbeat-lifecycle lease
and a legacy one. Timestamps are deliberately not consulted: a lock minted
before this lifecycle existed has ``last_heartbeat_at == created_at``
forever, and reading that equality as "recently heartbeated" would treat
every never-heartbeated legacy lock as fresh — the precise inversion AC-N8
forbids. A newly minted heartbeat lease also has the two equal, so the
equality carries no information in either direction.
"""
lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
if isinstance(lease, dict):
recorded = str(lease.get("lifecycle_version") or "").strip()
if recorded:
return recorded
return lease_policy.LIFECYCLE_LEGACY
def is_legacy_lease(lock_data: dict[str, Any] | None) -> bool:
"""True when a lock predates the shared heartbeat lifecycle."""
return lease_lifecycle_version(lock_data) != lease_policy.LIFECYCLE_HEARTBEAT_V1
def lease_task_session_id(lock_data: dict[str, Any] | None) -> str:
"""Recorded per-task session identifier, or empty for a legacy lock."""
lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
if isinstance(lease, dict):
return str(lease.get("task_session_id") or "").strip()
return ""
def mint_task_session_id(task_class: str = AUTHOR_ISSUE_WORK_LEASE) -> str:
"""Mint an ownership key for one task (#790 AC-N1).
Deliberately contains no process identifier. The recorded PID belongs to the
long-lived MCP daemon, which outlives any individual task and is reused by
every task it serves, so PID digits cannot identify *which* task holds a
claim. The PID is still recorded alongside this value as evidence.
"""
prefix = _sanitize_segment(str(task_class or AUTHOR_ISSUE_WORK_LEASE))
return f"{prefix}-{uuid.uuid4().hex[:16]}"
def _lease_heartbeat_at(lock_data: dict[str, Any] | None) -> datetime | None:
lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
heartbeat_at = None
if isinstance(lock_data, dict):
heartbeat_at = _parse_lease_timestamp(lock_data.get("last_heartbeat_at"))
if heartbeat_at is None and isinstance(lease, dict):
heartbeat_at = _parse_lease_timestamp(lease.get("last_heartbeat_at"))
return heartbeat_at
def assess_lock_freshness( def assess_lock_freshness(
lock_data: dict[str, Any] | None, lock_data: dict[str, Any] | None,
*, *,
now: datetime | None = None, now: datetime | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Classify a lock as live, expired, stale, or absent.""" """Classify a lock as live, expired, stale, or absent.
#790 Slice A makes the heartbeat load-bearing. Before this change
``last_heartbeat_at`` was parsed and then never consulted: liveness was
decided entirely by the absolute ``expires_at`` and by PID liveness, and
since the recorded PID is the long-lived MCP daemon, an abandoned author
task stayed "live" for the full four-hour TTL.
Two rules govern the rewrite:
* **An alive PID never establishes freshness** (AC-N2). It proves the daemon
is up, nothing about the task. It is recorded as evidence and no branch
returns ``live`` because of it.
* **A dead PID still corroborates staleness.** The dead-PID band is
unchanged and still precedes every heartbeat evaluation, so #753
dead-session recovery keys on exactly the classification it always did.
Legacy leases (AC-N8) keep their recorded absolute expiry and are never
evaluated against the short heartbeat grace, so deploying this change cannot
make an existing claim instantly reclaimable.
"""
current = _lease_now(now) current = _lease_now(now)
if not lock_data: if not lock_data:
return { return {
"status": "absent", "status": STATUS_ABSENT,
"live": False, "live": False,
"stale": False, "stale": False,
"reason": "no lock record", "reason": "no lock record",
} }
expires_at = lease_expires_at(lock_data)
lease = lock_data.get("work_lease") lease = lock_data.get("work_lease")
heartbeat_at = _parse_lease_timestamp(lock_data.get("last_heartbeat_at")) expires_at = lease_expires_at(lock_data)
if heartbeat_at is None and isinstance(lease, dict): heartbeat_at = _lease_heartbeat_at(lock_data)
heartbeat_at = _parse_lease_timestamp(lease.get("last_heartbeat_at")) created_at = (
_parse_lease_timestamp(lease.get("created_at"))
if isinstance(lease, dict)
else None
)
pid = lock_data.get("session_pid") pid = lock_data.get("session_pid")
if pid is None: if pid is None:
@@ -397,56 +831,103 @@ def assess_lock_freshness(
pid_int = None pid_int = None
pid_alive = is_process_alive(pid_int) if pid_int is not None else False pid_alive = is_process_alive(pid_int) if pid_int is not None else False
if expires_at and expires_at <= current: lifecycle = lease_lifecycle_version(lock_data)
return { legacy = lifecycle != lease_policy.LIFECYCLE_HEARTBEAT_V1
"status": "expired", policy = lease_policy.policy_for(lease_task_class(lock_data))
"live": False,
"stale": True,
"reason": f"lease expired at {expires_at.isoformat()}",
"pid_alive": pid_alive,
"pid_missing": pid_missing,
}
# #860: a PID-less lock must never be considered live merely because evidence: dict[str, Any] = {
# expiration / heartbeat fields are absent. Missing PID is insufficient
# evidence of a live owner; treat as malformed/stale so recovery routes
# can evaluate corroborating pins instead of blocking on a false live flag.
if pid_missing:
return {
"status": "malformed",
"live": False,
"stale": True,
"reason": (
"lock has no usable session pid; cannot prove live ownership "
"(PID-less locks are never live by missing expiry alone)"
),
"pid_alive": False,
"pid_missing": True,
"heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None,
"expires_at": expires_at.isoformat() if expires_at else None,
}
if pid_int is not None and not pid_alive:
return {
"status": "stale",
"live": False,
"stale": True,
"reason": f"owner pid {pid_int} is not alive",
"pid_alive": False,
"pid_missing": False,
}
return {
"status": "live",
"live": True,
"stale": False,
"reason": "lock heartbeat and lease are fresh",
"pid_alive": pid_alive, "pid_alive": pid_alive,
"pid_missing": False, "pid_missing": pid_missing,
"lifecycle": lifecycle,
"legacy_lease": legacy,
"task_session_id": lease_task_session_id(lock_data) or None,
"heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None, "heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None,
"expires_at": expires_at.isoformat() if expires_at else None, "expires_at": expires_at.isoformat() if expires_at else None,
} }
def _result(status: str, *, live: bool, reason: str, **extra: Any) -> dict[str, Any]:
return {
"status": status,
"live": live,
"stale": not live and status != STATUS_ABSENT,
"reason": reason,
**evidence,
**extra,
}
if legacy:
# AC-N8: the preserved absolute expiry is the only clock for a lock
# written before task-session heartbeats existed.
if expires_at and expires_at <= current:
return _result(
STATUS_EXPIRED,
live=False,
reason=f"lease expired at {expires_at.isoformat()}",
)
if pid is not None and not pid_alive:
return _result(
STATUS_STALE, live=False, reason=f"owner pid {pid} is not alive"
)
return _result(
STATUS_LIVE,
live=True,
reason=(
"legacy lease is within its recorded absolute expiry; the "
"heartbeat grace does not apply retroactively"
),
legacy_expiry_preserved=True,
)
# ── Heartbeat lifecycle ──
if pid is not None and not pid_alive:
# Unchanged dead-PID band: #753 recovery depends on this exact status.
return _result(STATUS_STALE, live=False, reason=f"owner pid {pid} is not alive")
if heartbeat_at is None:
# Contradictory: a heartbeat lease must carry a heartbeat. Fail closed.
return _result(
STATUS_STALE_MISSED_HEARTBEAT,
live=False,
reason=(
f"lease declares lifecycle '{lifecycle}' but records no "
"last_heartbeat_at (fail closed)"
),
)
if policy.absolute_cap_hours and created_at is not None:
cap_at = created_at + timedelta(hours=policy.absolute_cap_hours)
if cap_at <= current:
return _result(
STATUS_STALE_ABSOLUTE_CAP,
live=False,
reason=(
f"lease exceeded its {policy.absolute_cap_hours}h absolute cap "
f"at {cap_at.isoformat()}; canonical re-adoption is required"
),
absolute_cap_at=cap_at.isoformat(),
)
grace_at = heartbeat_at + timedelta(minutes=policy.missed_heartbeat_grace_minutes)
if grace_at <= current or (expires_at is not None and expires_at <= current):
return _result(
STATUS_STALE_MISSED_HEARTBEAT,
live=False,
reason=(
f"no valid heartbeat since {heartbeat_at.isoformat()}; the "
f"{policy.missed_heartbeat_grace_minutes}min grace lapsed at "
f"{grace_at.isoformat()}"
),
missed_heartbeat_since=grace_at.isoformat(),
)
warning_at = heartbeat_at + timedelta(minutes=policy.stale_warning_minutes)
return _result(
STATUS_LIVE,
live=True,
reason="lease heartbeat is fresh within the configured grace",
heartbeat_warning=warning_at <= current,
)
def _same_realpath(left: str | None, right: str | None) -> bool: def _same_realpath(left: str | None, right: str | None) -> bool:
if not left or not right: if not left or not right:
@@ -483,6 +964,27 @@ def assess_expired_lock_reclaim(
"reasons": ["lock is still live; cannot reclaim (fail closed)"], "reasons": ["lock is still live; cannot reclaim (fail closed)"],
"freshness": freshness, "freshness": freshness,
} }
status = str(freshness.get("status") or "")
if status in (STATUS_STALE_MISSED_HEARTBEAT, STATUS_STALE_ABSOLUTE_CAP):
# #790: under the heartbeat lifecycle the heartbeat *is* the liveness
# proof, so a session that stopped heartbeating past its grace has
# released its claim by definition. Requiring a dead PID on top of that
# would reinstate the original defect — the recorded PID is the shared
# daemon, which stays alive across every abandoned task it ever served.
#
# This band is unreachable for a legacy lease (AC-N8), so no lock
# written before this lifecycle can be reclaimed by this path.
return {
"reclaim_allowed": True,
"reasons": [
f"heartbeat-lifecycle lease is {status}: {freshness.get('reason')}"
],
"freshness": freshness,
"prior_branch": existing_lock.get("branch_name"),
"prior_worktree": existing_lock.get("worktree_path"),
"prior_pid": existing_lock.get("session_pid") or existing_lock.get("pid"),
"prior_task_session_id": lease_task_session_id(existing_lock) or None,
}
pid = existing_lock.get("session_pid") pid = existing_lock.get("session_pid")
if pid is None: if pid is None:
pid = existing_lock.get("pid") pid = existing_lock.get("pid")
@@ -557,7 +1059,28 @@ def assess_same_issue_lease_conflict(
) )
if recovery_sanctioned and existing_issue == issue_number and existing_branch == branch_name: if recovery_sanctioned and existing_issue == issue_number and existing_branch == branch_name:
return None return None
if is_lease_expired(existing_lock, now=now): expired = is_lease_expired(existing_lock, now=now)
# #790 review #502/#516: a heartbeat-lifecycle lease that is non-live but
# whose absolute expires_at is still in the future must enter the same
# reclaim/renewal disposition as an expired lease. This happens for
# stale_absolute_cap (a session that keeps heartbeating past the 8h cap has
# expires_at = last_heartbeat + TTL in the future) and for
# stale_missed_heartbeat under an independent TTL>grace policy. Keying this
# gate on is_lease_expired alone left assess_expired_lock_reclaim — which
# already permits exactly those bands — unreachable from the acquisition
# path, so the load-bearing heartbeat was not load-bearing for foreign
# reclaim: the precise abandonment scenario #790 exists to fix.
# assess_foreign_lock_overwrite already keys on is_lease_live; this makes the
# same-issue path consistent with it. Legacy leases keep their absolute-expiry
# clock (AC-N8): is_lease_live already decides a legacy lease from expires_at
# / dead-PID alone and is_legacy_lease excludes it here, so this widening is a
# no-op for every pre-lifecycle lock.
heartbeat_non_live_reclaimable = (
not expired
and not is_legacy_lease(existing_lock)
and not is_lease_live(existing_lock, now=now)
)
if expired or heartbeat_non_live_reclaimable:
# #760 AC1/AC2: exact-owner renewal is a different disposition from # #760 AC1/AC2: exact-owner renewal is a different disposition from
# foreign takeover and is evaluated first. Before this, both branches # foreign takeover and is evaluated first. Before this, both branches
# below returned unconditionally, so the same_owner allowance further # below returned unconditionally, so the same_owner allowance further
@@ -571,10 +1094,12 @@ def assess_same_issue_lease_conflict(
reclaim = assess_expired_lock_reclaim(existing_lock, now=now) reclaim = assess_expired_lock_reclaim(existing_lock, now=now)
if reclaim.get("reclaim_allowed"): if reclaim.get("reclaim_allowed"):
# #601: expired + dead pid / missing worktree may be reclaimed # #601: expired + dead pid / missing worktree may be reclaimed
# through the normal lock path (sanctioned overwrite). # through the normal lock path (sanctioned overwrite). #790: the new
# stale heartbeat bands are reclaimable here on the same evidence.
return None return None
descriptor = "expired" if expired else "non-live"
return ( return (
f"Issue #{issue_number} has an expired {operation_type} lease on " f"Issue #{issue_number} has an {descriptor} {operation_type} lease on "
f"branch '{existing_branch}' from worktree '{existing_worktree}'. " f"branch '{existing_branch}' from worktree '{existing_worktree}'. "
"Recovery review is required before takeover (fail closed)" "Recovery review is required before takeover (fail closed)"
) )
+212
View File
@@ -0,0 +1,212 @@
"""Central lease policy configuration (#790 Slice A, AC-N7).
The single authoritative source for every lease duration in the project. Before
this module the numbers were scattered: a four-hour author TTL was declared
twice (``issue_lock_store`` and ``gitea_mcp_server``), the reviewer/merger
sliding window lived in ``reviewer_pr_lease``, the conflict-fix window in
``pr_work_lease``, and the control-plane default in ``control_plane_db``.
Nothing tied them together, so tuning one class silently diverged from the
others and no reader could answer "how long does a lease live?" without
grepping four files.
AC-N7 requires that this configuration exist *before* the first heartbeat and
TTL behavior that reads from it, so it ships in Slice A rather than trailing the
code it governs.
Deliberate boundaries:
* **Declaration is not rewiring.** Every task class is declared here, but only
those with ``heartbeat_lifecycle_active`` were migrated onto the shared
heartbeat lifecycle in Slice A — currently ``author_issue_work`` alone.
Reviewer, merger, and conflict-fix leases keep their own existing behavior
until Slice C moves them; their numbers are recorded here so the two cannot
drift apart unnoticed, and ``tests/test_issue_790_lease_policy.py`` asserts
the recorded values still equal the constants those modules use.
* **No policy decision lives here.** This module answers "how long", never "may
this session proceed". Freshness, reclaim, and renewal dispositions stay in
``issue_lock_store``.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Any
# Task classes. Only the first is migrated onto the shared lifecycle in Slice A.
TASK_CLASS_AUTHOR_ISSUE_WORK = "author_issue_work"
TASK_CLASS_REVIEWER_PR = "reviewer_pr"
TASK_CLASS_MERGER_PR = "merger_pr"
TASK_CLASS_CONFLICT_FIX = "conflict_fix"
# Durable marker for a lease minted under the shared heartbeat lifecycle.
#
# #790 AC-N8: this explicit marker — never a timestamp comparison — is what
# distinguishes a heartbeat-lifecycle lease from a legacy one. A lock written
# before this lifecycle existed carries no marker and reads as
# ``LIFECYCLE_LEGACY``.
LIFECYCLE_HEARTBEAT_V1 = "heartbeat-v1"
LIFECYCLE_LEGACY = "legacy"
_ENV_PREFIX = "GITEA_LEASE_POLICY"
@dataclass(frozen=True)
class LeasePolicy:
"""Durations governing one task class.
All intervals are minutes except ``absolute_cap_hours``. ``None`` for the
cap means the class has no maximum continuous duration.
"""
task_class: str
initial_ttl_minutes: float
heartbeat_cadence_minutes: float
stale_warning_minutes: float
missed_heartbeat_grace_minutes: float
absolute_cap_hours: float | None
recovery_grace_minutes: float
terminal_race_drain_minutes: float
terminal_retirement_eligible: bool
heartbeat_lifecycle_active: bool
# Defaults. ``author_issue_work`` adopts the reviewer window proven by #747
# rather than inventing new numbers: a lease expires 10 minutes after its last
# valid heartbeat, warns at half that, and an actively heartbeating session is
# never evicted. The prior value was a fixed four hours (240 minutes) that no
# heartbeat could shorten — the defect this issue exists to correct.
_DEFAULTS: dict[str, LeasePolicy] = {
TASK_CLASS_AUTHOR_ISSUE_WORK: LeasePolicy(
task_class=TASK_CLASS_AUTHOR_ISSUE_WORK,
initial_ttl_minutes=10.0,
heartbeat_cadence_minutes=2.0,
stale_warning_minutes=5.0,
missed_heartbeat_grace_minutes=10.0,
absolute_cap_hours=8.0,
recovery_grace_minutes=10.0,
terminal_race_drain_minutes=2.0,
terminal_retirement_eligible=True,
heartbeat_lifecycle_active=True,
),
# Declared, not rewired. These mirror reviewer_pr_lease.LEASE_TTL_MINUTES
# and STALE_WARNING_MINUTES; Slice C migrates the call sites.
TASK_CLASS_REVIEWER_PR: LeasePolicy(
task_class=TASK_CLASS_REVIEWER_PR,
initial_ttl_minutes=10.0,
heartbeat_cadence_minutes=2.0,
stale_warning_minutes=5.0,
missed_heartbeat_grace_minutes=10.0,
absolute_cap_hours=None,
recovery_grace_minutes=10.0,
terminal_race_drain_minutes=2.0,
terminal_retirement_eligible=False,
heartbeat_lifecycle_active=False,
),
TASK_CLASS_MERGER_PR: LeasePolicy(
task_class=TASK_CLASS_MERGER_PR,
initial_ttl_minutes=10.0,
heartbeat_cadence_minutes=2.0,
stale_warning_minutes=5.0,
missed_heartbeat_grace_minutes=10.0,
absolute_cap_hours=None,
recovery_grace_minutes=10.0,
terminal_race_drain_minutes=2.0,
terminal_retirement_eligible=False,
heartbeat_lifecycle_active=False,
),
# Mirrors pr_work_lease.DEFAULT_CONFLICT_FIX_TTL_MINUTES. Deliberately left
# at its current window; shortening it is Slice C's call, not this slice's.
TASK_CLASS_CONFLICT_FIX: LeasePolicy(
task_class=TASK_CLASS_CONFLICT_FIX,
initial_ttl_minutes=120.0,
heartbeat_cadence_minutes=2.0,
stale_warning_minutes=5.0,
missed_heartbeat_grace_minutes=10.0,
absolute_cap_hours=None,
recovery_grace_minutes=10.0,
terminal_race_drain_minutes=2.0,
terminal_retirement_eligible=False,
heartbeat_lifecycle_active=False,
),
}
_NUMERIC_FIELDS = (
"initial_ttl_minutes",
"heartbeat_cadence_minutes",
"stale_warning_minutes",
"missed_heartbeat_grace_minutes",
"absolute_cap_hours",
"recovery_grace_minutes",
"terminal_race_drain_minutes",
)
def env_var_name(task_class: str, field: str) -> str:
"""Environment variable that overrides one field of one task class."""
return f"{_ENV_PREFIX}_{task_class.upper()}_{field.upper()}"
def _override(task_class: str, field: str, default: float | None) -> float | None:
"""Read one override, falling back to *default* on anything unusable.
A malformed or non-positive override is ignored rather than raised: a typo
in an environment variable must not be able to mint a zero-length lease that
makes every claim instantly reclaimable, nor crash the server at import.
"""
raw = (os.environ.get(env_var_name(task_class, field)) or "").strip()
if not raw:
return default
try:
value = float(raw)
except (TypeError, ValueError):
return default
if value <= 0:
return default
return value
def policy_for(task_class: str) -> LeasePolicy:
"""Return the effective policy for *task_class*.
Unknown task classes fall back to the author policy, which is the most
conservative migrated class, rather than raising — a new caller must never
be able to crash a lock write by naming a class this table has not learned.
"""
key = str(task_class or "").strip() or TASK_CLASS_AUTHOR_ISSUE_WORK
base = _DEFAULTS.get(key) or _DEFAULTS[TASK_CLASS_AUTHOR_ISSUE_WORK]
resolved = {
field: _override(base.task_class, field, getattr(base, field))
for field in _NUMERIC_FIELDS
}
if all(resolved[field] == getattr(base, field) for field in _NUMERIC_FIELDS):
return base
return LeasePolicy(
task_class=base.task_class,
terminal_retirement_eligible=base.terminal_retirement_eligible,
heartbeat_lifecycle_active=base.heartbeat_lifecycle_active,
**resolved,
)
def known_task_classes() -> tuple[str, ...]:
"""Every declared task class, migrated or not."""
return tuple(_DEFAULTS)
def describe(task_class: str) -> dict[str, Any]:
"""Serializable view of a policy, for audit records and tool payloads."""
policy = policy_for(task_class)
return {
"task_class": policy.task_class,
"initial_ttl_minutes": policy.initial_ttl_minutes,
"heartbeat_cadence_minutes": policy.heartbeat_cadence_minutes,
"stale_warning_minutes": policy.stale_warning_minutes,
"missed_heartbeat_grace_minutes": policy.missed_heartbeat_grace_minutes,
"absolute_cap_hours": policy.absolute_cap_hours,
"recovery_grace_minutes": policy.recovery_grace_minutes,
"terminal_race_drain_minutes": policy.terminal_race_drain_minutes,
"terminal_retirement_eligible": policy.terminal_retirement_eligible,
"heartbeat_lifecycle_active": policy.heartbeat_lifecycle_active,
"lifecycle_version": LIFECYCLE_HEARTBEAT_V1,
}
+791
View File
@@ -0,0 +1,791 @@
"""Post-restart MCP reconciliation and completion proof (#662).
After an MCP process restart, sessions, leases, capabilities, worktrees, and
interrupted mutations are not systematically reconciled; operators rebuild
context from chat. This module is the pure classification core of the
post-restart reconcile path.
Design rules (mirrors ``restart_coordinator`` / ``workflow_dashboard``):
* **Pure classification.** :func:`reconcile_after_restart` takes an already
gathered inventory and returns a structured *completion proof*. It never
touches the network, the filesystem, or a live process, so multi-session
fixtures can drive every branch in unit tests.
* **Fail closed.** Incomplete inventory never reports overall ``complete``.
Ambiguous interrupted mutations are ``unresolved`` (never silently resumed).
* **No blind write resume.** The proof never authorizes replaying a mutation;
it only classifies evidence and names follow-up work.
* **#660 soft dependency.** When durable session checkpoints are not present
in the inventory, the checkpoint dimension is ``skipped`` with an explicit
reason rather than inventing a schema (#660 lands separately).
* **Log-only then enforce.** Default mode is ``log_only``. ``enforce`` sets
``mutation_hold`` when anything remains unresolved so callers can block
write ops until reconcile is complete or degraded mode is documented.
The single sanctioned gather+classify entry point is the MCP tool
``gitea_reconcile_after_restart`` (read-only inventory gather + pure classify).
Creating durable follow-up Gitea issues from unresolved items is an explicit
apply step outside this pure module.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Mapping, Sequence
from uuid import uuid4
import lease_lifecycle
RECONCILE_VERSION = "1.0.0-issue-662"
SCHEMA_VERSION = 1
# Overall proof statuses.
STATUS_COMPLETE = "complete"
STATUS_DEGRADED = "degraded"
STATUS_FAILED = "failed"
# Per-dimension item statuses.
ITEM_RESOLVED = "resolved"
ITEM_UNRESOLVED = "unresolved"
ITEM_DEGRADED = "degraded"
ITEM_SKIPPED = "skipped"
# Modes.
MODE_LOG_ONLY = "log_only"
MODE_ENFORCE = "enforce"
# Lease / session phases that imply a write critical section was in flight.
MUTATING_PHASES = frozenset(
{
"implementing",
"publishing",
"merging",
"reviewing",
"committing",
"pushing",
"closing",
"mutating",
"critical_section",
}
)
# Dimensions the acceptance criteria require.
DIM_SERVICE_HEALTH = "service_health"
DIM_CLIENTS = "clients"
DIM_SESSIONS = "sessions"
DIM_CHECKPOINTS = "checkpoints"
DIM_LEASES = "leases"
DIM_CAPABILITIES = "capabilities"
DIM_WORKTREES = "worktrees"
DIM_MUTATIONS = "interrupted_mutations"
DIM_DUPLICATES = "duplicates"
DIM_QUEUE = "queue"
def _utc_now() -> datetime:
return datetime.now(timezone.utc)
def _ts(dt: datetime) -> str:
return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
@dataclass(frozen=True)
class ReconcileItem:
"""One dimension of the post-restart reconcile report."""
dimension: str
status: str
summary: str
details: dict[str, Any] = field(default_factory=dict)
follow_up_required: bool = False
def as_dict(self) -> dict[str, Any]:
return {
"dimension": self.dimension,
"status": self.status,
"summary": self.summary,
"details": dict(self.details),
"follow_up_required": self.follow_up_required,
}
@dataclass(frozen=True)
class FollowUpIssue:
"""A durable follow-up issue the apply path may create for unresolved work."""
title: str
body: str
dimension: str
severity: str = "high"
def as_dict(self) -> dict[str, Any]:
return {
"title": self.title,
"body": self.body,
"dimension": self.dimension,
"severity": self.severity,
}
@dataclass(frozen=True)
class RestartCompletionProof:
"""Machine-readable post-restart completion proof (#662 AC2)."""
schema_version: int
reconcile_version: str
reconcile_id: str
started_at: str
finished_at: str
boot_head_sha: str | None
current_head_sha: str | None
inventory_complete: bool
incomplete_reasons: tuple[str, ...]
mode: str
mutation_hold: bool
overall_status: str
items: tuple[ReconcileItem, ...]
proposed_follow_ups: tuple[FollowUpIssue, ...]
resolved_count: int
unresolved_count: int
skipped_count: int
note: str
def as_dict(self) -> dict[str, Any]:
return {
"schema_version": self.schema_version,
"reconcile_version": self.reconcile_version,
"reconcile_id": self.reconcile_id,
"started_at": self.started_at,
"finished_at": self.finished_at,
"boot_head_sha": self.boot_head_sha,
"current_head_sha": self.current_head_sha,
"inventory_complete": self.inventory_complete,
"incomplete_reasons": list(self.incomplete_reasons),
"mode": self.mode,
"mutation_hold": self.mutation_hold,
"overall_status": self.overall_status,
"items": [i.as_dict() for i in self.items],
"proposed_follow_ups": [f.as_dict() for f in self.proposed_follow_ups],
"resolved_count": self.resolved_count,
"unresolved_count": self.unresolved_count,
"skipped_count": self.skipped_count,
"note": self.note,
"links": {
"umbrella": 655,
"vision": 652,
"roadmap": 653,
"issue": 662,
"checkpoint_schema": 660,
"drain_proof": 661,
},
}
def _item(
dimension: str,
status: str,
summary: str,
*,
details: dict[str, Any] | None = None,
follow_up: bool = False,
) -> ReconcileItem:
return ReconcileItem(
dimension=dimension,
status=status,
summary=summary,
details=dict(details or {}),
follow_up_required=follow_up,
)
def _lease_freshness(lease: Mapping[str, Any]) -> str:
fr = lease.get("freshness")
if isinstance(fr, Mapping):
return str(fr.get("freshness") or fr.get("status") or "unknown")
if isinstance(fr, str):
return fr
# Fall back to pure classifier when raw lease rows are supplied.
try:
return str(lease_lifecycle.classify_lease_freshness(dict(lease)).get("freshness") or "unknown")
except Exception: # noqa: BLE001 - pure path must not raise on bad rows
return "unknown"
def _is_live_freshness(freshness: str) -> bool:
return freshness in {"active", "live", "fresh"}
def _is_mutating_phase(phase: str | None) -> bool:
p = (phase or "").strip().lower()
if not p:
return False
if p in MUTATING_PHASES:
return True
# Soft match for compound phases like "author_implementing".
return any(token in p for token in MUTATING_PHASES)
def _detect_interrupted_mutations(
leases: Sequence[Mapping[str, Any]],
pending_mutations: Sequence[Mapping[str, Any]],
) -> list[dict[str, Any]]:
"""Return interrupted-mutation evidence (never auto-resumes writes)."""
found: list[dict[str, Any]] = []
for raw in pending_mutations or ():
if not isinstance(raw, Mapping):
continue
found.append(
{
"source": "pending_mutation_inventory",
"status": "unresolved",
"phase": raw.get("phase"),
"session_id": raw.get("session_id"),
"work_kind": raw.get("work_kind") or raw.get("kind"),
"work_number": raw.get("work_number") or raw.get("number"),
"reason": raw.get("reason")
or "pending mutation recorded across process restart",
"resume_allowed": False,
}
)
for lease in leases or ():
if not isinstance(lease, Mapping):
continue
phase = lease.get("phase")
freshness = _lease_freshness(lease)
if not _is_mutating_phase(str(phase) if phase is not None else None):
continue
# A mutating phase whose owner is not live is interrupted.
if _is_live_freshness(freshness):
# Still live after restart is itself surprising — flag for review.
found.append(
{
"source": "lease_mutating_phase",
"status": "unresolved",
"phase": phase,
"freshness": freshness,
"lease_id": lease.get("lease_id"),
"session_id": lease.get("session_id"),
"work_kind": lease.get("work_kind"),
"work_number": lease.get("work_number"),
"worktree_path": lease.get("worktree_path"),
"reason": (
"mutating lease phase still classified live after restart; "
"do not auto-resume writes"
),
"resume_allowed": False,
}
)
else:
found.append(
{
"source": "lease_mutating_phase",
"status": "unresolved",
"phase": phase,
"freshness": freshness,
"lease_id": lease.get("lease_id"),
"session_id": lease.get("session_id"),
"work_kind": lease.get("work_kind"),
"work_number": lease.get("work_number"),
"worktree_path": lease.get("worktree_path"),
"reason": (
f"mutating lease phase '{phase}' with non-live freshness "
f"'{freshness}' — interrupted by restart"
),
"resume_allowed": False,
}
)
return found
def _detect_duplicate_work(
leases: Sequence[Mapping[str, Any]],
) -> list[dict[str, Any]]:
"""Surface duplicate live claims on the same work item."""
by_work: dict[tuple[Any, Any], list[Mapping[str, Any]]] = {}
for lease in leases or ():
if not isinstance(lease, Mapping):
continue
if not _is_live_freshness(_lease_freshness(lease)):
continue
key = (lease.get("work_kind"), lease.get("work_number"))
if key[0] is None or key[1] is None:
continue
by_work.setdefault(key, []).append(lease)
dups: list[dict[str, Any]] = []
for (kind, number), rows in sorted(by_work.items(), key=lambda kv: str(kv[0])):
if len(rows) < 2:
continue
dups.append(
{
"work_kind": kind,
"work_number": number,
"claim_count": len(rows),
"session_ids": [r.get("session_id") for r in rows],
"lease_ids": [r.get("lease_id") for r in rows],
}
)
return dups
def _follow_up_for_item(item: ReconcileItem) -> FollowUpIssue | None:
if not item.follow_up_required:
return None
title = f"[post-restart] unresolved {item.dimension} after MCP restart"
body = (
f"## Post-restart reconcile follow-up (#662)\n\n"
f"**Dimension:** `{item.dimension}`\n"
f"**Status:** `{item.status}`\n"
f"**Summary:** {item.summary}\n\n"
f"```json\n{item.details!r}\n```\n\n"
f"Parent umbrella: #655 · Vision: #652 · Roadmap: #653 · Reconcile: #662\n"
f"Do **not** auto-resume write mutations; reconcile evidence first.\n"
)
return FollowUpIssue(
title=title,
body=body,
dimension=item.dimension,
severity="high" if item.dimension == DIM_MUTATIONS else "medium",
)
def reconcile_after_restart(
inventory: Mapping[str, Any],
*,
now: datetime | None = None,
mode: str = MODE_LOG_ONLY,
reconcile_id: str | None = None,
) -> RestartCompletionProof:
"""Classify a post-restart inventory into a completion proof (#662).
Parameters
----------
inventory:
Gathered facts. Expected keys (all optional except completeness):
* ``inventory_complete`` (bool) — fail closed when false
* ``incomplete_reasons`` (list[str])
* ``service_health`` (dict with ``healthy`` bool)
* ``clients`` (list) — connected client descriptors
* ``sessions`` (list)
* ``leases`` (list, optionally with ``freshness``)
* ``checkpoints`` (list | None) — durable session checkpoints (#660)
* ``checkpoints_available`` (bool) — False when #660 schema absent
* ``worktree_bindings`` (list)
* ``pending_mutations`` (list) — explicit interrupted-mutation evidence
* ``capabilities`` (dict with optional ``stale`` / heads)
* ``boot_head_sha`` / ``current_head_sha``
* ``queue_state`` (dict)
mode:
``log_only`` (default) or ``enforce`` (sets mutation_hold on unresolved).
"""
started = now or _utc_now()
mode_norm = (mode or MODE_LOG_ONLY).strip().lower()
if mode_norm not in {MODE_LOG_ONLY, MODE_ENFORCE}:
mode_norm = MODE_LOG_ONLY
inventory_complete = bool(inventory.get("inventory_complete", False))
incomplete_reasons = tuple(
str(r) for r in (inventory.get("incomplete_reasons") or []) if str(r).strip()
)
items: list[ReconcileItem] = []
# --- service health -------------------------------------------------
health = inventory.get("service_health") or {}
if not isinstance(health, Mapping):
health = {}
if not inventory_complete and "service_health" not in inventory:
items.append(
_item(
DIM_SERVICE_HEALTH,
ITEM_UNRESOLVED,
"service health unknown because inventory is incomplete",
details={"inventory_complete": False},
follow_up=True,
)
)
elif health.get("healthy") is True:
items.append(
_item(
DIM_SERVICE_HEALTH,
ITEM_RESOLVED,
"service health verified",
details=dict(health),
)
)
elif health.get("healthy") is False:
items.append(
_item(
DIM_SERVICE_HEALTH,
ITEM_UNRESOLVED,
"service health check failed",
details=dict(health),
follow_up=True,
)
)
else:
items.append(
_item(
DIM_SERVICE_HEALTH,
ITEM_DEGRADED,
"service health not reported; treating as degraded",
details=dict(health),
follow_up=True,
)
)
# --- clients --------------------------------------------------------
clients = list(inventory.get("clients") or [])
disconnected = [
c
for c in clients
if isinstance(c, Mapping) and c.get("connected") is False
]
if "clients" not in inventory:
items.append(
_item(
DIM_CLIENTS,
ITEM_SKIPPED,
"client inventory not supplied",
details={},
)
)
elif disconnected:
items.append(
_item(
DIM_CLIENTS,
ITEM_UNRESOLVED,
f"{len(disconnected)} disconnected client(s) need reconnect",
details={"disconnected": disconnected, "total": len(clients)},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_CLIENTS,
ITEM_RESOLVED,
f"{len(clients)} client(s) accounted for",
details={"total": len(clients)},
)
)
# --- sessions -------------------------------------------------------
sessions = [s for s in (inventory.get("sessions") or []) if isinstance(s, Mapping)]
orphan_sessions = [
s
for s in sessions
if str(s.get("status") or "").lower() == "active"
and s.get("pid") is not None
and not lease_lifecycle.is_process_alive(s.get("pid"))
]
if orphan_sessions:
items.append(
_item(
DIM_SESSIONS,
ITEM_UNRESOLVED,
f"{len(orphan_sessions)} active session row(s) with dead owner pid",
details={
"orphan_session_ids": [s.get("session_id") for s in orphan_sessions],
"total_sessions": len(sessions),
},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_SESSIONS,
ITEM_RESOLVED,
f"{len(sessions)} session row(s) reconciled (no dead-pid orphans)",
details={"total_sessions": len(sessions)},
)
)
# --- checkpoints (#660 soft) ----------------------------------------
checkpoints_available = inventory.get("checkpoints_available")
checkpoints = inventory.get("checkpoints")
if checkpoints_available is False or (
checkpoints is None and "checkpoints" not in inventory
):
items.append(
_item(
DIM_CHECKPOINTS,
ITEM_SKIPPED,
"durable session checkpoint schema not available yet (#660)",
details={"depends_on": 660},
)
)
else:
cp_list = [c for c in (checkpoints or []) if isinstance(c, Mapping)]
stale_cp = [c for c in cp_list if c.get("stale") or c.get("invalid")]
if stale_cp:
items.append(
_item(
DIM_CHECKPOINTS,
ITEM_UNRESOLVED,
f"{len(stale_cp)} checkpoint(s) invalid or stale vs live state",
details={"stale_count": len(stale_cp), "total": len(cp_list)},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_CHECKPOINTS,
ITEM_RESOLVED,
f"{len(cp_list)} checkpoint(s) consistent with live state",
details={"total": len(cp_list)},
)
)
# --- leases / locks -------------------------------------------------
leases = [L for L in (inventory.get("leases") or []) if isinstance(L, Mapping)]
live_leases = [L for L in leases if _is_live_freshness(_lease_freshness(L))]
items.append(
_item(
DIM_LEASES,
ITEM_RESOLVED if inventory_complete else ITEM_DEGRADED,
f"{len(live_leases)} live lease(s) of {len(leases)} inventoried",
details={
"live_count": len(live_leases),
"total": len(leases),
"live_lease_ids": [L.get("lease_id") for L in live_leases],
},
follow_up=not inventory_complete,
)
)
# --- capabilities / stale runtime -----------------------------------
caps = inventory.get("capabilities") or {}
if not isinstance(caps, Mapping):
caps = {}
if caps.get("stale") is True:
items.append(
_item(
DIM_CAPABILITIES,
ITEM_UNRESOLVED,
"runtime code is stale vs on-disk master; restart did not reach parity",
details=dict(caps),
follow_up=True,
)
)
else:
items.append(
_item(
DIM_CAPABILITIES,
ITEM_RESOLVED,
"capability/runtime parity acceptable",
details=dict(caps) if caps else {"stale": False},
)
)
# --- worktrees ------------------------------------------------------
bindings = [
b for b in (inventory.get("worktree_bindings") or []) if isinstance(b, Mapping)
]
missing_wt = [
b
for b in bindings
if b.get("missing") is True or b.get("exists") is False
]
if "worktree_bindings" not in inventory:
items.append(
_item(
DIM_WORKTREES,
ITEM_SKIPPED,
"worktree binding inventory not supplied",
)
)
elif missing_wt:
items.append(
_item(
DIM_WORKTREES,
ITEM_UNRESOLVED,
f"{len(missing_wt)} worktree binding(s) missing on disk",
details={"missing": missing_wt, "total": len(bindings)},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_WORKTREES,
ITEM_RESOLVED,
f"{len(bindings)} worktree binding(s) present",
details={"total": len(bindings)},
)
)
# --- interrupted mutations (AC4) ------------------------------------
pending = [
m
for m in (inventory.get("pending_mutations") or [])
if isinstance(m, Mapping)
]
interrupted = _detect_interrupted_mutations(leases, pending)
if interrupted:
items.append(
_item(
DIM_MUTATIONS,
ITEM_UNRESOLVED,
f"{len(interrupted)} interrupted mutation(s); write resume forbidden",
details={"interrupted": interrupted},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_MUTATIONS,
ITEM_RESOLVED,
"no interrupted mutations detected",
details={"interrupted": []},
)
)
# --- duplicates -----------------------------------------------------
dups = _detect_duplicate_work(leases)
if dups:
items.append(
_item(
DIM_DUPLICATES,
ITEM_UNRESOLVED,
f"{len(dups)} work item(s) have multiple live claims",
details={"duplicates": dups},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_DUPLICATES,
ITEM_RESOLVED,
"no duplicate live claims detected",
details={"duplicates": []},
)
)
# --- queue ----------------------------------------------------------
queue = inventory.get("queue_state")
if queue is None:
items.append(
_item(
DIM_QUEUE,
ITEM_SKIPPED,
"allocator queue state not supplied",
)
)
elif isinstance(queue, Mapping) and queue.get("safe_to_resume") is False:
items.append(
_item(
DIM_QUEUE,
ITEM_UNRESOLVED,
"allocator queue not safe to resume",
details=dict(queue),
follow_up=True,
)
)
else:
items.append(
_item(
DIM_QUEUE,
ITEM_RESOLVED,
"allocator queue state acceptable",
details=dict(queue) if isinstance(queue, Mapping) else {},
)
)
# Incomplete inventory always degrades the whole proof.
if not inventory_complete:
# Ensure at least one follow-up names the incomplete inventory.
items.append(
_item(
"inventory",
ITEM_UNRESOLVED,
"control-plane inventory incomplete; reconcile cannot claim success",
details={"reasons": list(incomplete_reasons)},
follow_up=True,
)
)
resolved = sum(1 for i in items if i.status == ITEM_RESOLVED)
unresolved = sum(1 for i in items if i.status in {ITEM_UNRESOLVED, ITEM_DEGRADED})
skipped = sum(1 for i in items if i.status == ITEM_SKIPPED)
if not inventory_complete or any(i.status == ITEM_UNRESOLVED for i in items):
if any(i.status == ITEM_UNRESOLVED for i in items) and inventory_complete:
overall = STATUS_DEGRADED
elif not inventory_complete:
overall = STATUS_FAILED
else:
overall = STATUS_DEGRADED
elif any(i.status == ITEM_DEGRADED for i in items):
overall = STATUS_DEGRADED
else:
overall = STATUS_COMPLETE
# Enforce mode holds mutations whenever anything is unresolved/failed.
mutation_hold = False
if mode_norm == MODE_ENFORCE and overall in {STATUS_DEGRADED, STATUS_FAILED}:
mutation_hold = True
if mode_norm == MODE_ENFORCE and any(
i.dimension == DIM_MUTATIONS and i.status == ITEM_UNRESOLVED for i in items
):
mutation_hold = True
follow_ups = tuple(
fu for i in items if (fu := _follow_up_for_item(i)) is not None
)
finished = _utc_now() if now is None else now
note = (
"Read-only completion proof. Never auto-resumes write mutations. "
"Unresolved items require durable follow-up before claiming clean restart. "
f"Mode={mode_norm}."
)
return RestartCompletionProof(
schema_version=SCHEMA_VERSION,
reconcile_version=RECONCILE_VERSION,
reconcile_id=(reconcile_id or f"reconcile-{uuid4().hex[:12]}"),
started_at=_ts(started),
finished_at=_ts(finished),
boot_head_sha=(
str(inventory.get("boot_head_sha")).strip()
if inventory.get("boot_head_sha")
else None
),
current_head_sha=(
str(inventory.get("current_head_sha")).strip()
if inventory.get("current_head_sha")
else None
),
inventory_complete=inventory_complete,
incomplete_reasons=incomplete_reasons,
mode=mode_norm,
mutation_hold=mutation_hold,
overall_status=overall,
items=tuple(items),
proposed_follow_ups=follow_ups,
resolved_count=resolved,
unresolved_count=unresolved,
skipped_count=skipped,
note=note,
)
def mutations_allowed(proof: RestartCompletionProof | Mapping[str, Any] | None) -> bool:
"""Return whether write mutations may proceed under the given proof."""
if proof is None:
return True # no proof yet → caller decides; enforce path sets hold
if isinstance(proof, RestartCompletionProof):
return not proof.mutation_hold
if isinstance(proof, Mapping):
return not bool(proof.get("mutation_hold"))
return True
+2
View File
@@ -73,6 +73,8 @@ AUTHOR_TASKS = frozenset({
"claim_issue", "claim_issue",
"create_branch", "create_branch",
"push_branch", "push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
"create_pr", "create_pr",
"comment_pr", "comment_pr",
"address_pr_change_requests", "address_pr_change_requests",
+58
View File
@@ -32,6 +32,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.issue.comment", "permission": "gitea.issue.comment",
"role": "author", "role": "author",
}, },
# #790 Slice A: prove an owned author lease is still active. Strictly
# narrower than lock_issue — it can only slide a lease this exact session
# already owns, never acquire, take over, or revive one — so it gates on the
# same authority rather than introducing an operation name that every
# already-configured author profile would be missing.
"heartbeat_issue_lock": {
"permission": "gitea.issue.comment",
"role": "author",
},
# #860: dirty orphaned same-claimant worktree recovery (explicit operation). # #860: dirty orphaned same-claimant worktree recovery (explicit operation).
"recover_dirty_orphaned_issue_worktree": { "recover_dirty_orphaned_issue_worktree": {
"permission": "gitea.issue.comment", "permission": "gitea.issue.comment",
@@ -78,6 +87,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.branch.create", "permission": "gitea.branch.create",
"role": "author", "role": "author",
}, },
"bootstrap_author_issue_worktree": {
"permission": "gitea.branch.create",
"role": "author",
},
"gitea_bootstrap_author_issue_worktree": {
"permission": "gitea.branch.create",
"role": "author",
},
"push_branch": { "push_branch": {
"permission": "gitea.branch.push", "permission": "gitea.branch.push",
"role": "author", "role": "author",
@@ -115,6 +132,16 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.branch.push", "permission": "gitea.branch.push",
"role": "author", "role": "author",
}, },
# #662: post-restart reconcile is read-only inventory + pure classification.
# Durable follow-up issue creation is a separate apply path (not this task).
"reconcile_after_restart": {
"permission": "gitea.read",
"role": "author",
},
"gitea_reconcile_after_restart": {
"permission": "gitea.read",
"role": "author",
},
# PR synchronization lifecycle: assess is read-only (any role with gitea.read); # PR synchronization lifecycle: assess is read-only (any role with gitea.read);
# update-by-merge is author-only and mutates the PR head via Gitea API. # update-by-merge is author-only and mutates the PR head via Gitea API.
"assess_pr_sync_status": { "assess_pr_sync_status": {
@@ -355,6 +382,21 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"role": "controller", "role": "controller",
}, },
# #642: sanctioned host-daemon lifecycle controls. Deliberately *not* a
# ``gitea.*`` operation — restarting an MCP namespace is a host action, not
# a Gitea API call, and no configured Gitea profile should be able to
# satisfy it by accident. Authority comes from the console RBAC model plus
# out-of-band operator authorization (#630); these entries exist so the
# console cannot invent an authority the capability layer never declared.
"restart_namespace": {
"permission": "runtime.restart_namespace",
"role": "controller",
},
"reload_namespace": {
"permission": "runtime.reload_namespace",
"role": "controller",
},
# #601 first-class lease lifecycle — inspect/list need read; mutations gate on # #601 first-class lease lifecycle — inspect/list need read; mutations gate on
# ownership in the control-plane DB (not a separate Gitea write permission). # ownership in the control-plane DB (not a separate Gitea write permission).
"list_workflow_leases": { "list_workflow_leases": {
@@ -487,6 +529,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.issue.comment", "permission": "gitea.issue.comment",
"role": "author", "role": "author",
}, },
# #651 console analytics ingest — control-plane DB write, not a Gitea API
# call. Authority comes from console RBAC (operator+) plus phase gating;
# permission string is a non-Gitea runtime capability so no Gitea profile
# can satisfy it by accident.
"record_analytics_usage": {
"permission": "runtime.record_analytics_usage",
"role": "author",
},
} }
@@ -497,6 +548,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
# merger lease (#763). # merger lease (#763).
_PREFLIGHT_TASK_TRANSITIONS = frozenset({ _PREFLIGHT_TASK_TRANSITIONS = frozenset({
("review_pr", "acquire_reviewer_pr_lease"), ("review_pr", "acquire_reviewer_pr_lease"),
# #850: native author issue worktree bootstrap
("work_issue", "bootstrap_author_issue_worktree"),
("bootstrap_author_issue_worktree", "lock_issue"),
# #860: dirty-orphan recovery and related work_issue transitions (master)
("work_issue", "lock_issue"), ("work_issue", "lock_issue"),
("work_issue", "recover_dirty_orphaned_issue_worktree"), ("work_issue", "recover_dirty_orphaned_issue_worktree"),
("work_issue", "gitea_recover_dirty_orphaned_issue_worktree"), ("work_issue", "gitea_recover_dirty_orphaned_issue_worktree"),
@@ -548,6 +603,8 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
"gitea_release_merger_pr_lease", "gitea_release_merger_pr_lease",
"create_branch", "create_branch",
"push_branch", "push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
"publish_unpublished_branch", "publish_unpublished_branch",
"create_pr", "create_pr",
"commit_files", "commit_files",
@@ -573,6 +630,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
"gitea_set_issue_labels": "set_issue_labels", "gitea_set_issue_labels": "set_issue_labels",
"gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels", "gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels",
"gitea_create_label": "create_label", "gitea_create_label": "create_label",
"gitea_bootstrap_author_issue_worktree": "bootstrap_author_issue_worktree",
"gitea_commit_files": "commit_files", "gitea_commit_files": "commit_files",
} }
+601
View File
@@ -0,0 +1,601 @@
"""Regression test suite for native author issue worktree bootstrap (#850)."""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import tempfile
import unittest
from unittest import mock
import author_issue_bootstrap
import task_capability_map
def _concurrent_bootstrap_worker(args: tuple[str, int, str, str, str, str]) -> dict:
repo_dir, issue_num, key, lock_dir, journal_dir, master_sha = args
os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = journal_dir
return author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=issue_num,
canonical_repo_root=repo_dir,
expected_base_sha=master_sha,
idempotency_key=key,
lock_dir=lock_dir,
owner_session="session-concurrent-test",
active_identity="jcwalker3",
active_profile="prgs-author",
)
class TestAuthorIssueBootstrap(unittest.TestCase):
"""Test suite covering AC1-AC10 and comment #14959 specification."""
def setUp(self):
self.tmp_dir = tempfile.mkdtemp(prefix="test_bootstrap_")
self.repo_dir = os.path.join(self.tmp_dir, "repo")
os.makedirs(self.repo_dir)
# Initialize synthetic git repo
subprocess.run(["git", "init", "-b", "master"], cwd=self.repo_dir, check=True, capture_output=True)
subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.repo_dir, check=True)
subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.repo_dir, check=True)
readme = os.path.join(self.repo_dir, "README.md")
with open(readme, "w", encoding="utf-8") as f:
f.write("# Test Repo\n")
subprocess.run(["git", "add", "README.md"], cwd=self.repo_dir, check=True, capture_output=True)
subprocess.run(["git", "commit", "-m", "initial commit"], cwd=self.repo_dir, check=True, capture_output=True)
rev_res = subprocess.run(["git", "rev-parse", "HEAD"], cwd=self.repo_dir, capture_output=True, text=True, check=True)
self.master_sha = rev_res.stdout.strip()
self.branches_dir = os.path.join(self.repo_dir, "branches")
os.makedirs(self.branches_dir, exist_ok=True)
self.lock_dir = os.path.join(self.tmp_dir, "locks")
os.makedirs(self.lock_dir, exist_ok=True)
self.journal_dir = os.path.join(self.tmp_dir, "journals")
os.makedirs(self.journal_dir, exist_ok=True)
os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = self.journal_dir
def tearDown(self):
os.environ.pop("GITEA_BOOTSTRAP_JOURNAL_DIR", None)
shutil.rmtree(self.tmp_dir, ignore_errors=True)
def test_bootstrap_success_path(self):
"""AC1/AC3/AC8: Successful bootstrap creates branch, worktree, registration, and lock proof."""
key = "test_key_success_1"
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
# assignment_id/lease_id omitted: optional unless verified live.
expected_base_sha=self.master_sha,
idempotency_key=key,
remote="prgs",
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertTrue(res.get("success"), f"Bootstrap failed: {res}")
self.assertFalse(res.get("replayed"))
self.assertEqual(res.get("issue_number"), 850)
self.assertEqual(res.get("base_sha"), self.master_sha)
self.assertIn("branches/fix-issue-850-native-mcp-bootstrap", res.get("worktree_path"))
# Verify worktree directory exists and is registered
worktree_path = res["worktree_path"]
self.assertTrue(os.path.isdir(worktree_path))
wt_list = subprocess.run(["git", "-C", self.repo_dir, "worktree", "list"], capture_output=True, text=True, check=True)
self.assertIn(worktree_path, wt_list.stdout)
# Verify phase journal written
journal = author_issue_bootstrap.load_phase_journal(key, journal_dir=self.lock_dir)
self.assertIsNotNone(journal)
self.assertTrue(journal.get("completed"))
self.assertEqual(journal.get("current_phase"), author_issue_bootstrap.PHASE_7_TRANSITION_COMPLETED)
def test_idempotent_replay(self):
"""Item 2: Replaying with identical key returns cached transition without duplicate creation."""
key = "test_key_idempotent_1"
res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertTrue(res1["success"], f"res1 failed: {res1}")
self.assertFalse(res1.get("replayed"))
# Second call
res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertTrue(res2["success"], f"res2 failed: {res2}")
self.assertTrue(res2.get("replayed"))
self.assertEqual(res1["worktree_path"], res2["worktree_path"])
def test_stale_concurrency_pin_refusal(self):
"""Item 3: Mismatched expected base SHA fails closed without silent rebasing."""
stale_sha = "0000000000000000000000000000000000000000"
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
expected_base_sha=stale_sha,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "stale_concurrency_pin")
self.assertIn("exact_next_action", res)
def test_path_outside_branches_root_refusal(self):
"""Item 6: Worktree path outside branches/ root is refused."""
outside_path = os.path.join(self.tmp_dir, "outside_worktree")
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
worktree_path=outside_path,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "path_outside_canonical_branches_root")
def test_preexisting_dirty_worktree_preservation(self):
"""Item 6: Preexisting dirty worktree fails closed and is NOT modified or cleaned."""
branch = "fix/issue-850-dirty-test"
wt_path = os.path.join(self.branches_dir, "fix-issue-850-dirty-test")
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True)
# Create dirty untracked file
dirty_file = os.path.join(wt_path, "dirty.txt")
with open(dirty_file, "w") as f:
f.write("dirty edits\n")
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name=branch,
worktree_path=wt_path,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "preexisting_dirty_worktree")
# Prove dirty file is preserved byte-for-byte
self.assertTrue(os.path.exists(dirty_file))
with open(dirty_file, "r") as f:
self.assertEqual(f.read(), "dirty edits\n")
def test_compensating_recovery_on_failed_phase(self):
"""AC4/Item 4: Failure during transition rolls back ONLY newly created artifacts."""
key = "test_key_recovery_1"
# Simulate partial progress in journal
journal = {
"idempotency_key": key,
"issue_number": 850,
"branch_name": "fix/issue-850-recovery-test",
"worktree_path": os.path.join(self.branches_dir, "fix-issue-850-recovery-test"),
"artifacts_created": {
"branch_created": True,
"worktree_dir_created": True,
"worktree_registered": True,
"lock_created": False,
},
"failure_reason": "simulated lock failure",
"current_phase": author_issue_bootstrap.PHASE_5_REGISTRATION_VERIFIED,
"completed": False,
}
# Create the branch and worktree manually to simulate partial state
subprocess.run(["git", "-C", self.repo_dir, "branch", journal["branch_name"]], check=True, capture_output=True)
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", journal["worktree_path"], journal["branch_name"]], check=True, capture_output=True)
# Run compensating recovery
rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir)
self.assertTrue(rec["executed"])
self.assertIn(f"worktree_path:{journal['worktree_path']}", rec["rolled_back"])
self.assertIn(f"branch:{journal['branch_name']}", rec["rolled_back"])
# Prove worktree directory and branch were rolled back
self.assertFalse(os.path.exists(journal["worktree_path"]))
branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", journal["branch_name"]], capture_output=True, text=True, check=False)
self.assertNotEqual(branch_check.returncode, 0)
def test_cross_process_concurrency(self):
"""Review #525 Finding 1: Genuine cross-process concurrency locking prevents corruption."""
import concurrent.futures
key = "test_concurrent_key_850"
args = (self.repo_dir, 850, key, self.lock_dir, self.journal_dir, self.master_sha)
with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
fut1 = executor.submit(_concurrent_bootstrap_worker, args)
fut2 = executor.submit(_concurrent_bootstrap_worker, args)
res1 = fut1.result(timeout=10)
res2 = fut2.result(timeout=10)
self.assertTrue(res1["success"], f"res1 failed: {res1}")
self.assertTrue(res2["success"], f"res2 failed: {res2}")
# One process performs creation, the other process receives idempotent replay
replayed_count = sum(1 for r in (res1, res2) if r.get("replayed"))
created_count = sum(1 for r in (res1, res2) if not r.get("replayed"))
self.assertEqual(replayed_count, 1)
self.assertEqual(created_count, 1)
self.assertEqual(res1["worktree_path"], res2["worktree_path"])
def test_interrupted_replay_preserves_artifacts_created_provenance(self):
"""Review #525 Finding 2: Replaying incomplete journal preserves creation provenance monotonically."""
key = "test_key_interrupted_replay_1"
branch = "fix/issue-850-interrupted-replay"
wt_path = os.path.join(self.branches_dir, "fix-issue-850-interrupted-replay")
# Simulate Phase 2/3 completion where branch and worktree directory were created by this transition
journal = {
"idempotency_key": key,
"issue_number": 850,
"branch_name": branch,
"worktree_path": wt_path,
"active_identity": "jcwalker3",
"active_profile": "prgs-author",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"phases": {
author_issue_bootstrap.PHASE_1_REQUEST_ACCEPTED: {"status": "completed"},
author_issue_bootstrap.PHASE_2_BRANCH_CONFIRMED: {"status": "completed", "created": True},
},
"artifacts_created": {
"branch_created": True,
"worktree_dir_created": True,
"worktree_registered": True,
"lock_created": False,
},
"current_phase": author_issue_bootstrap.PHASE_3_PATH_RESERVED,
"completed": False,
}
# Pre-create the branch and worktree on disk to simulate partial state after crash
subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True)
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True)
author_issue_bootstrap.save_phase_journal(journal, journal_dir=self.lock_dir)
# Now resume/replay the transition but simulate lock binding failure during Phase 6
with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Lock failure test")):
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name=branch,
worktree_path=wt_path,
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "issue_lock_acquisition_failed")
# Verify that compensating recovery correctly deleted transition-created branch & worktree
# because creation provenance was preserved across replay (NOT downgraded to False!)
self.assertFalse(os.path.exists(wt_path))
branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False)
self.assertNotEqual(branch_check.returncode, 0)
def test_transition_created_only_compensation(self):
"""Review #525 Finding 4: Preexisting branch is NOT deleted by compensation when only worktree was transition-created."""
key = "test_key_preexisting_branch_compensation"
preexisting_branch = "fix/issue-850-preexisting"
wt_path = os.path.join(self.branches_dir, "fix-issue-850-preexisting")
# Create branch BEFORE bootstrap (preexisting branch)
subprocess.run(["git", "-C", self.repo_dir, "branch", preexisting_branch, self.master_sha], check=True, capture_output=True)
# Call bootstrap with simulated failure during Phase 6 (lock binding)
with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Simulated lock failure")):
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name=preexisting_branch,
worktree_path=wt_path,
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
# Worktree dir was created by transition -> removed by compensation
self.assertFalse(os.path.exists(wt_path))
# Preexisting branch was NOT created by transition -> MUST BE PRESERVED!
branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", preexisting_branch], capture_output=True, text=True, check=False)
self.assertEqual(branch_check.returncode, 0, "Preexisting branch was deleted by mistake!")
def test_incompatible_idempotency_replay_refusal(self):
"""Review #525 Finding 4: Replaying key with incompatible parameters returns refusal."""
key = "test_key_incompatible_replay"
res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name="fix/issue-850-param-a",
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertTrue(res1["success"])
# Second call with different branch_name
res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name="fix/issue-850-param-b",
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res2["success"])
self.assertEqual(res2.get("reason_code"), "incompatible_idempotency_replay")
def test_exact_next_action_satisfiable_via_mcp(self):
"""Review #525 Finding 4: exact_next_action provides satisfiable MCP actions, not shell commands."""
key = "test_key_next_action_mcp"
stale_sha = "0000000000000000000000000000000000000000"
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
expected_base_sha=stale_sha,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
next_action = res.get("exact_next_action", "")
self.assertNotIn("scripts/worktree-start", next_action)
self.assertNotIn("git worktree add", next_action)
self.assertNotIn("bash", next_action.lower())
def test_missing_owner_session_refusal(self):
"""Finding D: Missing owner_session context fails closed with typed refusal and zero mutation."""
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
owner_session=None,
lock_dir=self.lock_dir,
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "missing_owner_session")
self.assertIn("exact_next_action", res)
def test_symlink_lock_file_refusal(self):
"""Finding C: BootstrapTransitionLock refuses to follow symlinks."""
key = "test_symlink_lock_key"
safe_key = "".join(c if c.isalnum() or c in ("-", "_", ".") else "_" for c in key)
lock_path = os.path.join(self.lock_dir, f"{safe_key}.lock")
target_file = os.path.join(self.tmp_dir, "fake_target")
with open(target_file, "w") as f:
f.write("target")
os.symlink(target_file, lock_path)
with self.assertRaises(RuntimeError) as ctx:
with author_issue_bootstrap.BootstrapTransitionLock(key, journal_dir=self.lock_dir):
pass
self.assertIn("symlink", str(ctx.exception).lower())
def test_lock_directory_escape_refusal(self):
"""Finding C: BootstrapTransitionLock refuses keys that escape lock directory."""
with mock.patch("os.path.abspath", return_value="/tmp/outside/evil_key.lock"):
with self.assertRaises(RuntimeError) as ctx:
author_issue_bootstrap.BootstrapTransitionLock("key", journal_dir=self.lock_dir)
self.assertIn("escapes", str(ctx.exception).lower())
def test_missing_active_identity_refusal(self):
"""F-5: Missing active_identity parameter fails closed."""
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
owner_session="session-test-1234",
active_identity=None,
active_profile="prgs-author",
lock_dir=self.lock_dir,
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "missing_active_identity")
def test_missing_active_profile_refusal(self):
"""F-5: Missing active_profile parameter fails closed."""
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
owner_session="session-test-1234",
active_identity="jcwalker3",
active_profile=None,
lock_dir=self.lock_dir,
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "missing_active_profile")
def test_dirty_worktree_preserved_during_recovery(self):
"""F-4: Compensating recovery does not delete dirty worktree."""
branch = "fix/issue-850-rec-dirty"
wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-dirty")
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True)
dirty_file = os.path.join(wt_path, "dirty.txt")
with open(dirty_file, "w") as f:
f.write("uncommitted work")
journal = {
"idempotency_key": "test_dirty_rec",
"issue_number": 850,
"branch_name": branch,
"worktree_path": wt_path,
"artifacts_created": {
"worktree_dir_created": True,
"worktree_registered": True,
},
"failure_reason": "test dirty recovery",
}
rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir)
self.assertTrue(os.path.exists(wt_path))
self.assertIn(f"worktree_path_preserved_dirty:{wt_path}", rec["rolled_back"])
def test_branch_with_commits_preserved_during_recovery(self):
"""F-4: Compensating recovery does not delete branch with author commits."""
branch = "fix/issue-850-rec-commits"
subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True)
# Add a commit on the branch
wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-commits")
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True)
cfile = os.path.join(wt_path, "commit.txt")
with open(cfile, "w") as f:
f.write("author commit")
subprocess.run(["git", "-C", wt_path, "add", "commit.txt"], check=True, capture_output=True)
subprocess.run(["git", "-C", wt_path, "commit", "-m", "author commit"], check=True, capture_output=True)
subprocess.run(["git", "-C", self.repo_dir, "worktree", "remove", "--force", wt_path], check=True, capture_output=True)
journal = {
"idempotency_key": "test_commits_rec",
"issue_number": 850,
"branch_name": branch,
"resolved_base_sha": self.master_sha,
"artifacts_created": {
"branch_created": True,
},
"failure_reason": "test commit branch recovery",
}
rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir)
branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False)
self.assertEqual(branch_check.returncode, 0, "Branch with commits was deleted!")
self.assertIn(f"branch_preserved_commits:{branch}", rec["rolled_back"])
def test_task_capability_map_integration(self):
"""Verify task_capability_map has bootstrap_author_issue_worktree configured correctly."""
self.assertEqual(task_capability_map.required_role("bootstrap_author_issue_worktree"), "author")
self.assertEqual(task_capability_map.required_permission("bootstrap_author_issue_worktree"), "gitea.branch.create")
self.assertTrue(task_capability_map.preflight_task_matches("work_issue", "bootstrap_author_issue_worktree"))
self.assertTrue(task_capability_map.preflight_task_matches("bootstrap_author_issue_worktree", "lock_issue"))
def test_unverified_assignment_lease_ids_fail_closed(self):
"""Review #531 Finding 4: fabricated assignment/lease IDs are refused."""
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
assignment_id="asn-fabricated",
lease_id="lease-fabricated",
expected_base_sha=self.master_sha,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
self.assertIn(
res.get("reason_code"),
{
"unknown_lease_id",
"assignment_lease_lookup_failed",
"incomplete_assignment_lease_ids",
},
)
def test_partial_assignment_lease_ids_fail_closed(self):
"""Review #531 Finding 4: one of assignment_id/lease_id alone is incomplete."""
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
assignment_id="asn-only",
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "incomplete_assignment_lease_ids")
def test_stale_diverged_branch_is_not_accepted_via_merge_base(self):
"""Review #531 Finding 3: any common ancestor is not enough; require master ⊆ branch."""
branch = "fix/issue-850-stale-divergent"
# Create branch from current master, then advance master so branch lacks tip.
subprocess.run(
["git", "-C", self.repo_dir, "branch", branch, self.master_sha],
check=True,
capture_output=True,
)
# Make a new commit on master (orphan path so branch does not contain it).
marker = os.path.join(self.repo_dir, "master-advance.txt")
with open(marker, "w") as f:
f.write("advance master\n")
subprocess.run(["git", "-C", self.repo_dir, "add", "master-advance.txt"], check=True, capture_output=True)
subprocess.run(
["git", "-C", self.repo_dir, "commit", "-m", "advance master past branch"],
check=True,
capture_output=True,
)
new_master = subprocess.run(
["git", "-C", self.repo_dir, "rev-parse", "HEAD"],
capture_output=True,
text=True,
check=True,
).stdout.strip()
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name=branch,
expected_base_sha=new_master,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "incompatible_existing_branch")
def test_compensating_recovery_attempts_lease_release(self):
"""Review #531 Finding 5: recovery invokes lease release when lease_id is present."""
from unittest import mock
journal = {
"idempotency_key": "test_lease_rec",
"issue_number": 850,
"owner_session": "session-test-1234",
"lease_id": "lease-abc",
"branch_name": "fix/issue-850-lease-rec",
"artifacts_created": {"lock_created": True},
"failure_reason": "simulated",
"completed": False,
}
with mock.patch.object(
author_issue_bootstrap.lease_lifecycle,
"release_lease",
return_value={"success": True},
) as rel, mock.patch.object(
author_issue_bootstrap.control_plane_db,
"ControlPlaneDB",
return_value=mock.Mock(),
):
rec = author_issue_bootstrap.run_compensating_recovery(
journal, self.repo_dir, journal_dir=self.lock_dir
)
self.assertTrue(rec["executed"])
rel.assert_called_once()
self.assertIn("lease:lease-abc", rec["rolled_back"])
class TestCanonicalRootNoStringSplit(unittest.TestCase):
def test_fallback_uses_commonpath_not_substring_split(self):
"""Review #531 Finding 2: no norm.split('/branches/') fallback."""
import inspect
import author_mutation_worktree as amw
src = inspect.getsource(amw.resolve_canonical_repo_root)
self.assertNotIn('split("/branches/")', src)
self.assertNotIn("split('/branches/')", src)
# Fallback recovers repo root from a nested branches worktree path.
with tempfile.TemporaryDirectory() as tmp:
repo = os.path.join(tmp, "repo")
wt = os.path.join(repo, "branches", "fix-issue-850-x")
os.makedirs(wt)
# git unavailable path: pass missing workspace so fallback is used.
root = amw.resolve_canonical_repo_root("/missing/path", wt)
self.assertEqual(root, os.path.realpath(repo))
if __name__ == "__main__":
unittest.main()
+22
View File
@@ -28,6 +28,21 @@ class TestPathUnderBranches(unittest.TestCase):
amw.is_path_under_branches("/repo/other-checkout", self.ROOT) amw.is_path_under_branches("/repo/other-checkout", self.ROOT)
) )
def test_unrelated_branches_dir_fails(self):
self.assertFalse(
amw.is_path_under_branches("/tmp/branches/evil", self.ROOT)
)
def test_prefix_confusion_fails(self):
self.assertFalse(
amw.is_path_under_branches(f"{self.ROOT}/branches-other/foo", self.ROOT)
)
def test_traversal_fails(self):
self.assertFalse(
amw.is_path_under_branches(f"{self.ROOT}/branches/../evil", self.ROOT)
)
class TestAssessAuthorMutationWorktree(unittest.TestCase): class TestAssessAuthorMutationWorktree(unittest.TestCase):
ROOT = "/repo/Gitea-Tools" ROOT = "/repo/Gitea-Tools"
@@ -71,6 +86,13 @@ class TestAssessAuthorMutationWorktree(unittest.TestCase):
self.assertTrue(result["proven"]) self.assertTrue(result["proven"])
self.assertFalse(result["block"]) self.assertFalse(result["block"])
def test_path_shaped_branches_ancestry_without_isdir(self):
"""Review #551: commonpath recovery must not require on-disk isdir."""
fake_wt = "/repo/Gitea-Tools/branches/issue-274"
root = amw.resolve_canonical_repo_root(fake_wt, fake_wt)
self.assertEqual(root, "/repo/Gitea-Tools")
self.assertNotIn('split("/branches/")', open(amw.__file__).read())
class TestPreflightIntegration(unittest.TestCase): class TestPreflightIntegration(unittest.TestCase):
def test_verify_preflight_blocks_control_checkout_with_test_porcelain(self): def test_verify_preflight_blocks_control_checkout_with_test_porcelain(self):
+199
View File
@@ -0,0 +1,199 @@
"""Regression tests for #628 building blocks (child scope only).
Honest scope: unit coverage of pre-existing APIs used by umbrella #628.
This module does **not** implement or verify all 21 umbrella acceptance
criteria, automatic handoff store/retrieve, multi-worker product wiring,
or end-to-end orchestration.
Covered building blocks:
- CTH format / parse / assess (`format_cth_body`, `parse_cth_comment`,
`assess_cth_comment`)
- Exclusive-ownership skip classification (`classify_skip` with
OWNERSHIP_FOREIGN vs OWNERSHIP_OWN)
- Durable dependency edges (`upsert_dependency_edge` / list) and skip
when dependency_unmet
- Edge state transition UNMET -> MET
Parent umbrella remains #628; this slice is a scoped child issue only.
"""
import unittest
from unittest.mock import MagicMock, patch
import os
import json
import tempfile
from canonical_thread_handoff import (
format_cth_body,
parse_cth_comment,
assess_cth_comment,
)
import dependency_graph
from control_plane_db import ControlPlaneDB
from allocator_service import (
WorkCandidate,
classify_skip,
ROLE_AUTHOR,
ROLE_REVIEWER,
ROLE_MERGER,
ROLE_RECONCILER,
OWNERSHIP_OWN,
OWNERSHIP_FOREIGN,
)
class TestIssue628Orchestration(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
self.db = ControlPlaneDB(self.db_path)
def tearDown(self):
self._tmp.cleanup()
def test_canonical_handoff_serialization_and_retrieval(self):
"""Building block: format/parse/assess a CTH body (not full AC1/AC2 product path)."""
handoff = format_cth_body(
cth_type="Author Handoff",
status="completed",
next_owner="reviewer",
current_blocker="none",
decision="Implementation complete, tests passing",
proof="pytest tests/test_issue_628_orchestration.py passed",
next_action="Review PR and run reviewer pre-flight",
ready_to_paste_prompt="Review PR for child issue #878 (parent #628)",
)
self.assertIn("CTH: Author Handoff", handoff)
parsed = parse_cth_comment(handoff)
self.assertIsNotNone(parsed)
self.assertEqual(parsed["cth_type"], "Author Handoff")
assessment = assess_cth_comment(handoff)
self.assertFalse(assessment["block"])
def test_exclusive_task_unit_single_owner(self):
"""Building block: classify_skip foreign vs own ownership."""
candidate = WorkCandidate(
kind="issue",
number=878,
title="Child #878 ownership classify candidate",
state="open",
labels=["status:in-progress"],
blocked=False,
dependency_unmet=False,
)
# Foreign ownership MUST be skipped
skip_foreign = classify_skip(
c=candidate,
role=ROLE_AUTHOR,
terminal_pr=None,
claim_ownership=OWNERSHIP_FOREIGN,
)
self.assertIsNotNone(skip_foreign)
self.assertIn("active lease", skip_foreign)
# Own/Self claim remains selectable for session resumption
skip_self = classify_skip(
c=candidate,
role=ROLE_AUTHOR,
terminal_pr=None,
claim_ownership=OWNERSHIP_OWN,
)
self.assertIsNone(skip_self)
def test_durable_dependency_graph_blocking(self):
"""Building block: unmet dependency edge + classify_skip on dependency_unmet."""
# Upsert a blocking dependency edge between issue 878 and blocker 601
self.db.upsert_dependency_edge(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
source_kind="issue",
source_number=878,
target_kind="issue",
target_number=601,
edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
state=dependency_graph.STATE_UNMET,
blocking_condition="Target issue #601 is not closed",
completion_condition="Target issue #601 is closed",
evidence={"source": "unit_test"},
)
edges = self.db.list_dependency_edges(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
source_kind="issue",
source_number=878,
)
self.assertEqual(len(edges), 1)
self.assertEqual(edges[0]["state"], "unmet")
self.assertEqual(edges[0]["target_number"], 601)
# When dependency is unmet, candidate is blocked from selection
candidate = WorkCandidate(
kind="issue",
number=878,
title="Blocked candidate",
state="open",
labels=[],
blocked=False,
dependency_unmet=True,
dependency_reason="issue#878 is blocked by unmet dependency issue#601",
)
skip_reason = classify_skip(
c=candidate,
role=ROLE_AUTHOR,
terminal_pr=None,
claim_ownership=OWNERSHIP_OWN,
)
self.assertIsNotNone(skip_reason)
self.assertIn("issue#601", skip_reason)
def test_dependency_completion_reevaluation(self):
"""Building block: dependency edge state can transition UNMET -> MET."""
self.db.upsert_dependency_edge(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
source_kind="issue",
source_number=878,
target_kind="issue",
target_number=601,
edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
state=dependency_graph.STATE_UNMET,
blocking_condition="Target issue #601 is open",
completion_condition="Target issue #601 is closed",
evidence={"source": "unit_test"},
)
# Mark edge as met upon target issue closure
self.db.upsert_dependency_edge(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
source_kind="issue",
source_number=878,
target_kind="issue",
target_number=601,
edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
state=dependency_graph.STATE_MET,
blocking_condition="Target issue #601 is open",
completion_condition="Target issue #601 is closed",
evidence={"source": "target_closed_event"},
)
edges = self.db.list_dependency_edges(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
source_kind="issue",
source_number=878,
)
self.assertEqual(len(edges), 1)
self.assertEqual(edges[0]["state"], "met")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,249 @@
"""Tests for post-restart MCP reconciliation and completion proof (#662)."""
from __future__ import annotations
import json
import os
import unittest
from datetime import datetime, timezone
import post_restart_reconcile as prr
NOW = datetime(2026, 7, 24, 12, 0, 0, tzinfo=timezone.utc)
def _base_inventory(**overrides):
inv = {
"inventory_complete": True,
"incomplete_reasons": [],
"service_health": {"healthy": True},
"clients": [{"session_id": "c1", "connected": True}],
"sessions": [
{
"session_id": "s-live",
"status": "active",
"pid": os.getpid(),
"role": "author",
}
],
"leases": [],
"checkpoints_available": False,
"worktree_bindings": [{"path": "/tmp/wt", "exists": True}],
"pending_mutations": [],
"capabilities": {"stale": False},
"boot_head_sha": "a" * 40,
"current_head_sha": "a" * 40,
"queue_state": {"safe_to_resume": True},
}
inv.update(overrides)
return inv
class IncompleteInventoryTest(unittest.TestCase):
def test_incomplete_inventory_fails_closed(self) -> None:
proof = prr.reconcile_after_restart(
{
"inventory_complete": False,
"incomplete_reasons": ["control-plane DB unavailable"],
},
now=NOW,
mode=prr.MODE_ENFORCE,
reconcile_id="test-incomplete",
)
self.assertEqual(proof.overall_status, prr.STATUS_FAILED)
self.assertTrue(proof.mutation_hold)
self.assertFalse(proof.inventory_complete)
self.assertTrue(proof.proposed_follow_ups)
self.assertIn("control-plane DB unavailable", proof.incomplete_reasons)
class HappyPathTest(unittest.TestCase):
def test_clean_restart_is_complete_without_mutation_hold(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(),
now=NOW,
mode=prr.MODE_ENFORCE,
reconcile_id="test-clean",
)
self.assertEqual(proof.overall_status, prr.STATUS_COMPLETE)
self.assertFalse(proof.mutation_hold)
self.assertEqual(proof.unresolved_count, 0)
cp = next(i for i in proof.items if i.dimension == prr.DIM_CHECKPOINTS)
self.assertEqual(cp.status, prr.ITEM_SKIPPED)
links = proof.as_dict()["links"]
self.assertEqual(links["umbrella"], 655)
self.assertEqual(links["issue"], 662)
self.assertEqual(links["vision"], 652)
self.assertEqual(links["roadmap"], 653)
class InterruptedMutationTest(unittest.TestCase):
def test_mutating_lease_with_dead_owner_is_unresolved(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
leases=[
{
"lease_id": "lease-mut",
"session_id": "s-dead",
"phase": "implementing",
"work_kind": "issue",
"work_number": 662,
"worktree_path": "/tmp/wt-662",
"freshness": {"freshness": "stale_dead_process"},
}
]
),
now=NOW,
mode=prr.MODE_ENFORCE,
)
mut = next(i for i in proof.items if i.dimension == prr.DIM_MUTATIONS)
self.assertEqual(mut.status, prr.ITEM_UNRESOLVED)
self.assertTrue(mut.follow_up_required)
interrupted = mut.details["interrupted"]
self.assertEqual(len(interrupted), 1)
self.assertFalse(interrupted[0]["resume_allowed"])
self.assertTrue(proof.mutation_hold)
self.assertTrue(
any(f.dimension == prr.DIM_MUTATIONS for f in proof.proposed_follow_ups)
)
def test_explicit_pending_mutation_inventory(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
pending_mutations=[
{
"session_id": "s1",
"phase": "publishing",
"work_kind": "pr",
"work_number": 856,
"reason": "push interrupted mid-flight",
}
]
),
now=NOW,
mode=prr.MODE_LOG_ONLY,
)
mut = next(i for i in proof.items if i.dimension == prr.DIM_MUTATIONS)
self.assertEqual(mut.status, prr.ITEM_UNRESOLVED)
# log_only never holds mutations even when unresolved
self.assertFalse(proof.mutation_hold)
self.assertEqual(proof.overall_status, prr.STATUS_DEGRADED)
class DuplicateClaimsTest(unittest.TestCase):
def test_duplicate_live_claims_flagged(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
leases=[
{
"lease_id": "l1",
"session_id": "s1",
"phase": "allocated",
"work_kind": "issue",
"work_number": 100,
"freshness": {"freshness": "active"},
},
{
"lease_id": "l2",
"session_id": "s2",
"phase": "allocated",
"work_kind": "issue",
"work_number": 100,
"freshness": {"freshness": "active"},
},
]
),
now=NOW,
mode=prr.MODE_ENFORCE,
)
dups = next(i for i in proof.items if i.dimension == prr.DIM_DUPLICATES)
self.assertEqual(dups.status, prr.ITEM_UNRESOLVED)
self.assertEqual(dups.details["duplicates"][0]["claim_count"], 2)
self.assertTrue(proof.mutation_hold)
class OrphanSessionTest(unittest.TestCase):
def test_active_session_dead_pid_is_unresolved(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
sessions=[
{
"session_id": "ghost",
"status": "active",
"pid": 2_000_000_000,
"role": "author",
}
]
),
now=NOW,
mode=prr.MODE_ENFORCE,
)
sess = next(i for i in proof.items if i.dimension == prr.DIM_SESSIONS)
self.assertEqual(sess.status, prr.ITEM_UNRESOLVED)
self.assertIn("ghost", sess.details["orphan_session_ids"])
class CapabilityStaleTest(unittest.TestCase):
def test_stale_runtime_unresolved(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(capabilities={"stale": True, "startup_head": "aaa"}),
now=NOW,
mode=prr.MODE_ENFORCE,
)
caps = next(i for i in proof.items if i.dimension == prr.DIM_CAPABILITIES)
self.assertEqual(caps.status, prr.ITEM_UNRESOLVED)
self.assertTrue(proof.mutation_hold)
class MutationsAllowedHelperTest(unittest.TestCase):
def test_mutations_allowed_respects_hold(self) -> None:
held = prr.reconcile_after_restart(
_base_inventory(
pending_mutations=[{"phase": "merging", "session_id": "x"}]
),
now=NOW,
mode=prr.MODE_ENFORCE,
)
self.assertFalse(prr.mutations_allowed(held))
self.assertFalse(prr.mutations_allowed(held.as_dict()))
clean = prr.reconcile_after_restart(
_base_inventory(), now=NOW, mode=prr.MODE_ENFORCE
)
self.assertTrue(prr.mutations_allowed(clean))
class CheckpointSoftDependencyTest(unittest.TestCase):
def test_checkpoints_when_schema_present(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
checkpoints_available=True,
checkpoints=[{"session_id": "s1", "stale": False}],
),
now=NOW,
)
cp = next(i for i in proof.items if i.dimension == prr.DIM_CHECKPOINTS)
self.assertEqual(cp.status, prr.ITEM_RESOLVED)
def test_stale_checkpoints_unresolved(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
checkpoints_available=True,
checkpoints=[{"session_id": "s1", "stale": True}],
),
now=NOW,
mode=prr.MODE_ENFORCE,
)
cp = next(i for i in proof.items if i.dimension == prr.DIM_CHECKPOINTS)
self.assertEqual(cp.status, prr.ITEM_UNRESOLVED)
class ProofSerializationTest(unittest.TestCase):
def test_as_dict_is_json_friendly(self) -> None:
proof = prr.reconcile_after_restart(_base_inventory(), now=NOW)
blob = json.dumps(proof.as_dict())
self.assertIn("reconcile_id", blob)
self.assertIn("proposed_follow_ups", blob)
if __name__ == "__main__":
unittest.main()
+444
View File
@@ -0,0 +1,444 @@
"""Task heartbeat through the native MCP author path (#790 Slice A, AC-N6).
Assessor-level coverage is not sufficient here, and this project has already
paid for learning that: in review #499 on PR #791 the #760 renewal waiver was
computed correctly and then *discarded* at two later gates, so every real
renewal still failed while the unit suite stayed green. AC-N6 exists because of
that, and requires driving the real tools against a real git repository and a
real durable lock file, composing the gates in production order.
These tests therefore call ``gitea_lock_issue`` and
``gitea_heartbeat_issue_lock`` themselves and assert on what lands on disk,
never on an assessor's return value alone.
"""
from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mutation_profile_fixture import shared_mutation_env # noqa: E402
import issue_lock_provenance # noqa: E402
import issue_lock_store # noqa: E402
import lease_policy # noqa: E402
import mcp_server # noqa: E402
ISSUE = 9791
BRANCH = f"fix/issue-{ISSUE}-heartbeat-mcp"
IDENTITY = "example-user"
PROFILE = "test-author-prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"
def _ts(moment: datetime) -> str:
return (
moment.astimezone(timezone.utc)
.replace(microsecond=0)
.isoformat()
.replace("+00:00", "Z")
)
class _HeartbeatMcpBase(unittest.TestCase):
"""Real git repo plus a real durable lock, driven through the real tools."""
def setUp(self):
self.lock_dir = tempfile.TemporaryDirectory()
self.addCleanup(self.lock_dir.cleanup)
self.repo = tempfile.mkdtemp(prefix="issue790-mcp-")
self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.repo], check=False))
self._init_worktree()
self.remotes = patch.dict(
mcp_server.REMOTES,
{"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
)
self.remotes.start()
self.addCleanup(patch.stopall)
mcp_server._IDENTITY_CACHE.clear()
def _git(self, *args):
return subprocess.run(
["git", "-C", self.repo, *args], capture_output=True, text=True, check=True
)
def _init_worktree(self):
self._git("init", "-q", "-b", "master")
self._git("config", "user.email", "[email protected]")
self._git("config", "user.name", "Test")
with open(os.path.join(self.repo, "seed.txt"), "w") as fh:
fh.write("seed\n")
self._git("add", "seed.txt")
self._git("commit", "-q", "-m", "seed")
self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
# A fresh claim starts base-equivalent, which is the ordinary first-lock
# shape and exercises assess_issue_lock_worktree on its normal path.
self._git("checkout", "-q", "-b", BRANCH)
self.head_sha = self.base_sha
self.worktree = os.path.realpath(self.repo)
def _lock_path(self):
return issue_lock_store.lock_file_path(
remote="prgs",
org=ORG,
repo=REPO,
issue_number=ISSUE,
lock_dir=self.lock_dir.name,
)
def _tool_env(self):
env = shared_mutation_env(
PROFILE, include_example_repo=True, GITEA_ISSUE_LOCK_DIR=self.lock_dir.name
)
env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
return env
def _git_state(self, *, porcelain="", base_equivalent=True):
return {
"current_branch": BRANCH,
"porcelain_status": porcelain,
"base_equivalent": base_equivalent,
"head_sha": self.head_sha,
"inspected_git_root": self.worktree,
"base_branch": "master",
}
def run_lock_issue(
self,
*,
branch_entries=None,
open_prs=None,
git_state=None,
identity=IDENTITY,
profile=PROFILE,
):
branch_entries = branch_entries if branch_entries is not None else []
open_prs = open_prs if open_prs is not None else []
git_state = git_state or self._git_state()
env = self._tool_env()
with patch(
"mcp_server.api_get_all", return_value=list(branch_entries)
), patch(
"mcp_server._list_open_pulls", return_value=list(open_prs)
), patch(
"mcp_server.get_auth_header", return_value="token x"
), patch(
"mcp_server._work_lease_claimant",
return_value={"username": identity, "profile": profile},
), patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=git_state,
), patch(
"mcp_server.issue_duplicate_context_fetcher",
side_effect=lambda h, o, r, auth, issue_number: (
list(open_prs),
[b.get("name") for b in branch_entries if isinstance(b, dict)],
{"status": "not_claimed"},
),
), patch.dict(os.environ, env, clear=True):
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
return mcp_server.gitea_lock_issue(
issue_number=ISSUE,
branch_name=BRANCH,
remote="prgs",
worktree_path=self.worktree,
)
def run_heartbeat(
self, *, task_session_id, identity=IDENTITY, profile=PROFILE, **kwargs
):
env = self._tool_env()
with patch(
"mcp_server._work_lease_claimant",
return_value={"username": identity, "profile": profile},
), patch("mcp_server.get_auth_header", return_value="token x"), patch.dict(
os.environ, env, clear=True
):
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
return mcp_server.gitea_heartbeat_issue_lock(
issue_number=ISSUE,
branch_name=kwargs.pop("branch_name", BRANCH),
task_session_id=task_session_id,
remote="prgs",
worktree_path=kwargs.pop("worktree_path", self.worktree),
**kwargs,
)
def write_legacy_lock(self, *, hours_old: float = 3.0, ttl_hours: float = 4.0):
"""A durable lock in the shape the store wrote before this slice."""
now = datetime.now(timezone.utc)
claimant = {"username": IDENTITY, "profile": PROFILE}
created = now - timedelta(hours=hours_old)
record = {
"issue_number": ISSUE,
"branch_name": BRANCH,
"remote": "prgs",
"org": ORG,
"repo": REPO,
"worktree_path": self.worktree,
"session_pid": os.getpid(),
"pid": os.getpid(),
"lock_generation": 1,
"work_lease": {
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
"issue_number": ISSUE,
"pr_number": None,
"branch": BRANCH,
"worktree_path": self.worktree,
"claimant": claimant,
"created_at": _ts(created),
# The legacy signature: never advanced past creation.
"last_heartbeat_at": _ts(created),
"expires_at": _ts(created + timedelta(hours=ttl_hours)),
},
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue", claimant=claimant
),
}
path = self._lock_path()
record["lock_file_path"] = path
issue_lock_store.save_lock_file(path, record)
return record
class TestLockIssueMintsTheLifecycle(_HeartbeatMcpBase):
"""Durable lock creation and read-back through the real tool."""
def test_native_lock_writes_the_marker_and_a_task_session_id(self):
result = self.run_lock_issue()
self.assertTrue(result["success"], result)
written = issue_lock_store.read_lock_file(result["lock_file_path"])
lease = written["work_lease"]
self.assertEqual(
lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1
)
self.assertTrue(lease["task_session_id"])
self.assertFalse(issue_lock_store.is_legacy_lease(written))
# AC-N1: the ownership key is not the daemon pid, which is recorded
# separately as evidence.
self.assertNotIn(str(written["session_pid"]), lease["task_session_id"])
self.assertEqual(written["session_pid"], os.getpid())
def test_native_lease_uses_the_policy_window_not_four_hours(self):
result = self.run_lock_issue()
lease = result["work_lease"]
created = datetime.fromisoformat(lease["created_at"].replace("Z", "+00:00"))
expires = datetime.fromisoformat(lease["expires_at"].replace("Z", "+00:00"))
policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
self.assertEqual(
(expires - created).total_seconds() / 60.0, policy.initial_ttl_minutes
)
def test_freshness_of_a_new_native_lock_is_live(self):
result = self.run_lock_issue()
self.assertEqual(
result["lock_freshness"]["status"], issue_lock_store.STATUS_LIVE
)
self.assertTrue(result["lock_freshness"]["live"])
class TestHeartbeatThroughTheTool(_HeartbeatMcpBase):
def _lock_and_session(self):
result = self.run_lock_issue()
self.assertTrue(result["success"], result)
return result, result["work_lease"]["task_session_id"]
def test_heartbeat_slides_the_lease_and_advances_the_generation(self):
locked, session = self._lock_and_session()
before = issue_lock_store.read_lock_file(locked["lock_file_path"])
beat = self.run_heartbeat(task_session_id=session)
self.assertTrue(beat["success"], beat)
self.assertEqual(beat["operation"], "heartbeat")
after = issue_lock_store.read_lock_file(locked["lock_file_path"])
self.assertGreater(
issue_lock_store.lock_generation(after),
issue_lock_store.lock_generation(before),
)
self.assertGreaterEqual(
after["work_lease"]["expires_at"], before["work_lease"]["expires_at"]
)
self.assertEqual(after["work_lease"]["heartbeat_count"], 2)
def test_heartbeat_evidence_survives_the_downstream_mutation_gate(self):
"""The #499 F2 lesson, applied.
A sanction that is computed and then discarded downstream is worthless.
After a heartbeat the lock must still satisfy the gate every author
mutation runs through.
"""
locked, session = self._lock_and_session()
self.run_heartbeat(task_session_id=session)
written = issue_lock_store.read_lock_file(locked["lock_file_path"])
verdict = issue_lock_store.verify_lock_for_mutation(
written,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.worktree,
)
self.assertTrue(verdict["proven"], verdict)
self.assertFalse(verdict["block"])
def _duplicate_gate(self, *, open_prs, branches):
env = self._tool_env()
with patch("mcp_server.get_auth_header", return_value="token x"), patch(
"mcp_server.issue_duplicate_context_fetcher",
side_effect=lambda h, o, r, auth, issue_number: (
list(open_prs),
list(branches),
{"status": "not_claimed"},
),
), patch.dict(os.environ, env, clear=True):
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
return mcp_server.gitea_assess_work_issue_duplicate(
issue_number=ISSUE, branch_name=BRANCH, remote="prgs"
)
def test_heartbeat_does_not_change_the_duplicate_gate_verdict(self):
"""The gate must be invariant under heartbeating.
The point is not that the gate passes — with a linked open PR at the
lock phase it correctly blocks (#400), heartbeat or not. The property
that matters is that sliding a lease neither loosens the gate nor
corrupts the lock state it reads: the verdict before and after a
heartbeat must be identical, for both the clear and the blocking shape.
"""
_, session = self._lock_and_session()
linked = [{"number": 4242, "head": {"ref": BRANCH, "sha": self.head_sha}}]
clear_before = self._duplicate_gate(open_prs=[], branches=[])
blocked_before = self._duplicate_gate(open_prs=linked, branches=[BRANCH])
self.assertTrue(self.run_heartbeat(task_session_id=session)["success"])
clear_after = self._duplicate_gate(open_prs=[], branches=[])
blocked_after = self._duplicate_gate(open_prs=linked, branches=[BRANCH])
self.assertEqual(clear_before["outcome"], clear_after["outcome"])
self.assertFalse(clear_after["block"])
self.assertEqual(blocked_before["outcome"], blocked_after["outcome"])
self.assertTrue(blocked_after["block"])
self.assertEqual(blocked_after["linked_open_pr"], 4242)
def test_foreign_session_id_is_refused_through_the_tool(self):
self._lock_and_session()
beat = self.run_heartbeat(task_session_id="author_issue_work-ffffffffffffffff")
self.assertFalse(beat["success"])
self.assertIn("task_session_id does not match", " ".join(beat["reasons"]))
def test_stale_generation_is_refused_through_the_tool(self):
locked, session = self._lock_and_session()
current = issue_lock_store.lock_generation(
issue_lock_store.read_lock_file(locked["lock_file_path"])
)
beat = self.run_heartbeat(
task_session_id=session, expected_generation=current + 5
)
self.assertFalse(beat["success"])
self.assertIn("generation changed", beat["reasons"][0])
def test_foreign_claimant_is_refused_through_the_tool(self):
_, session = self._lock_and_session()
beat = self.run_heartbeat(task_session_id=session, identity="someone-else")
self.assertFalse(beat["success"])
def test_heartbeat_cannot_acquire_a_missing_lock(self):
beat = self.run_heartbeat(task_session_id="author_issue_work-000000000000")
self.assertFalse(beat["success"])
self.assertIn("no durable lock", beat["reasons"][0])
def test_alive_pid_alone_does_not_keep_a_lease_live_through_the_tool(self):
"""PID-only refusal, end to end.
The recorded pid is this live process. The lock is aged past its grace
with no heartbeat, so the tool must refuse to slide it and the durable
record must classify as a missed heartbeat rather than as live.
"""
locked, session = self._lock_and_session()
record = issue_lock_store.read_lock_file(locked["lock_file_path"])
record["work_lease"]["last_heartbeat_at"] = _ts(
datetime.now(timezone.utc) - timedelta(minutes=30)
)
record["work_lease"]["expires_at"] = _ts(
datetime.now(timezone.utc) + timedelta(hours=2)
)
issue_lock_store.save_lock_file(locked["lock_file_path"], record)
self.assertTrue(issue_lock_store.is_process_alive(record["session_pid"]))
fresh = issue_lock_store.assess_lock_freshness(record)
self.assertEqual(
fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
)
self.assertTrue(fresh["pid_alive"])
beat = self.run_heartbeat(task_session_id=session)
self.assertFalse(beat["success"])
self.assertIn("reclaimed", " ".join(beat["reasons"]))
class TestLegacyLocksThroughTheTool(_HeartbeatMcpBase):
"""AC-N8 end to end: protected on deployment, and rebindable."""
def test_legacy_lock_stays_protected_after_deployment(self):
record = self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
fresh = issue_lock_store.assess_lock_freshness(record)
self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
self.assertTrue(fresh["legacy_lease"])
self.assertTrue(fresh["legacy_expiry_preserved"])
# It had never heartbeated, so under the new grace alone it would be
# long gone; the preserved absolute expiry is what protects it.
self.assertEqual(
record["work_lease"]["created_at"],
record["work_lease"]["last_heartbeat_at"],
)
def test_tool_rebinds_a_legacy_lock_and_mints_a_first_heartbeat(self):
self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
result = self.run_heartbeat(task_session_id=None)
self.assertTrue(result["success"], result)
self.assertEqual(result["operation"], "legacy_rebind")
self.assertTrue(result["task_session_id"])
written = issue_lock_store.read_lock_file(self._lock_path())
lease = written["work_lease"]
self.assertEqual(
lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1
)
self.assertEqual(lease["heartbeat_count"], 1)
self.assertNotEqual(
lease["created_at"],
written["legacy_rebind"]["legacy_origin"]["created_at"],
)
self.assertFalse(issue_lock_store.is_legacy_lease(written))
def test_rebound_lock_then_heartbeats_through_the_tool(self):
self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
rebound = self.run_heartbeat(task_session_id=None)
beat = self.run_heartbeat(task_session_id=rebound["task_session_id"])
self.assertTrue(beat["success"], beat)
self.assertEqual(beat["operation"], "heartbeat")
self.assertEqual(beat["heartbeat_count"], 2)
def test_rebind_refuses_a_foreign_owner_through_the_tool(self):
self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
result = self.run_heartbeat(task_session_id=None, identity="someone-else")
self.assertFalse(result["success"])
self.assertEqual(result["operation"], "legacy_rebind")
if __name__ == "__main__":
unittest.main()
+594
View File
@@ -0,0 +1,594 @@
"""Central lease policy and load-bearing heartbeat freshness (#790 Slice A).
Before this slice, ``issue_lock_store.assess_lock_freshness`` parsed
``last_heartbeat_at`` and then never consulted it: liveness was decided by an
absolute four-hour ``expires_at`` and by PID liveness. Because the recorded PID
is the long-lived MCP daemon rather than the authoring task, an abandoned claim
stayed "live" for the full four hours, and a claim whose work had already landed
blocked reconciliation for just as long (Issue #787 / PR #789, and again Issue
#760 / PR #791).
These tests pin the corrected semantics, including the two asymmetries that are
easy to lose in a refactor:
* an **alive** PID must never make anything live (AC-N2), while
* a **dead** PID must still mark a lease stale, because #753 dead-session
recovery keys on exactly that classification.
Durable-state helpers here write real lock files through the real flock path;
they are not mocks of the store.
"""
from __future__ import annotations
import os
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import issue_lock_store # noqa: E402
import lease_policy # noqa: E402
import pr_work_lease # noqa: E402
import reviewer_pr_lease # noqa: E402
ISSUE = 9790
BRANCH = f"fix/issue-{ISSUE}-heartbeat"
IDENTITY = "example-user"
PROFILE = "test-author-prgs"
ORG = "Example-Org"
REPO = "Example-Repo"
REMOTE = "prgs"
DEAD_PID = 2**22 # far above any live pid on a test host
def _ts(moment: datetime) -> str:
return (
moment.astimezone(timezone.utc)
.replace(microsecond=0)
.isoformat()
.replace("+00:00", "Z")
)
class _LockFixture(unittest.TestCase):
def setUp(self):
self.lock_dir = tempfile.TemporaryDirectory()
self.addCleanup(self.lock_dir.cleanup)
self.now = datetime.now(timezone.utc)
self.worktree = os.path.realpath(tempfile.mkdtemp(prefix="issue790-"))
self.addCleanup(patch.stopall)
def _path(self):
return issue_lock_store.lock_file_path(
remote=REMOTE,
org=ORG,
repo=REPO,
issue_number=ISSUE,
lock_dir=self.lock_dir.name,
)
def write_lock(
self,
*,
lifecycle: str | None = lease_policy.LIFECYCLE_HEARTBEAT_V1,
created_delta: timedelta = timedelta(minutes=1),
heartbeat_delta: timedelta = timedelta(minutes=1),
expires_delta: timedelta = timedelta(minutes=9),
pid: int | None = None,
task_session_id: str | None = "author_issue_work-aaaabbbbccccdddd",
generation: int = 1,
identity: str = IDENTITY,
profile: str = PROFILE,
branch: str = BRANCH,
worktree: str | None = None,
) -> dict:
"""Write a real durable lock and return the record.
Deltas are relative to ``self.now``; ``expires_delta`` is added, the
others subtracted, so "in the past" reads naturally at each call site.
"""
lease: dict = {
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
"issue_number": ISSUE,
"pr_number": None,
"branch": branch,
"worktree_path": worktree or self.worktree,
"claimant": {"username": identity, "profile": profile},
"created_at": _ts(self.now - created_delta),
"last_heartbeat_at": _ts(self.now - heartbeat_delta),
"expires_at": _ts(self.now + expires_delta),
}
if lifecycle is not None:
lease["lifecycle_version"] = lifecycle
if task_session_id is not None:
lease["task_session_id"] = task_session_id
pid_value = os.getpid() if pid is None else pid
record = {
"issue_number": ISSUE,
"branch_name": branch,
"remote": REMOTE,
"org": ORG,
"repo": REPO,
"worktree_path": worktree or self.worktree,
"session_pid": pid_value,
"pid": pid_value,
"lock_generation": generation,
"work_lease": lease,
}
path = self._path()
record["lock_file_path"] = path
issue_lock_store.save_lock_file(path, record)
return record
class TestPolicyIsTheSingleSource(unittest.TestCase):
"""AC-N7: one authoritative configuration source for every duration."""
def test_author_policy_carries_the_agreed_values(self):
policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
self.assertEqual(policy.initial_ttl_minutes, 10.0)
self.assertEqual(policy.heartbeat_cadence_minutes, 2.0)
self.assertEqual(policy.stale_warning_minutes, 5.0)
self.assertEqual(policy.missed_heartbeat_grace_minutes, 10.0)
self.assertEqual(policy.absolute_cap_hours, 8.0)
self.assertEqual(policy.recovery_grace_minutes, 10.0)
self.assertEqual(policy.terminal_race_drain_minutes, 2.0)
self.assertTrue(policy.terminal_retirement_eligible)
self.assertTrue(policy.heartbeat_lifecycle_active)
def test_the_four_hour_author_ttl_literal_is_gone(self):
"""The duplicated literal AC-N7 exists to remove."""
self.assertFalse(hasattr(issue_lock_store, "WORK_LEASE_TTL_HOURS"))
import gitea_mcp_server
self.assertFalse(hasattr(gitea_mcp_server, "WORK_LEASE_TTL_HOURS"))
def test_declared_reviewer_values_match_the_module_still_using_them(self):
"""Slice A declares reviewer/merger numbers without rewiring them.
Recording a value in two places is only safe if drift is detectable, so
this asserts the declaration still equals the constants #747 owns. When
Slice C migrates those call sites, this test becomes the proof the
migration changed nothing.
"""
policy = lease_policy.policy_for(lease_policy.TASK_CLASS_REVIEWER_PR)
self.assertEqual(
policy.initial_ttl_minutes, float(reviewer_pr_lease.LEASE_TTL_MINUTES)
)
self.assertEqual(
policy.stale_warning_minutes,
float(reviewer_pr_lease.STALE_WARNING_MINUTES),
)
self.assertFalse(policy.heartbeat_lifecycle_active)
def test_declared_conflict_fix_value_matches_its_module(self):
policy = lease_policy.policy_for(lease_policy.TASK_CLASS_CONFLICT_FIX)
self.assertEqual(
policy.initial_ttl_minutes,
float(pr_work_lease.DEFAULT_CONFLICT_FIX_TTL_MINUTES),
)
self.assertFalse(policy.heartbeat_lifecycle_active)
def test_environment_override_applies(self):
var = lease_policy.env_var_name(
lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK, "initial_ttl_minutes"
)
with patch.dict(os.environ, {var: "7"}):
self.assertEqual(
lease_policy.policy_for(
lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK
).initial_ttl_minutes,
7.0,
)
def test_unusable_override_falls_back_instead_of_minting_a_zero_lease(self):
"""A typo must not make every claim instantly reclaimable."""
var = lease_policy.env_var_name(
lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK, "initial_ttl_minutes"
)
for bad in ("0", "-5", "not-a-number", " "):
with self.subTest(value=bad), patch.dict(os.environ, {var: bad}):
self.assertEqual(
lease_policy.policy_for(
lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK
).initial_ttl_minutes,
10.0,
)
def test_unknown_task_class_does_not_raise(self):
policy = lease_policy.policy_for("something-new")
self.assertEqual(policy.task_class, lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
class TestLifecycleDiscrimination(_LockFixture):
"""AC-N8: the marker, never a timestamp, decides legacy vs heartbeat."""
def test_missing_marker_reads_as_legacy(self):
record = self.write_lock(lifecycle=None)
self.assertTrue(issue_lock_store.is_legacy_lease(record))
self.assertEqual(
issue_lock_store.lease_lifecycle_version(record),
lease_policy.LIFECYCLE_LEGACY,
)
def test_marker_present_reads_as_heartbeat_lifecycle(self):
record = self.write_lock()
self.assertFalse(issue_lock_store.is_legacy_lease(record))
def test_equal_created_and_heartbeat_never_implies_a_fresh_heartbeat(self):
"""The exact inversion AC-N8 forbids.
A legacy lock has ``last_heartbeat_at == created_at`` forever because
nothing ever advanced it. Reading that equality as "recently
heartbeated" would classify every never-heartbeated lock as fresh.
"""
legacy = self.write_lock(
lifecycle=None,
created_delta=timedelta(hours=3),
heartbeat_delta=timedelta(hours=3),
)
lease = legacy["work_lease"]
self.assertEqual(lease["created_at"], lease["last_heartbeat_at"])
self.assertTrue(issue_lock_store.is_legacy_lease(legacy))
# A brand-new heartbeat lease has them equal too, so the equality
# carries no information in either direction.
fresh = self.write_lock(
created_delta=timedelta(seconds=0), heartbeat_delta=timedelta(seconds=0)
)
self.assertEqual(
fresh["work_lease"]["created_at"],
fresh["work_lease"]["last_heartbeat_at"],
)
self.assertFalse(issue_lock_store.is_legacy_lease(fresh))
def test_minted_session_id_contains_no_pid(self):
"""AC-N1: the ownership key must not be derived from the daemon pid."""
minted = issue_lock_store.mint_task_session_id()
self.assertNotIn(str(os.getpid()), minted)
self.assertNotEqual(minted, issue_lock_store.mint_task_session_id())
class TestFreshnessIsHeartbeatDriven(_LockFixture):
"""AC-N2 and the new bands."""
def test_fresh_heartbeat_is_live(self):
record = self.write_lock(heartbeat_delta=timedelta(minutes=1))
fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
self.assertTrue(fresh["live"])
self.assertFalse(fresh["heartbeat_warning"])
def test_heartbeat_past_warning_is_still_live_but_flagged(self):
record = self.write_lock(heartbeat_delta=timedelta(minutes=6))
fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
self.assertTrue(fresh["heartbeat_warning"])
def test_missed_heartbeat_past_grace_is_classified_explicitly(self):
record = self.write_lock(
heartbeat_delta=timedelta(minutes=11),
expires_delta=timedelta(minutes=30),
)
fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
self.assertEqual(
fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
)
self.assertFalse(fresh["live"])
self.assertTrue(fresh["stale"])
def test_alive_pid_never_establishes_freshness(self):
"""The defect in one assertion.
The recorded PID is this very process, so it is unambiguously alive —
and the lease is still not live, because the task stopped heartbeating.
"""
record = self.write_lock(
pid=os.getpid(),
heartbeat_delta=timedelta(hours=4),
expires_delta=timedelta(hours=4),
)
fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
self.assertTrue(fresh["pid_alive"])
self.assertFalse(fresh["live"])
self.assertEqual(
fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
)
def test_dead_pid_still_marks_stale_for_issue_753(self):
"""The opposite asymmetry: dead-PID corroboration is preserved."""
record = self.write_lock(pid=DEAD_PID, heartbeat_delta=timedelta(minutes=1))
fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
self.assertEqual(fresh["status"], issue_lock_store.STATUS_STALE)
self.assertFalse(fresh["live"])
self.assertIn("not alive", fresh["reason"])
def test_absolute_cap_requires_readoption(self):
record = self.write_lock(
created_delta=timedelta(hours=9), heartbeat_delta=timedelta(minutes=1)
)
fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
self.assertEqual(fresh["status"], issue_lock_store.STATUS_STALE_ABSOLUTE_CAP)
self.assertIn("re-adoption", fresh["reason"])
def test_heartbeat_lifecycle_without_a_heartbeat_fails_closed(self):
record = self.write_lock()
del record["work_lease"]["last_heartbeat_at"]
issue_lock_store.save_lock_file(self._path(), record)
fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
self.assertEqual(
fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
)
self.assertIn("fail closed", fresh["reason"])
def test_absent_lock(self):
fresh = issue_lock_store.assess_lock_freshness(None)
self.assertEqual(fresh["status"], issue_lock_store.STATUS_ABSENT)
self.assertFalse(fresh["stale"])
class TestLegacyLocksStayProtected(_LockFixture):
"""AC-N8: deployment must not retroactively shorten an existing claim."""
def test_legacy_lock_with_a_stale_heartbeat_remains_live(self):
"""The deployment-safety case.
A four-hour legacy lease minted three hours ago has not heartbeated
once. Under the new grace it would be long gone; under its preserved
absolute expiry it is still live, and must stay that way.
"""
record = self.write_lock(
lifecycle=None,
created_delta=timedelta(hours=3),
heartbeat_delta=timedelta(hours=3),
expires_delta=timedelta(hours=1),
)
fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
self.assertTrue(fresh["live"])
self.assertTrue(fresh["legacy_lease"])
self.assertTrue(fresh["legacy_expiry_preserved"])
def test_legacy_lock_past_its_absolute_expiry_is_expired_as_before(self):
record = self.write_lock(
lifecycle=None,
created_delta=timedelta(hours=5),
heartbeat_delta=timedelta(hours=5),
expires_delta=timedelta(hours=-1),
)
fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
self.assertEqual(fresh["status"], issue_lock_store.STATUS_EXPIRED)
def test_legacy_lock_is_never_reclaimed_by_the_heartbeat_band(self):
record = self.write_lock(
lifecycle=None,
created_delta=timedelta(hours=3),
heartbeat_delta=timedelta(hours=3),
expires_delta=timedelta(hours=1),
)
reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
self.assertFalse(reclaim["reclaim_allowed"])
class TestReclaimAfterMissedHeartbeat(_LockFixture):
def test_missed_heartbeat_makes_ownership_reclaimable(self):
record = self.write_lock(
pid=os.getpid(),
heartbeat_delta=timedelta(minutes=15),
expires_delta=timedelta(hours=3),
)
reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
self.assertTrue(reclaim["reclaim_allowed"])
self.assertIn("stale_missed_heartbeat", reclaim["reasons"][0])
def test_live_lease_is_never_reclaimable(self):
record = self.write_lock(heartbeat_delta=timedelta(minutes=1))
reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
self.assertFalse(reclaim["reclaim_allowed"])
def test_dead_pid_reclaim_path_is_unchanged(self):
"""#753 must keep working through its original conditions."""
record = self.write_lock(pid=DEAD_PID, heartbeat_delta=timedelta(minutes=1))
reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
self.assertTrue(reclaim["reclaim_allowed"])
self.assertTrue(reclaim["owner_pid_dead"])
class TestHeartbeatWriter(_LockFixture):
"""A4: flock + CAS + exact verification, and no revival path."""
def _heartbeat(self, **kwargs):
params = {
"remote": REMOTE,
"org": ORG,
"repo": REPO,
"issue_number": ISSUE,
"branch_name": BRANCH,
"worktree_path": self.worktree,
"identity": IDENTITY,
"profile": PROFILE,
"task_session_id": "author_issue_work-aaaabbbbccccdddd",
"lock_dir": self.lock_dir.name,
"now": self.now,
}
params.update(kwargs)
return issue_lock_store.heartbeat_session_lock(**params)
def test_heartbeat_slides_expiry_and_advances_generation(self):
self.write_lock(heartbeat_delta=timedelta(minutes=4), generation=5)
result = self._heartbeat()
self.assertTrue(result["success"], result)
self.assertEqual(result["prior_generation"], 5)
self.assertEqual(result["lock_generation"], 6)
self.assertEqual(result["heartbeat_count"], 1)
self.assertEqual(result["last_heartbeat_at"], _ts(self.now))
self.assertEqual(result["expires_at"], _ts(self.now + timedelta(minutes=10)))
self.assertTrue(result["freshness"]["live"])
def test_heartbeat_is_durable_and_repeatable(self):
self.write_lock(heartbeat_delta=timedelta(minutes=4))
self._heartbeat()
second = self._heartbeat(now=self.now + timedelta(minutes=1))
self.assertTrue(second["success"], second)
self.assertEqual(second["heartbeat_count"], 2)
written = issue_lock_store.read_lock_file(self._path())
self.assertEqual(written["work_lease"]["heartbeat_count"], 2)
def test_stale_generation_is_refused(self):
self.write_lock(generation=5)
result = self._heartbeat(expected_generation=4)
self.assertFalse(result["success"])
self.assertIn("generation changed", result["reasons"][0])
def test_foreign_session_is_refused(self):
self.write_lock()
result = self._heartbeat(task_session_id="author_issue_work-ffffffffffffffff")
self.assertFalse(result["success"])
self.assertIn("task_session_id does not match", " ".join(result["reasons"]))
def test_missing_session_id_is_refused(self):
self.write_lock()
result = self._heartbeat(task_session_id="")
self.assertFalse(result["success"])
def test_foreign_claimant_is_refused(self):
self.write_lock()
for field, value in (
("identity", "someone-else"),
("profile", "other-profile"),
):
with self.subTest(field=field):
result = self._heartbeat(**{field: value})
self.assertFalse(result["success"])
def test_branch_and_worktree_mismatch_are_refused(self):
self.write_lock()
wrong_branch = self._heartbeat(branch_name=f"fix/issue-{ISSUE}-other")
self.assertFalse(wrong_branch["success"])
wrong_worktree = self._heartbeat(worktree_path="/tmp/not-the-worktree")
self.assertFalse(wrong_worktree["success"])
def test_lapsed_lease_cannot_be_heartbeated_back_to_life(self):
"""No revival path (A4).
A session that stopped proving liveness must reclaim under a fresh
generation, not restore ownership retroactively.
"""
self.write_lock(
heartbeat_delta=timedelta(minutes=30), expires_delta=timedelta(hours=1)
)
result = self._heartbeat()
self.assertFalse(result["success"])
self.assertIn("reclaimed", " ".join(result["reasons"]))
def test_absent_lock_cannot_be_created_by_heartbeat(self):
result = self._heartbeat()
self.assertFalse(result["success"])
self.assertIn("no durable lock", result["reasons"][0])
def test_legacy_lock_is_refused_until_rebound(self):
self.write_lock(lifecycle=None)
result = self._heartbeat()
self.assertFalse(result["success"])
self.assertTrue(result["legacy_lease"])
self.assertIn("rebound", " ".join(result["reasons"]))
class TestLegacyRebind(_LockFixture):
"""AC-N8 exit route: canonical exact-owner rebinding."""
def _rebind(self, **kwargs):
params = {
"remote": REMOTE,
"org": ORG,
"repo": REPO,
"issue_number": ISSUE,
"branch_name": BRANCH,
"worktree_path": self.worktree,
"identity": IDENTITY,
"profile": PROFILE,
"lock_dir": self.lock_dir.name,
"now": self.now,
}
params.update(kwargs)
return issue_lock_store.rebind_legacy_lock(**params)
def test_rebind_mints_a_session_and_a_genuine_first_heartbeat(self):
self.write_lock(
lifecycle=None,
created_delta=timedelta(hours=3),
heartbeat_delta=timedelta(hours=3),
expires_delta=timedelta(hours=1),
generation=2,
)
result = self._rebind()
self.assertTrue(result["success"], result)
self.assertTrue(result["task_session_id"])
self.assertEqual(result["lock_generation"], 3)
written = issue_lock_store.read_lock_file(self._path())
lease = written["work_lease"]
self.assertEqual(
lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1
)
self.assertEqual(lease["last_heartbeat_at"], _ts(self.now))
self.assertEqual(lease["expires_at"], _ts(self.now + timedelta(minutes=10)))
self.assertFalse(issue_lock_store.is_legacy_lease(written))
# The original claim is preserved for audit rather than overwritten.
origin = written["legacy_rebind"]["legacy_origin"]
self.assertTrue(origin["created_at"])
self.assertEqual(origin["lifecycle"], lease_policy.LIFECYCLE_LEGACY)
def test_rebound_lock_can_then_heartbeat(self):
self.write_lock(
lifecycle=None,
created_delta=timedelta(hours=3),
heartbeat_delta=timedelta(hours=3),
expires_delta=timedelta(hours=1),
)
rebound = self._rebind()
beat = issue_lock_store.heartbeat_session_lock(
remote=REMOTE,
org=ORG,
repo=REPO,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.worktree,
identity=IDENTITY,
profile=PROFILE,
task_session_id=rebound["task_session_id"],
lock_dir=self.lock_dir.name,
now=self.now + timedelta(minutes=1),
)
self.assertTrue(beat["success"], beat)
def test_rebind_refuses_a_foreign_owner(self):
self.write_lock(lifecycle=None, expires_delta=timedelta(hours=1))
result = self._rebind(identity="someone-else")
self.assertFalse(result["success"])
def test_rebind_refuses_a_lock_already_on_the_lifecycle(self):
self.write_lock()
result = self._rebind()
self.assertFalse(result["success"])
self.assertFalse(result["legacy_lease"])
def test_rebind_is_not_a_recovery_path_for_a_lapsed_legacy_lease(self):
"""An expired legacy lease belongs to #760 renewal or #601 reclaim."""
self.write_lock(
lifecycle=None,
created_delta=timedelta(hours=5),
heartbeat_delta=timedelta(hours=5),
expires_delta=timedelta(hours=-1),
)
result = self._rebind()
self.assertFalse(result["success"])
self.assertIn("not a recovery path", " ".join(result["reasons"]))
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,193 @@
"""Conflict gate honors heartbeat-lifecycle non-live reclaim bands (#790 review #502/#516).
``assess_expired_lock_reclaim`` already permits reclaim for the heartbeat-lifecycle
stale bands (``stale_missed_heartbeat``, ``stale_absolute_cap``) without a dead PID.
But ``assess_same_issue_lease_conflict`` used to enter that reclaim branch only under
``is_lease_expired`` (``expires_at <= now``). For a heartbeat-lifecycle lease that is
non-live yet whose ``expires_at`` is still in the future, the acquisition gate fell
through to the foreign "already has an active lease" block and never consulted the
reclaim assessor — so the load-bearing heartbeat was not load-bearing for foreign
reclaim, the exact abandonment scenario #790 exists to fix.
Two future-``expires_at`` non-live shapes are reachable:
* ``stale_absolute_cap`` — a session that keeps heartbeating past the 8h absolute cap
has ``expires_at = last_heartbeat + TTL`` in the future (default policy).
* ``stale_missed_heartbeat`` — under an independent TTL>grace policy the heartbeat
grace lapses while ``expires_at`` is still ahead.
These tests pin: both reclaim from a foreign acquirer; a live lease still blocks a
foreign acquirer; the same owner may reclaim its own abandoned heartbeat lease; and a
legacy (pre-lifecycle) lock keeps its absolute-``expires_at`` clock — a non-expired
legacy lock with a dead PID is *not* reclaimable through this path.
"""
from __future__ import annotations
import os
import sys
import unittest
from datetime import datetime, timedelta, timezone
from unittest import mock
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import issue_lock_store as ils # noqa: E402
import lease_policy # noqa: E402
ISSUE = 790
OWNER_BRANCH = f"fix/issue-{ISSUE}-slice-a-heartbeat-policy"
OWNER_WORKTREE = "/tmp/wt-790-owner"
FOREIGN_BRANCH = f"fix/issue-{ISSUE}-foreign-attempt"
FOREIGN_WORKTREE = "/tmp/wt-790-foreign"
def _ts(moment: datetime) -> str:
return (
moment.astimezone(timezone.utc)
.replace(microsecond=0)
.isoformat()
.replace("+00:00", "Z")
)
class _ConflictGateBase(unittest.TestCase):
def setUp(self):
self.now = datetime(2026, 7, 23, 12, 0, 0, tzinfo=timezone.utc)
def _lock(
self,
*,
lifecycle: str | None,
created_ago: timedelta,
heartbeat_ago: timedelta,
expires_in: timedelta,
pid: int,
) -> dict:
lease: dict = {
"operation_type": ils.AUTHOR_ISSUE_WORK_LEASE,
"issue_number": ISSUE,
"branch": OWNER_BRANCH,
"worktree_path": OWNER_WORKTREE,
"created_at": _ts(self.now - created_ago),
"last_heartbeat_at": _ts(self.now - heartbeat_ago),
"expires_at": _ts(self.now + expires_in),
}
if lifecycle is not None:
lease["lifecycle_version"] = lifecycle
lease["task_session_id"] = "author_issue_work-deadbeefdeadbeef"
return {
"issue_number": ISSUE,
"branch_name": OWNER_BRANCH,
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"worktree_path": OWNER_WORKTREE,
"session_pid": pid,
"pid": pid,
"work_lease": lease,
}
def _foreign_conflict(self, existing: dict) -> str | None:
return ils.assess_same_issue_lease_conflict(
existing,
issue_number=ISSUE,
branch_name=FOREIGN_BRANCH,
worktree_path=FOREIGN_WORKTREE,
now=self.now,
)
def _same_owner_conflict(self, existing: dict) -> str | None:
return ils.assess_same_issue_lease_conflict(
existing,
issue_number=ISSUE,
branch_name=OWNER_BRANCH,
worktree_path=OWNER_WORKTREE,
now=self.now,
)
class TestHeartbeatNonLiveFutureExpiresReclaim(_ConflictGateBase):
def test_stale_absolute_cap_future_expires_allows_foreign_reclaim(self):
# Created >8h ago, heartbeated one minute ago, expires 9 min in the FUTURE,
# owner PID alive: freshness = stale_absolute_cap, live=False, not expired.
existing = self._lock(
lifecycle=lease_policy.LIFECYCLE_HEARTBEAT_V1,
created_ago=timedelta(hours=9),
heartbeat_ago=timedelta(minutes=1),
expires_in=timedelta(minutes=9),
pid=os.getpid(),
)
freshness = ils.assess_lock_freshness(existing, now=self.now)
self.assertEqual(freshness["status"], ils.STATUS_STALE_ABSOLUTE_CAP)
self.assertFalse(freshness["live"])
self.assertFalse(ils.is_lease_expired(existing, now=self.now))
with mock.patch.object(ils, "is_process_alive", return_value=True):
self.assertIsNone(self._foreign_conflict(existing))
def test_missed_heartbeat_ttl_gt_grace_future_expires_allows_foreign_reclaim(self):
# Heartbeat grace (default 10 min) lapsed 5 min ago, but a TTL>grace policy
# leaves expires_at 20 min in the FUTURE: stale_missed_heartbeat, not expired.
existing = self._lock(
lifecycle=lease_policy.LIFECYCLE_HEARTBEAT_V1,
created_ago=timedelta(minutes=30),
heartbeat_ago=timedelta(minutes=15),
expires_in=timedelta(minutes=20),
pid=os.getpid(),
)
freshness = ils.assess_lock_freshness(existing, now=self.now)
self.assertEqual(freshness["status"], ils.STATUS_STALE_MISSED_HEARTBEAT)
self.assertFalse(freshness["live"])
self.assertFalse(ils.is_lease_expired(existing, now=self.now))
with mock.patch.object(ils, "is_process_alive", return_value=True):
self.assertIsNone(self._foreign_conflict(existing))
def test_same_owner_may_reclaim_its_own_abandoned_heartbeat_lease(self):
existing = self._lock(
lifecycle=lease_policy.LIFECYCLE_HEARTBEAT_V1,
created_ago=timedelta(hours=9),
heartbeat_ago=timedelta(minutes=1),
expires_in=timedelta(minutes=9),
pid=os.getpid(),
)
with mock.patch.object(ils, "is_process_alive", return_value=True):
self.assertIsNone(self._same_owner_conflict(existing))
class TestLiveAndLegacyStillBlockForeign(_ConflictGateBase):
def test_live_heartbeat_lease_still_blocks_foreign(self):
existing = self._lock(
lifecycle=lease_policy.LIFECYCLE_HEARTBEAT_V1,
created_ago=timedelta(minutes=5),
heartbeat_ago=timedelta(minutes=1),
expires_in=timedelta(minutes=9),
pid=os.getpid(),
)
freshness = ils.assess_lock_freshness(existing, now=self.now)
self.assertTrue(freshness["live"])
with mock.patch.object(ils, "is_process_alive", return_value=True):
block = self._foreign_conflict(existing)
self.assertIn("already has an active", block or "")
def test_legacy_lease_future_expires_dead_pid_is_not_reclaimed_here(self):
# AC-N8: a legacy lock keeps its absolute expires_at clock. Non-expired +
# dead PID is non-live, but the widened band excludes legacy, so the
# foreign acquirer is still blocked rather than silently reclaiming.
existing = self._lock(
lifecycle=None,
created_ago=timedelta(hours=9),
heartbeat_ago=timedelta(hours=9),
expires_in=timedelta(hours=2),
pid=4_194_304, # far above any live pid on a test host
)
with mock.patch.object(ils, "is_process_alive", return_value=False):
freshness = ils.assess_lock_freshness(existing, now=self.now)
self.assertTrue(freshness["legacy_lease"])
self.assertFalse(freshness["live"])
self.assertFalse(ils.is_lease_expired(existing, now=self.now))
block = self._foreign_conflict(existing)
self.assertIn("already has an active", block or "")
if __name__ == "__main__":
unittest.main()
@@ -139,6 +139,8 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
"gitea_release_merger_pr_lease", "gitea_release_merger_pr_lease",
"create_branch", "create_branch",
"push_branch", "push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
# #812 AC20: publishing an unpublished local head is author-only for the # #812 AC20: publishing an unpublished local head is author-only for the
# same reason every other push is — it writes a branch to the remote. # same reason every other push is — it writes a branch to the remote.
"publish_unpublished_branch", "publish_unpublished_branch",
+63 -7
View File
@@ -165,7 +165,8 @@ class AnalyticsWebUITest(unittest.TestCase):
self.assertEqual(data["total_events"], 1) self.assertEqual(data["total_events"], 1)
self.assertIn("gemini-3.6-flash", data["by_model"]) self.assertIn("gemini-3.6-flash", data["by_model"])
def test_analytics_ingest_endpoint(self) -> None: def test_analytics_ingest_unauthorized_denied(self) -> None:
"""F2: unauthenticated POST must not write the control-plane DB."""
payload = { payload = {
"remote": "dadeschools", "remote": "dadeschools",
"org": "Scaled-Tech-Consulting", "org": "Scaled-Tech-Consulting",
@@ -180,16 +181,71 @@ class AnalyticsWebUITest(unittest.TestCase):
"metadata": "Review note token=secret456", "metadata": "Review note token=secret456",
} }
response = self.client.post("/api/v1/analytics/usage", json=payload) response = self.client.post("/api/v1/analytics/usage", json=payload)
self.assertEqual(response.status_code, 201) self.assertEqual(response.status_code, 403)
res_json = response.json() res_json = response.json()
self.assertTrue(res_json["ok"]) self.assertFalse(res_json.get("ok", True))
self.assertGreater(res_json["usage_id"], 0) self.assertEqual(res_json.get("error"), "unauthorized")
authorization = res_json.get("authorization") or {}
self.assertFalse(authorization.get("allowed"))
self.assertFalse(authorization.get("execution_enabled"))
# Check that it appears in GET /api/v1/analytics # No new row written
res2 = self.client.get("/api/v1/analytics") res2 = self.client.get("/api/v1/analytics")
self.assertEqual(res2.status_code, 200) self.assertEqual(res2.status_code, 200)
data2 = res2.json() self.assertEqual(res2.json()["total_events"], 1)
self.assertEqual(data2["total_events"], 2)
def test_html_escapes_script_bearing_model_role_stage(self) -> None:
"""F1: stored XSS — dynamic model/role/stage must render escaped."""
xss = '<script>alert(1)</script>'
record_usage(
db_path=self.db_path,
remote="dadeschools",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
role=xss,
model=xss,
stage=xss,
issue_number=999,
status="success",
)
response = self.client.get("/analytics")
self.assertEqual(response.status_code, 200)
# Raw tag must not appear; escaped form must.
self.assertNotIn("<script>alert(1)</script>", response.text)
self.assertIn("&lt;script&gt;alert(1)&lt;/script&gt;", response.text)
def test_load_analytics_coerces_none_scope(self) -> None:
"""F4: None remote/org/repo become empty strings, never None."""
snapshot = load_analytics(db_path=self.db_path, remote=None, org=None, repo=None)
self.assertIsInstance(snapshot.remote, str)
self.assertIsInstance(snapshot.org, str)
self.assertIsInstance(snapshot.repo, str)
self.assertEqual(snapshot.remote, "")
self.assertEqual(snapshot.org, "")
self.assertEqual(snapshot.repo, "")
def test_usage_events_retention_max_rows(self) -> None:
"""F3: record_usage_event enforces USAGE_EVENTS_MAX_ROWS."""
db = control_plane_db.ControlPlaneDB(db_path=self.db_path)
original_max = db.USAGE_EVENTS_MAX_ROWS
try:
db.USAGE_EVENTS_MAX_ROWS = 3
for i in range(5):
db.record_usage_event(
remote="dadeschools",
org="org",
repo="repo",
role="author",
model=f"model-{i}",
stage="test",
)
rows = db.query_usage_events(limit=100)
self.assertLessEqual(len(rows), 3)
# Newest three retained
models = {r["model"] for r in rows}
self.assertEqual(models, {"model-2", "model-3", "model-4"})
finally:
db.USAGE_EVENTS_MAX_ROWS = original_max
if __name__ == "__main__": if __name__ == "__main__":
+502
View File
@@ -0,0 +1,502 @@
"""Sanctioned restart / graceful reload control tests (#642).
Acceptance criteria under test:
1. The sanctioned restart path is implemented behind gates (capability,
confirmation, operator authorization, host hook).
2. Manual ``pkill`` stays forbidden and is classified as contamination.
3. Post-restart mutations require clean health/session proof.
4. Authorized restart preview, unauthorized deny, contamination classification.
5. No entry point exposes a raw kill.
"""
import json
import os
import tempfile
import unittest
import mcp_namespace_health
import runtime_recovery_guard
from task_capability_map import TASK_CAPABILITY_MAP
from webui import console_audit, console_authz, gated_actions, sanctioned_restart
NAMESPACE = "gitea-author"
# An operator-authorized, hook-configured host. Passed explicitly so no test
# depends on (or mutates) the real process environment.
READY_ENV = {
sanctioned_restart.RESTART_HOOK_ENV: "launchd:cc.prgs.gitea-author",
runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV: "ops-ticket-4821",
}
def admin(subject: str = "[email protected]") -> console_authz.Principal:
return console_authz.Principal(
subject=subject,
role=console_authz.ADMIN,
identity_source=console_authz.IDENTITY_ACCESS_PROXY,
authenticated=True,
)
def viewer() -> console_authz.Principal:
return console_authz.Principal(
subject="[email protected]",
role=console_authz.VIEWER,
identity_source=console_authz.IDENTITY_ACCESS_PROXY,
authenticated=True,
)
class TestCapabilityWiring(unittest.TestCase):
"""AC1: authority is declared, not invented by the console."""
def test_actions_resolve_through_the_capability_map(self):
for action_id in (
sanctioned_restart.ACTION_RESTART_NAMESPACE,
sanctioned_restart.ACTION_RELOAD_NAMESPACE,
):
with self.subTest(action=action_id):
action = console_authz.get_action(action_id)
self.assertIsNotNone(action)
self.assertIn(action.task_key, TASK_CAPABILITY_MAP)
self.assertEqual(
action.mcp_permission,
TASK_CAPABILITY_MAP[action.task_key]["permission"],
)
def test_restart_permission_is_not_a_gitea_operation(self):
"""No configured Gitea profile should satisfy a host restart."""
permission = TASK_CAPABILITY_MAP["restart_namespace"]["permission"]
self.assertFalse(permission.startswith("gitea."))
def test_restart_is_destructive_dual_control_break_glass(self):
action = console_authz.get_action(
sanctioned_restart.ACTION_RESTART_NAMESPACE
)
self.assertEqual(action.action_class, console_authz.CLASS_DESTRUCTIVE)
self.assertEqual(action.minimum_role, console_authz.ADMIN)
self.assertTrue(action.dual_control)
self.assertTrue(action.break_glass)
self.assertTrue(action.requires_confirmation)
def test_reload_is_privileged_but_not_destructive(self):
action = console_authz.get_action(
sanctioned_restart.ACTION_RELOAD_NAMESPACE
)
self.assertEqual(action.action_class, console_authz.CLASS_PRIVILEGED)
self.assertTrue(action.requires_confirmation)
class TestPreview(unittest.TestCase):
"""AC4: an authorized preview renders the plan without executing it."""
def test_preview_lists_the_mutation_ledger(self):
preview = sanctioned_restart.build_restart_preview(
NAMESPACE, principal=admin(), env=READY_ENV
)
steps = [entry["step"] for entry in preview["mutation_ledger"]]
self.assertEqual(
steps, ["quiesce", "host_restart_hook", "health_recheck", "audit"]
)
self.assertTrue(preview["scope_valid"])
self.assertTrue(preview["post_restart_verification_required"])
def test_reload_preview_drains_instead_of_restarting(self):
preview = sanctioned_restart.build_restart_preview(
NAMESPACE, sanctioned_restart.MODE_RELOAD,
principal=admin(), env=READY_ENV,
)
steps = [entry["step"] for entry in preview["mutation_ledger"]]
self.assertIn("host_graceful_reload", steps)
self.assertNotIn("host_restart_hook", steps)
def test_preview_never_enables_execution(self):
preview = sanctioned_restart.build_restart_preview(
NAMESPACE, principal=admin(), env=READY_ENV
)
self.assertFalse(preview["execution_enabled"])
self.assertFalse(preview["authorization"]["execution_enabled"])
def test_confirmation_phrase_binds_the_namespace(self):
self.assertTrue(
sanctioned_restart.confirmation_matches(
NAMESPACE, sanctioned_restart.MODE_RESTART,
"restart gitea-author",
)
)
# A phrase typed for one namespace must not authorize another.
self.assertFalse(
sanctioned_restart.confirmation_matches(
"gitea-merger", sanctioned_restart.MODE_RESTART,
"restart gitea-author",
)
)
class TestGates(unittest.TestCase):
"""AC1/AC4: every gate denies with a stable reason code."""
def _assess(self, **kwargs):
params = {
"principal": admin(),
"confirmation": f"restart {NAMESPACE}",
"env": READY_ENV,
}
params.update(kwargs)
namespace = params.pop("namespace", NAMESPACE)
mode = params.pop("mode", sanctioned_restart.MODE_RESTART)
return sanctioned_restart.assess_restart_request(
namespace, mode, **params
)
def test_authorized_confirmed_request_passes_every_gate(self):
result = self._assess()
self.assertTrue(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.ALLOW_HOST_ACTION_REQUIRED
)
def test_passing_every_gate_is_not_an_execution_grant(self):
"""An allowed request still never lets the console touch the process."""
result = self._assess()
self.assertTrue(result["allowed"])
self.assertFalse(result["execution_enabled"])
self.assertFalse(result["console_executes"])
def test_unauthorized_principal_is_denied(self):
result = self._assess(principal=viewer())
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
)
def test_anonymous_principal_is_denied(self):
result = self._assess(principal=None)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
)
def test_missing_confirmation_is_denied(self):
result = self._assess(confirmation=None)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_CONFIRMATION_MISSING
)
def test_confirmation_for_another_namespace_is_denied(self):
result = self._assess(confirmation="restart gitea-merger")
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_CONFIRMATION_MISMATCH
)
def test_missing_operator_authorization_is_denied(self):
env = {sanctioned_restart.RESTART_HOOK_ENV: "launchd:cc.prgs.author"}
result = self._assess(env=env)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"],
sanctioned_restart.DENY_OPERATOR_AUTHORIZATION,
)
def test_missing_host_hook_is_denied_without_kill_fallback(self):
env = {
runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV: "ops-ticket-1",
}
result = self._assess(env=env)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_HOOK_NOT_CONFIGURED
)
def test_fleet_scope_is_refused(self):
for scope in ("all", "*", "fleet"):
with self.subTest(scope=scope):
result = self._assess(
namespace=scope, confirmation=f"restart {scope}"
)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_FLEET_SCOPE
)
def test_unknown_namespace_is_refused(self):
result = self._assess(
namespace="gitea-nope", confirmation="restart gitea-nope"
)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_UNKNOWN_NAMESPACE
)
def test_unknown_mode_is_refused(self):
result = self._assess(mode="obliterate")
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_UNKNOWN_MODE
)
def test_live_contamination_marker_blocks_restart(self):
marker = runtime_recovery_guard.build_contamination_record(
reason_class=runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL,
command_redacted="pkill -f mcp_server.py",
)
result = self._assess(contamination_marker=marker)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_CONTAMINATED_RUNTIME
)
def test_reconciler_cleared_marker_no_longer_blocks(self):
marker = runtime_recovery_guard.build_contamination_record(
reason_class=runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL,
command_redacted="pkill -f mcp_server.py",
)
marker = dict(marker, cleared_by_reconciler=True)
result = self._assess(contamination_marker=marker)
self.assertTrue(result["allowed"])
class TestExecutionNeverKills(unittest.TestCase):
"""AC5: no path exposes or runs a raw process kill."""
def test_authorized_execution_defers_to_the_host_supervisor(self):
result = sanctioned_restart.execute_restart(
NAMESPACE,
principal=admin(),
confirmation=f"restart {NAMESPACE}",
env=READY_ENV,
)
self.assertTrue(result["allowed"])
self.assertFalse(result["success"])
self.assertFalse(result["process_kill_executed"])
self.assertEqual(
result["outcome"], sanctioned_restart.ALLOW_HOST_ACTION_REQUIRED
)
def test_denied_execution_reports_the_refusing_gate(self):
result = sanctioned_restart.execute_restart(
NAMESPACE, principal=viewer(), confirmation=f"restart {NAMESPACE}",
env=READY_ENV,
)
self.assertFalse(result["allowed"])
self.assertEqual(
result["outcome"], sanctioned_restart.DENY_UNAUTHORIZED
)
self.assertFalse(result["process_kill_executed"])
def test_module_never_spawns_a_process(self):
path = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"webui", "sanctioned_restart.py",
)
with open(path, encoding="utf-8") as handle:
source = handle.read()
for forbidden in (
"import subprocess", "import signal", "os.kill", "os.system",
"popen",
):
with self.subTest(forbidden=forbidden):
self.assertNotIn(forbidden, source.lower())
def test_no_surface_returns_a_kill_command(self):
payloads = [
sanctioned_restart.build_restart_preview(
NAMESPACE, principal=admin(), env=READY_ENV
),
sanctioned_restart.restart_policy(),
sanctioned_restart.execute_restart(
NAMESPACE, principal=admin(),
confirmation=f"restart {NAMESPACE}", env=READY_ENV,
),
]
for payload in payloads:
rendered = json.dumps(payload, default=str).lower()
self.assertNotIn("kill -9", rendered)
self.assertNotIn("pkill -f", rendered)
def test_policy_declares_no_raw_kill_and_no_silent_restart(self):
policy = sanctioned_restart.restart_policy()
self.assertFalse(policy["raw_kill_exposed"])
self.assertFalse(policy["console_executes_process_kill"])
self.assertFalse(policy["fleet_scope_permitted"])
self.assertFalse(policy["silent_auto_restart_permitted"])
self.assertTrue(policy["audit_required"])
class TestContaminationClassification(unittest.TestCase):
"""AC2: manual pkill is contamination, and it blocks clean claims."""
def test_manual_daemon_pkill_is_contamination(self):
result = sanctioned_restart.classify_restart_command(
"pkill -f mcp_server.py"
)
self.assertTrue(result["contamination"])
self.assertFalse(result["clean_claim_allowed"])
self.assertIsNotNone(result["contamination_marker"])
self.assertEqual(
result["sanctioned_alternative"],
sanctioned_restart.ACTION_RESTART_NAMESPACE,
)
def test_broad_process_kill_is_contamination(self):
result = sanctioned_restart.classify_restart_command("killall -9 Python")
self.assertTrue(result["contamination"])
self.assertFalse(result["clean_claim_allowed"])
def test_marker_names_the_sanctioned_alternative(self):
result = sanctioned_restart.classify_restart_command(
"pkill -f mcp_server.py"
)
marker = result["contamination_marker"]
self.assertIn(
sanctioned_restart.ACTION_RESTART_NAMESPACE, marker["detail"]
)
def test_benign_command_is_not_contamination(self):
result = sanctioned_restart.classify_restart_command("git status")
self.assertFalse(result["contamination"])
self.assertTrue(result["clean_claim_allowed"])
def test_no_command_is_not_contamination(self):
result = sanctioned_restart.classify_restart_command(None)
self.assertFalse(result["contamination"])
self.assertTrue(result["clean_claim_allowed"])
class TestPostRestartHealth(unittest.TestCase):
"""AC3: a clean post-restart claim needs live client-namespace proof."""
def test_live_client_probe_clears_the_session(self):
result = sanctioned_restart.verify_post_restart_health(
NAMESPACE,
probe_result={"success": True},
probe_source=mcp_namespace_health.PROBE_SOURCE_CLIENT,
registered_tools=["gitea_whoami"],
required_tool="gitea_whoami",
)
self.assertEqual(result["status"], sanctioned_restart.HEALTH_CLEAN)
self.assertTrue(result["clean_claim_allowed"])
self.assertTrue(result["mutations_allowed"])
def test_offline_probe_does_not_clear_the_session(self):
result = sanctioned_restart.verify_post_restart_health(
NAMESPACE,
probe_result={"success": True},
probe_source=mcp_namespace_health.PROBE_SOURCE_OFFLINE,
registered_tools=["gitea_whoami"],
required_tool="gitea_whoami",
)
self.assertFalse(result["clean_claim_allowed"])
self.assertFalse(result["mutations_allowed"])
def test_failed_probe_is_unhealthy(self):
result = sanctioned_restart.verify_post_restart_health(
NAMESPACE,
probe_result={"success": False, "error": "client is closing: EOF"},
probe_source=mcp_namespace_health.PROBE_SOURCE_CLIENT,
registered_tools=["gitea_whoami"],
required_tool="gitea_whoami",
)
self.assertEqual(result["status"], sanctioned_restart.HEALTH_UNHEALTHY)
self.assertFalse(result["clean_claim_allowed"])
def test_static_registration_alone_never_clears_the_session(self):
result = sanctioned_restart.verify_post_restart_health(
NAMESPACE,
registered_tools=["gitea_whoami"],
required_tool="gitea_whoami",
)
self.assertFalse(result["clean_claim_allowed"])
class TestAuditEmission(unittest.TestCase):
"""Every restart attempt is audited with actor, target, and result."""
def _run(self, principal, sink):
prior = os.environ.get(console_audit.AUDIT_LOG_ENV)
os.environ[console_audit.AUDIT_LOG_ENV] = sink
try:
return sanctioned_restart.execute_restart(
NAMESPACE,
principal=principal,
confirmation=f"restart {NAMESPACE}",
env=READY_ENV,
request_id="req-642",
)
finally:
if prior is None:
os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
else:
os.environ[console_audit.AUDIT_LOG_ENV] = prior
def test_allowed_attempt_is_written_with_actor_and_target(self):
with tempfile.TemporaryDirectory() as tmp:
sink = os.path.join(tmp, "audit.jsonl")
result = self._run(admin(), sink)
self.assertTrue(result["audit"]["written"])
with open(sink, encoding="utf-8") as handle:
record = json.loads(handle.read().strip())
self.assertEqual(
record["action"], sanctioned_restart.ACTION_RESTART_NAMESPACE
)
self.assertEqual(record["target"]["namespace"], NAMESPACE)
self.assertEqual(record["target"]["mode"], "restart")
self.assertEqual(record["result"], console_audit.RESULT_ALLOWED)
self.assertEqual(record["actor"]["subject"], "[email protected]")
self.assertFalse(record["metadata"]["process_kill_executed"])
def test_denied_attempt_is_audited_too(self):
with tempfile.TemporaryDirectory() as tmp:
sink = os.path.join(tmp, "audit.jsonl")
self._run(viewer(), sink)
with open(sink, encoding="utf-8") as handle:
record = json.loads(handle.read().strip())
self.assertEqual(record["result"], console_audit.RESULT_DENIED)
self.assertEqual(
record["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
)
def test_restart_audit_uses_break_glass_retention(self):
action = console_authz.get_action(
sanctioned_restart.ACTION_RESTART_NAMESPACE
)
self.assertEqual(
console_audit.retention_class_for(action),
console_audit.RETENTION_BREAK_GLASS,
)
class TestRegistrySurface(unittest.TestCase):
"""AC5: the console surfaces the control, still disabled, with no kill."""
def test_registry_exposes_both_actions_disabled(self):
registry = gated_actions.load_action_registry()
for action_id in (
sanctioned_restart.ACTION_RESTART_NAMESPACE,
sanctioned_restart.ACTION_RELOAD_NAMESPACE,
):
with self.subTest(action=action_id):
action = registry.get(action_id)
self.assertIsNotNone(action)
self.assertFalse(action.enabled)
def test_registry_preview_names_the_namespace_target(self):
preview = gated_actions.preview_action(
sanctioned_restart.ACTION_RESTART_NAMESPACE, namespace=NAMESPACE
)
target = preview["mutation_ledger"][0]["target"]
self.assertIn(NAMESPACE, target)
self.assertFalse(preview["enabled"])
def test_registry_attempt_fails_closed(self):
result = gated_actions.attempt_action(
sanctioned_restart.ACTION_RESTART_NAMESPACE, namespace=NAMESPACE
)
self.assertFalse(result["success"])
if __name__ == "__main__":
unittest.main()
+345
View File
@@ -0,0 +1,345 @@
"""Tests for the system-health dashboard view (#639).
Covers the acceptance criteria directly: the page renders the health DTO
fields (AC1), degraded dependencies are visible (AC2), stale runtime is warned
prominently and never rendered as mutation-safe (AC3), healthy and degraded
fixtures both render (AC4), and the shell carries a nav entry (AC5).
"""
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.deployment_boundary import scan_text_for_client_secrets
from webui.layout import render_page
from webui.nav import iter_nav_items
from webui.system_health import (
STATUS_DEGRADED,
STATUS_DOWN,
STATUS_OK,
STATUS_SKIPPED,
STATUS_UNPROVEN,
DependencyProbe,
StaleRuntime,
SystemHealthSnapshot,
VersionInfo,
)
from webui.system_health_views import render_system_health_page
DASHBOARD_PATH = "/system-health"
def _version(*, known: bool = True) -> VersionInfo:
return VersionInfo(
git_sha="1c455b6ec0f9cb761fe6248de68c17e061fb5ecd" if known else None,
git_describe="v0.4.1-12-g1c455b6" if known else None,
control_plane_schema_version=4 if known else None,
python_version="3.13.1",
known=known,
)
def _parity(*, stale: bool = False, determinable: bool = True) -> StaleRuntime:
if stale:
return StaleRuntime(
daemon_head="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
checkout_head="bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
remote_head="cccccccccccccccccccccccccccccccccccccccc",
stale=True,
determinable=True,
mutation_safe=False,
reasons=("runtime, checkout, and remote commits disagree",),
)
if not determinable:
return StaleRuntime(
daemon_head=None,
checkout_head=None,
remote_head=None,
stale=False,
determinable=False,
mutation_safe=False,
reasons=("local checkout HEAD could not be read",),
)
return StaleRuntime(
daemon_head="1c455b6ec0f9cb761fe6248de68c17e061fb5ecd",
checkout_head="1c455b6ec0f9cb761fe6248de68c17e061fb5ecd",
remote_head="1c455b6ec0f9cb761fe6248de68c17e061fb5ecd",
stale=False,
determinable=True,
mutation_safe=True,
reasons=(),
)
def _snapshot(
*,
status: str = STATUS_OK,
ready: bool = True,
readiness_complete: bool = True,
readiness_reasons: tuple[str, ...] = (),
dependencies: tuple[DependencyProbe, ...] | None = None,
parity: StaleRuntime | None = None,
namespaces: tuple[dict, ...] = (),
probe_errors: tuple[str, ...] = (),
version_known: bool = True,
) -> SystemHealthSnapshot:
if dependencies is None:
dependencies = (
DependencyProbe(
name="control_plane_db",
kind="sqlite",
status=STATUS_OK,
detail="schema version 4",
required=True,
latency_ms=1.25,
metadata={"schema_version": 4},
),
)
return SystemHealthSnapshot(
status=status,
ready=ready,
readiness_complete=readiness_complete,
readiness_reasons=readiness_reasons,
service="mcp-control-plane-webui",
mode="read-only",
version=_version(known=version_known),
started_at="2026-07-23T19:50:47+00:00",
uptime_seconds=3661.5,
timestamp="2026-07-23T20:51:48+00:00",
deep_probes_requested=False,
dependencies=dependencies,
mcp_namespaces=namespaces,
stale_runtime=parity if parity is not None else _parity(),
probe_errors=probe_errors,
)
class TestHealthyRender(unittest.TestCase):
"""AC1 / AC4 — every health DTO field reaches the page."""
def setUp(self):
self.html = render_system_health_page(_snapshot())
def test_readiness_fields_render(self):
self.assertIn("System health", self.html)
self.assertIn("Ready", self.html)
self.assertIn("mcp-control-plane-webui", self.html)
self.assertIn("read-only", self.html)
self.assertIn("2026-07-23T20:51:48+00:00", self.html)
def test_version_and_uptime_render(self):
self.assertIn("1c455b6ec0f9cb761fe6248de68c17e061fb5ecd", self.html)
self.assertIn("v0.4.1-12-g1c455b6", self.html)
self.assertIn("3.13.1", self.html)
self.assertIn("3661.500s", self.html)
self.assertIn("1.02h", self.html)
def test_dependency_row_renders_with_latency(self):
self.assertIn("control_plane_db", self.html)
self.assertIn("sqlite", self.html)
self.assertIn("schema version 4", self.html)
self.assertIn("1.2 ms", self.html)
def test_healthy_page_shows_no_stale_warning(self):
self.assertNotIn("Stale runtime:", self.html)
self.assertNotIn("Staleness", self.html)
def test_unknown_version_is_labelled_not_faked(self):
html = render_system_health_page(_snapshot(version_known=False))
self.assertIn("unknown", html)
self.assertIn("unresolved", html)
class TestDegradedRender(unittest.TestCase):
"""AC2 — a degraded or unrun dependency is visible, not swallowed."""
def setUp(self):
self.deps = (
DependencyProbe(
name="control_plane_db",
kind="sqlite",
status=STATUS_OK,
detail="schema version 4",
required=True,
latency_ms=0.9,
),
DependencyProbe(
name="repository",
kind="git",
status=STATUS_DOWN,
detail="repository root is not a git checkout",
required=True,
latency_ms=4.0,
),
DependencyProbe(
name="gitea",
kind="http",
status=STATUS_SKIPPED,
detail="deep probe not requested",
required=False,
),
)
self.html = render_system_health_page(
_snapshot(
status=STATUS_DEGRADED,
ready=False,
readiness_complete=False,
readiness_reasons=("required dependency 'repository' is down",),
dependencies=self.deps,
)
)
def test_degraded_banner_names_the_dependency(self):
self.assertIn("Degraded dependencies:", self.html)
self.assertIn("repository", self.html)
def test_not_run_probe_is_reported_separately(self):
self.assertIn("Not probed:", self.html)
self.assertIn("gitea", self.html)
self.assertIn("not counted", self.html)
def test_not_ready_headline_and_reason(self):
self.assertIn("Not ready", self.html)
self.assertIn("required dependency &#x27;repository&#x27; is down", self.html)
def test_degraded_status_badge_present(self):
self.assertIn("badge-health-degraded", self.html)
self.assertIn("badge-health-down", self.html)
def test_ready_but_incomplete_is_not_shown_as_plain_ready(self):
html = render_system_health_page(
_snapshot(ready=True, readiness_complete=False)
)
self.assertIn("Ready (incomplete evidence)", html)
class TestStaleRuntimeWarning(unittest.TestCase):
"""AC3 — staleness is prominent and never claims mutation safety."""
def test_stale_runtime_warns_and_denies_mutation_safety(self):
html = render_system_health_page(_snapshot(parity=_parity(stale=True)))
self.assertIn("Stale runtime:", html)
self.assertIn("do not treat this runtime as mutation-safe", html)
self.assertIn("<tr><th>Mutation safe</th><td>False</td></tr>", html)
def test_indeterminate_parity_is_not_reported_safe(self):
html = render_system_health_page(
_snapshot(parity=_parity(determinable=False))
)
self.assertIn("Staleness", html)
self.assertIn("<tr><th>Mutation safe</th><td>False</td></tr>", html)
self.assertIn("<tr><th>Determinable</th><td>False</td></tr>", html)
def test_healthy_parity_reports_mutation_safe_true(self):
html = render_system_health_page(_snapshot())
self.assertIn("<tr><th>Mutation safe</th><td>True</td></tr>", html)
class TestNamespacesAndErrors(unittest.TestCase):
def test_unproven_namespace_rows_render(self):
html = render_system_health_page(
_snapshot(
namespaces=(
{
"namespace": "gitea-author",
"required_tool": "gitea_lock_issue",
"status": STATUS_UNPROVEN,
"ide_namespace_proven": False,
"reason": "the web console cannot invoke the IDE-managed MCP client",
},
)
)
)
self.assertIn("gitea-author", html)
self.assertIn("gitea_lock_issue", html)
self.assertIn("badge-health-unproven", html)
def test_no_namespaces_degrades_gracefully(self):
html = render_system_health_page(_snapshot(namespaces=()))
self.assertIn("No MCP namespaces are declared.", html)
def test_probe_errors_render_when_present(self):
html = render_system_health_page(
_snapshot(probe_errors=("probe raised: disk offline",))
)
self.assertIn("Probe errors", html)
self.assertIn("disk offline", html)
def test_probe_error_card_absent_when_clean(self):
self.assertNotIn("Probe errors", render_system_health_page(_snapshot()))
class TestReadOnlyAndRedaction(unittest.TestCase):
def test_no_restart_or_kill_controls(self):
html = render_system_health_page(_snapshot())
self.assertNotIn("<button", html)
self.assertNotIn("<form", html)
self.assertNotIn("pkill", html)
self.assertIn("read-only", html)
def test_recovery_points_at_sanctioned_path(self):
html = render_system_health_page(_snapshot())
self.assertIn("Reconnect the MCP client", html)
self.assertIn("Never kill the daemon process manually", html)
def test_secret_shaped_detail_is_redacted(self):
leaky = DependencyProbe(
name="gitea",
kind="http",
status=STATUS_DOWN,
detail="auth failed for token=ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789",
required=False,
latency_ms=12.0,
)
html = render_system_health_page(_snapshot(dependencies=(leaky,)))
self.assertNotIn("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", html)
def test_html_in_detail_is_escaped(self):
hostile = DependencyProbe(
name="repository",
kind="git",
status=STATUS_DOWN,
detail="<script>alert(1)</script>",
required=True,
)
html = render_system_health_page(_snapshot(dependencies=(hostile,)))
self.assertNotIn("<script>", html)
self.assertIn("&lt;script&gt;", html)
class TestNavAndRoute(unittest.TestCase):
"""AC5 — the shell links the dashboard, and the route serves it."""
def setUp(self):
self.client = TestClient(create_app())
def test_nav_contains_system_health(self):
self.assertIn(
(DASHBOARD_PATH, "System health"),
[(item.href, item.label) for item in iter_nav_items()],
)
def test_rendered_shell_links_dashboard(self):
page = render_page(title="Home", body_html="<p>x</p>")
self.assertIn(f'href="{DASHBOARD_PATH}"', page)
def test_route_renders_dashboard(self):
response = self.client.get(DASHBOARD_PATH)
self.assertEqual(response.status_code, 200)
self.assertIn("System health", response.text)
self.assertIn("Stale-runtime parity", response.text)
def test_route_is_read_only(self):
self.assertEqual(self.client.post(DASHBOARD_PATH).status_code, 405)
def test_live_page_leaks_no_client_secret(self):
findings = scan_text_for_client_secrets(self.client.get(DASHBOARD_PATH).text)
self.assertEqual(findings, [])
if __name__ == "__main__": # pragma: no cover
unittest.main()
+3 -1
View File
@@ -35,8 +35,10 @@ class TestCanonicalRepoRoot(unittest.TestCase):
self.assertEqual(root, CONTROL_ROOT) self.assertEqual(root, CONTROL_ROOT)
def test_falls_back_when_git_unavailable(self): def test_falls_back_when_git_unavailable(self):
# When fallback is a path under <repo>/branches/<worktree>, recover
# <repo> via commonpath ancestry (never string-split on "/branches/").
root = amw.resolve_canonical_repo_root("/missing/path", MCP_PROCESS_ROOT) root = amw.resolve_canonical_repo_root("/missing/path", MCP_PROCESS_ROOT)
self.assertEqual(root, os.path.realpath(MCP_PROCESS_ROOT)) self.assertEqual(root, os.path.realpath(CONTROL_ROOT))
class TestWorkspaceRepoMembership(unittest.TestCase): class TestWorkspaceRepoMembership(unittest.TestCase):
+19 -14
View File
@@ -306,6 +306,11 @@ def load_analytics(
model_filter = (model or "").strip() or None model_filter = (model or "").strip() or None
stage_filter = (stage or "").strip() or None stage_filter = (stage or "").strip() or None
# F4: coerce optional scope filters to str so AnalyticsSnapshot never holds None.
scope_remote = (remote or "").strip()
scope_org = (org or "").strip()
scope_repo = (repo or "").strip()
try: try:
db = control_plane_db.ControlPlaneDB(db_path=db_path) db = control_plane_db.ControlPlaneDB(db_path=db_path)
rows = db.query_usage_events( rows = db.query_usage_events(
@@ -326,9 +331,9 @@ def load_analytics(
ok=False, ok=False,
reason=f"control_plane_db_unavailable: {exc}", reason=f"control_plane_db_unavailable: {exc}",
schema_version=ANALYTICS_SCHEMA_VERSION, schema_version=ANALYTICS_SCHEMA_VERSION,
remote=remote, remote=scope_remote,
org=org, org=scope_org,
repo=repo, repo=scope_repo,
total_events=0, total_events=0,
overall_summary=empty_summary, overall_summary=empty_summary,
by_project={}, by_project={},
@@ -346,24 +351,24 @@ def load_analytics(
UsageEvent( UsageEvent(
usage_id=r["usage_id"], usage_id=r["usage_id"],
session_id=r.get("session_id"), session_id=r.get("session_id"),
remote=r.get("remote", remote), remote=r.get("remote") or scope_remote,
org=r.get("org", org), org=r.get("org") or scope_org,
repo=r.get("repo", repo), repo=r.get("repo") or scope_repo,
project_id=r.get("project_id"), project_id=r.get("project_id"),
role=r.get("role", "unknown"), role=r.get("role") or "unknown",
model=r.get("model", "unknown"), model=r.get("model") or "unknown",
issue_number=r.get("issue_number"), issue_number=r.get("issue_number"),
pr_number=r.get("pr_number"), pr_number=r.get("pr_number"),
stage=r.get("stage", "unknown"), stage=r.get("stage") or "unknown",
input_tokens=r.get("input_tokens"), input_tokens=r.get("input_tokens"),
output_tokens=r.get("output_tokens"), output_tokens=r.get("output_tokens"),
total_tokens=r.get("total_tokens"), total_tokens=r.get("total_tokens"),
estimated_cost_usd=r.get("estimated_cost_usd"), estimated_cost_usd=r.get("estimated_cost_usd"),
latency_ms=r.get("latency_ms"), latency_ms=r.get("latency_ms"),
duration_ms=r.get("duration_ms"), duration_ms=r.get("duration_ms"),
status=r.get("status", "success"), status=r.get("status") or "success",
metadata=meta, metadata=meta,
created_at=r.get("created_at", ""), created_at=r.get("created_at") or "",
) )
) )
@@ -410,9 +415,9 @@ def load_analytics(
ok=True, ok=True,
reason="ok", reason="ok",
schema_version=ANALYTICS_SCHEMA_VERSION, schema_version=ANALYTICS_SCHEMA_VERSION,
remote=remote, remote=scope_remote,
org=org, org=scope_org,
repo=repo, repo=scope_repo,
total_events=len(parsed_events), total_events=len(parsed_events),
overall_summary=overall_summary, overall_summary=overall_summary,
by_project=by_project, by_project=by_project,
+74 -27
View File
@@ -2,52 +2,59 @@
from __future__ import annotations from __future__ import annotations
import html
from webui.analytics_loader import AnalyticsSnapshot, GroupMetrics, UsageEvent from webui.analytics_loader import AnalyticsSnapshot, GroupMetrics, UsageEvent
from webui.layout import render_page from webui.layout import render_page
def _escape(text: object) -> str:
"""HTML-escape dynamic analytics fields (mirrors audit_views / project_views)."""
return html.escape(str(text), quote=True)
def _render_badge(text: str, badge_type: str = "muted") -> str: def _render_badge(text: str, badge_type: str = "muted") -> str:
return f'<span class="badge badge-{badge_type}">{text}</span>' return f'<span class="badge badge-{_escape(badge_type)}">{_escape(text)}</span>'
def _render_group_table(title: str, groups: dict[str, GroupMetrics], key_header: str = "Group") -> str: def _render_group_table(title: str, groups: dict[str, GroupMetrics], key_header: str = "Group") -> str:
if not groups: if not groups:
return ( return (
f"<h3>{title}</h3>" f"<h3>{_escape(title)}</h3>"
'<div class="card"><p class="muted">No telemetry events recorded for this dimension.</p></div>' '<div class="card"><p class="muted">No telemetry events recorded for this dimension.</p></div>'
) )
rows = [] rows = []
for key, g in sorted(groups.items(), key=lambda x: x[1].total_events, reverse=True): for key, g in sorted(groups.items(), key=lambda x: x[1].total_events, reverse=True):
cost_cell = ( cost_cell = (
f'<span class="accent">{g.display_cost}</span>' f'<span class="accent">{_escape(g.display_cost)}</span>'
if g.events_with_cost > 0 if g.events_with_cost > 0
else _render_badge("Unknown") else _render_badge("Unknown")
) )
tokens_cell = ( tokens_cell = (
g.display_tokens _escape(g.display_tokens)
if g.events_with_tokens > 0 if g.events_with_tokens > 0
else _render_badge("Unknown") else _render_badge("Unknown")
) )
lat_p50 = ( lat_p50 = (
g.display_latency_p50 _escape(g.display_latency_p50)
if g.events_with_latency > 0 if g.events_with_latency > 0
else _render_badge("Unknown") else _render_badge("Unknown")
) )
lat_p90 = ( lat_p90 = (
g.display_latency_p90 _escape(g.display_latency_p90)
if g.events_with_latency > 0 if g.events_with_latency > 0
else _render_badge("Unknown") else _render_badge("Unknown")
) )
dur_avg = ( dur_avg = (
g.display_duration_avg _escape(g.display_duration_avg)
if g.events_with_duration > 0 if g.events_with_duration > 0
else _render_badge("Unknown") else _render_badge("Unknown")
) )
rows.append( rows.append(
"<tr>" "<tr>"
f"<td><strong>{key}</strong></td>" f"<td><strong>{_escape(key)}</strong></td>"
f"<td>{g.total_events}</td>" f"<td>{g.total_events}</td>"
f"<td>{tokens_cell}</td>" f"<td>{tokens_cell}</td>"
f"<td>{cost_cell}</td>" f"<td>{cost_cell}</td>"
@@ -59,12 +66,12 @@ def _render_group_table(title: str, groups: dict[str, GroupMetrics], key_header:
rows_html = "".join(rows) rows_html = "".join(rows)
return f""" return f"""
<h3>{title}</h3> <h3>{_escape(title)}</h3>
<div class="card" style="overflow-x: auto;"> <div class="card" style="overflow-x: auto;">
<table class="data-table"> <table class="data-table">
<thead> <thead>
<tr> <tr>
<th>{key_header}</th> <th>{_escape(key_header)}</th>
<th>Events</th> <th>Events</th>
<th>Total Tokens</th> <th>Total Tokens</th>
<th>Est. Cost</th> <th>Est. Cost</th>
@@ -85,26 +92,49 @@ def _render_events_table(events: tuple[UsageEvent, ...]) -> str:
if not events: if not events:
return ( return (
"<h3>Recent Usage & Instrumentation Events</h3>" "<h3>Recent Usage & Instrumentation Events</h3>"
'<div class="card"><p class="muted">No individual telemetry events recorded yet. Opt-in instrumentation via session logging or POST /api/v1/analytics/usage.</p></div>' '<div class="card"><p class="muted">No individual telemetry events recorded yet. Opt-in instrumentation via session logging or authorized POST /api/v1/analytics/usage.</p></div>'
) )
rows = [] rows = []
for e in list(events)[-50:]: # Display latest 50 for e in list(events)[-50:]: # Display latest 50
work_item = f"issue #{e.issue_number}" if e.issue_number else (f"pr #{e.pr_number}" if e.pr_number else "unlinked") if e.issue_number is not None:
tokens = f"{e.total_tokens:,}" if e.total_tokens is not None else _render_badge("Unknown") work_item = f"issue #{e.issue_number}"
cost = f"${e.estimated_cost_usd:.4f}" if e.estimated_cost_usd is not None else _render_badge("Unknown") elif e.pr_number is not None:
latency = f"{e.latency_ms} ms" if e.latency_ms is not None else _render_badge("Unknown") work_item = f"pr #{e.pr_number}"
duration = f"{e.duration_ms} ms" if e.duration_ms is not None else _render_badge("Unknown") else:
status_badge = _render_badge(e.status, "success" if e.status == "success" else "danger") work_item = "unlinked"
tokens = (
_escape(f"{e.total_tokens:,}")
if e.total_tokens is not None
else _render_badge("Unknown")
)
cost = (
_escape(f"${e.estimated_cost_usd:.4f}")
if e.estimated_cost_usd is not None
else _render_badge("Unknown")
)
latency = (
_escape(f"{e.latency_ms} ms")
if e.latency_ms is not None
else _render_badge("Unknown")
)
duration = (
_escape(f"{e.duration_ms} ms")
if e.duration_ms is not None
else _render_badge("Unknown")
)
status_badge = _render_badge(
e.status, "success" if e.status == "success" else "danger"
)
rows.append( rows.append(
"<tr>" "<tr>"
f"<td>#{e.usage_id}</td>" f"<td>#{e.usage_id}</td>"
f"<td><small>{e.created_at}</small></td>" f"<td><small>{_escape(e.created_at)}</small></td>"
f"<td><span class=\"badge\">{e.role}</span></td>" f"<td><span class=\"badge\">{_escape(e.role)}</span></td>"
f"<td><strong>{e.model}</strong></td>" f"<td><strong>{_escape(e.model)}</strong></td>"
f"<td>{e.stage}</td>" f"<td>{_escape(e.stage)}</td>"
f"<td>{work_item}</td>" f"<td>{_escape(work_item)}</td>"
f"<td>{tokens}</td>" f"<td>{tokens}</td>"
f"<td>{cost}</td>" f"<td>{cost}</td>"
f"<td>{latency}</td>" f"<td>{latency}</td>"
@@ -145,15 +175,32 @@ def render_analytics_page(snapshot: AnalyticsSnapshot) -> str:
"""Render the main Model Usage & Performance Analytics console page.""" """Render the main Model Usage & Performance Analytics console page."""
summary = snapshot.overall_summary summary = snapshot.overall_summary
kpi_tokens = summary.display_tokens if summary.events_with_tokens > 0 else _render_badge("Unknown") kpi_tokens = (
kpi_cost = summary.display_cost if summary.events_with_cost > 0 else _render_badge("Unknown") _escape(summary.display_tokens)
kpi_lat_p50 = summary.display_latency_p50 if summary.events_with_latency > 0 else _render_badge("Unknown") if summary.events_with_tokens > 0
kpi_dur_avg = summary.display_duration_avg if summary.events_with_duration > 0 else _render_badge("Unknown") else _render_badge("Unknown")
)
kpi_cost = (
_escape(summary.display_cost)
if summary.events_with_cost > 0
else _render_badge("Unknown")
)
kpi_lat_p50 = (
_escape(summary.display_latency_p50)
if summary.events_with_latency > 0
else _render_badge("Unknown")
)
kpi_dur_avg = (
_escape(summary.display_duration_avg)
if summary.events_with_duration > 0
else _render_badge("Unknown")
)
status_notice = "" status_notice = ""
if not snapshot.ok: if not snapshot.ok:
status_notice = ( status_notice = (
f'<div class="card warning-card"><strong>Degraded Data Source:</strong> {snapshot.reason}</div>' f'<div class="card warning-card"><strong>Degraded Data Source:</strong> '
f'{_escape(snapshot.reason)}</div>'
) )
body_html = f""" body_html = f"""
+67 -5
View File
@@ -59,6 +59,7 @@ from webui.system_health import (
process_uptime, process_uptime,
snapshot_to_dict as system_health_to_dict, snapshot_to_dict as system_health_to_dict,
) )
from webui.system_health_views import render_system_health_page
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"}) _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"}) _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -167,6 +168,24 @@ async def api_system_health(request: Request) -> JSONResponse:
return JSONResponse(payload, status_code=200 if snapshot.ready else 503) return JSONResponse(payload, status_code=200 if snapshot.ready else 503)
async def system_health(request: Request) -> HTMLResponse:
"""Read-only system-health dashboard (#639).
Shares the #634 snapshot loader with the JSON API so the page can never
disagree with it. `?deep=1` opts into the network probe exactly as the API
does; the default page load stays cheap. The response is always 200: this
is an operator view that must render the degraded state, not withhold it.
"""
deep = _truthy_flag(request.query_params.get("deep"))
snapshot = load_system_health(deep=deep)
return HTMLResponse(
render_page(
title="System health",
body_html=render_system_health_page(snapshot),
)
)
async def queue(_request: Request) -> HTMLResponse: async def queue(_request: Request) -> HTMLResponse:
snapshot = load_queue_snapshot() snapshot = load_queue_snapshot()
return HTMLResponse(render_page(title="Queue", body_html=render_queue_page(snapshot))) return HTMLResponse(render_page(title="Queue", body_html=render_queue_page(snapshot)))
@@ -588,14 +607,53 @@ async def api_v1_analytics(request: Request) -> JSONResponse:
async def api_v1_analytics_ingest(request: Request) -> JSONResponse: async def api_v1_analytics_ingest(request: Request) -> JSONResponse:
"""Optional session instrumentation ingestion endpoint (#651).""" """Optional session instrumentation ingestion endpoint (#651).
Fail-closed write: every request is authorized through console_authz
(``record_analytics_usage``) before any control-plane DB mutation. Phase 1
keeps ``execution_enabled=False`` and denies unauthenticated callers, so
this route cannot be used as an unauthenticated write or XSS injection
vector (PR #876 F2).
"""
try: try:
body = await request.json() body = await request.json()
except Exception: except Exception:
return JSONResponse({"error": "invalid_json", "detail": "body must be valid JSON"}, status_code=400) body = {}
if not isinstance(body, dict): if not isinstance(body, dict):
return JSONResponse({"error": "invalid_payload", "detail": "payload must be a JSON object"}, status_code=400) body = {}
principal = resolve_principal(headers=dict(request.headers))
decision = authorize(
"record_analytics_usage", principal, for_execution=True
)
allowed = bool(decision.allowed and decision.execution_enabled)
console_audit.record_event(
action_id="record_analytics_usage",
result=(
console_audit.RESULT_ALLOWED
if allowed
else console_audit.RESULT_DENIED
),
decision=decision,
principal=principal,
target=_audit_target("record_analytics_usage", body),
request_id=_request_id(),
detail=decision.detail,
)
authorization = decision.to_dict()
if not allowed:
return JSONResponse(
{
"ok": False,
"error": "unauthorized",
"detail": (
"POST /api/v1/analytics/usage requires an authenticated "
"principal with record_analytics_usage execution enabled"
),
"authorization": authorization,
},
status_code=403,
)
usage_id = record_usage( usage_id = record_usage(
session_id=body.get("session_id"), session_id=body.get("session_id"),
@@ -617,7 +675,10 @@ async def api_v1_analytics_ingest(request: Request) -> JSONResponse:
status=body.get("status", "success"), status=body.get("status", "success"),
metadata=body.get("metadata"), metadata=body.get("metadata"),
) )
return JSONResponse({"ok": True, "usage_id": usage_id}, status_code=201) return JSONResponse(
{"ok": True, "usage_id": usage_id, "authorization": authorization},
status_code=201,
)
async def method_not_allowed(request: Request, _exc: Exception) -> Response: async def method_not_allowed(request: Request, _exc: Exception) -> Response:
@@ -643,6 +704,7 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
Route("/", home, methods=["GET"]), Route("/", home, methods=["GET"]),
Route("/health", health, methods=["GET"]), Route("/health", health, methods=["GET"]),
Route(SYSTEM_HEALTH_API_PATH, api_system_health, methods=["GET"]), Route(SYSTEM_HEALTH_API_PATH, api_system_health, methods=["GET"]),
Route("/system-health", system_health, methods=["GET"]),
Route("/queue", queue, methods=["GET"]), Route("/queue", queue, methods=["GET"]),
Route("/api/queue", api_queue, methods=["GET"]), Route("/api/queue", api_queue, methods=["GET"]),
Route("/projects", projects, methods=["GET"]), Route("/projects", projects, methods=["GET"]),
+41
View File
@@ -236,6 +236,47 @@ _ACTION_SPECS: tuple[ConsoleAction, ...] = (
phase=3, phase=3,
summary="Remove a remote feature branch.", summary="Remove a remote feature branch.",
), ),
# #651 analytics ingest: local control-plane write, not a Gitea mutation.
# Phase 2 gated write so Phase 1 (ACTIVE_PHASE=1) fails closed on execution.
ConsoleAction(
action_id="record_analytics_usage",
task_key="record_analytics_usage",
action_class=CLASS_WRITE,
minimum_role=OPERATOR,
requires_confirmation=True,
dual_control=False,
break_glass=False,
phase=2,
summary="Ingest a model-usage / latency analytics event into the control-plane DB.",
),
# #642: sanctioned daemon lifecycle. These exist so operators have an
# audited path off `pkill -f mcp_server.py` (#630). Restart drops every
# in-flight request on a namespace, so it carries the same dual-control and
# break-glass weight as a merge; reload drains first and is privileged but
# not destructive. Neither ever exposes a raw kill: execution is handed to
# a host supervisor by ``webui.sanctioned_restart``.
ConsoleAction(
action_id="system.reload_namespace",
task_key="reload_namespace",
action_class=CLASS_PRIVILEGED,
minimum_role=CONTROLLER,
requires_confirmation=True,
dual_control=False,
break_glass=False,
phase=2,
summary="Gracefully reload one MCP namespace via the host supervisor.",
),
ConsoleAction(
action_id="system.restart_namespace",
task_key="restart_namespace",
action_class=CLASS_DESTRUCTIVE,
minimum_role=ADMIN,
requires_confirmation=True,
dual_control=True,
break_glass=True,
phase=2,
summary="Restart one MCP namespace via the host supervisor.",
),
) )
ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS} ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}
+13
View File
@@ -110,6 +110,8 @@ def _format_target(action_id: str, params: dict[str, Any]) -> str:
) )
if action_id == "create_issue": if action_id == "create_issue":
return f"issue {params.get('title', '?')!r}" return f"issue {params.get('title', '?')!r}"
if action_id in {"system.restart_namespace", "system.reload_namespace"}:
return f"MCP namespace {params.get('namespace', '?')!r}"
return "unspecified" return "unspecified"
@@ -165,6 +167,17 @@ def build_action_registry() -> ActionRegistry:
"gitea_create_issue_comment", "Post a PR review thread comment."), "gitea_create_issue_comment", "Post a PR review thread comment."),
("close_pr", "Close PR", "close_pr", "gitea_edit_pr", ("close_pr", "Close PR", "close_pr", "gitea_edit_pr",
"Close a pull request without merge."), "Close a pull request without merge."),
# #642: the sanctioned replacement for the forbidden manual daemon-kill
# recovery path (#630). The "tool" is a host supervisor hook, not an MCP
# call — the console never signals a process. Preview and gating live in
# ``webui.sanctioned_restart``; these stay disabled like every other
# registry entry.
("system.reload_namespace", "Reload MCP namespace", "reload_namespace",
"host.supervisor_reload",
"Gracefully reload one MCP namespace via the host supervisor."),
("system.restart_namespace", "Restart MCP namespace",
"restart_namespace", "host.supervisor_restart",
"Restart one MCP namespace via the host supervisor."),
) )
actions = tuple( actions = tuple(
GatedAction( GatedAction(
+19
View File
@@ -236,6 +236,25 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
.badge-in-review {{ color: #9ec8f0; border-color: #3d5f7a; }} .badge-in-review {{ color: #9ec8f0; border-color: #3d5f7a; }}
.badge-duplicate {{ color: #e0c27a; border-color: #6b5730; }} .badge-duplicate {{ color: #e0c27a; border-color: #6b5730; }}
.badge-stale {{ color: #c9b8e8; border-color: #5a4a78; }} .badge-stale {{ color: #c9b8e8; border-color: #5a4a78; }}
.badge-health-ok {{ color: #8fd19e; border-color: #3d6b4a; }}
.badge-health-degraded {{ color: #e0c27a; border-color: #6b5730; }}
.badge-health-down {{ color: #f0a8a8; border-color: #7a3b3b; }}
.badge-health-skipped {{ color: var(--muted); }}
.badge-health-unproven {{ color: #c9b8e8; border-color: #5a4a78; }}
.health-card {{
margin: 1.25rem 0;
padding: 0.85rem 1rem 1rem;
border: 1px solid var(--border);
border-radius: 8px;
background: var(--surface);
}}
.health-card h3 {{ margin: 0 0 0.5rem; font-size: 1.05rem; }}
.health-card h4 {{ margin: 1rem 0 0.35rem; font-size: 0.92rem; color: var(--muted); }}
.health-headline {{ color: var(--text); font-size: 1rem; margin: 0 0 0.5rem; }}
.health-degraded {{ border-left-color: #e0c27a; }}
.health-stale {{ border-left-color: #f0a8a8; }}
ul.reasons {{ margin: 0.35rem 0; padding-left: 1.15rem; color: var(--muted); font-size: 0.9rem; }}
ul.reasons li {{ margin-bottom: 0.3rem; }}
</style> </style>
{extra_head} {extra_head}
</head> </head>
+1
View File
@@ -38,6 +38,7 @@ class NavGroup:
NAV_GROUPS: tuple[NavGroup, ...] = ( NAV_GROUPS: tuple[NavGroup, ...] = (
NavGroup("Health", ( NavGroup("Health", (
NavItem("/health", "Liveness"), NavItem("/health", "Liveness"),
NavItem("/system-health", "System health"),
)), )),
NavGroup("Traffic", ( NavGroup("Traffic", (
NavItem("/queue", "Queue"), NavItem("/queue", "Queue"),
+613
View File
@@ -0,0 +1,613 @@
"""Sanctioned MCP restart and graceful reload controls (#642, Phase 2).
Sessions have historically recovered MCP connectivity by killing the host
daemon (``pkill -f mcp_server.py``, #630). That path stays forbidden: it kills
every namespace on the host, contaminates the surviving session, and leaves no
audit trail. This module is the sanctioned replacement.
A restart is modelled as a *gated action*, never as a command:
1. **Capability** the console action resolves through ``console_authz``
against ``task_capability_map``, so the console cannot invent an authority
the MCP layer does not already define.
2. **Preview** :func:`build_restart_preview` renders a mutation ledger and
the exact confirmation phrase. It never returns a shell command.
3. **Confirmation** the operator echoes a phrase naming the exact namespace
and mode. A phrase for one namespace never authorizes another.
4. **Operator authorization** host daemon maintenance is authorized out of
band through the environment (#630, and #710 finding F1: a worker session
cannot set an env var for an already-running daemon, so this cannot be
self-asserted the way a tool argument could).
5. **Execution** :func:`execute_restart` never spawns a process. Once every
gate passes it hands the request to the configured host-managed restart
hook; with no hook configured it fails closed.
6. **Health recheck** :func:`verify_post_restart_health` requires live
client-namespace probe evidence before any post-restart clean claim.
Manual ``pkill`` remains forbidden and is classified as contamination by
:func:`classify_restart_command`, which blocks clean claims (#630 AC3).
This module performs no I/O beyond reading its own environment configuration,
imports no MCP client, and holds no credential.
"""
from __future__ import annotations
import os
from dataclasses import asdict, dataclass
from typing import Any
import mcp_namespace_health
import runtime_recovery_guard
from webui import console_audit, console_authz
# --- Operations -------------------------------------------------------------
MODE_RESTART = "restart"
MODE_RELOAD = "reload"
MODES: tuple[str, ...] = (MODE_RESTART, MODE_RELOAD)
ACTION_RESTART_NAMESPACE = "system.restart_namespace"
ACTION_RELOAD_NAMESPACE = "system.reload_namespace"
ACTION_FOR_MODE: dict[str, str] = {
MODE_RESTART: ACTION_RESTART_NAMESPACE,
MODE_RELOAD: ACTION_RELOAD_NAMESPACE,
}
# Namespaces the console may target. An unlisted name fails closed rather than
# being passed through to a host hook.
KNOWN_NAMESPACES: tuple[str, ...] = tuple(
sorted(
set(mcp_namespace_health.DEFAULT_NAMESPACES)
| {"gitea-author", "gitea-reviewer", "gitea-merger",
"gitea-reconciler", "gitea-controller"}
)
)
# Scope tokens that would mean "everything at once". Explicit non-goal: the
# console never offers a fleet-wide restart, because that is the blast radius
# `pkill -f mcp_server.py` already had.
_FLEET_TOKENS = frozenset({"*", "all", "fleet", "any", ""})
# --- Environment configuration ----------------------------------------------
# Read server-side only; the value is an opaque host hook reference (e.g. a
# launchd label), never a command line, and is never rendered to a client.
RESTART_HOOK_ENV = "GITEA_SANCTIONED_RESTART_HOOK"
# --- Reason codes -----------------------------------------------------------
DENY_UNKNOWN_MODE = "unknown_mode"
DENY_UNKNOWN_NAMESPACE = "unknown_namespace"
DENY_FLEET_SCOPE = "fleet_scope_not_permitted"
DENY_UNAUTHORIZED = "unauthorized"
DENY_CONFIRMATION_MISSING = "confirmation_required"
DENY_CONFIRMATION_MISMATCH = "confirmation_mismatch"
DENY_OPERATOR_AUTHORIZATION = "operator_authorization_missing"
DENY_HOOK_NOT_CONFIGURED = "restart_hook_not_configured"
DENY_CONTAMINATED_RUNTIME = "contaminated_runtime"
ALLOW_HOST_ACTION_REQUIRED = "host_action_required"
# Post-restart verification outcomes.
HEALTH_CLEAN = "clean"
HEALTH_UNPROVEN = "unproven"
HEALTH_UNHEALTHY = "unhealthy"
def _clean(value: Any) -> str:
return str(value or "").strip()
# --- Mutation ledger --------------------------------------------------------
@dataclass(frozen=True)
class RestartLedgerEntry:
"""One planned step, shown before anything is asked of the host."""
sequence: int
step: str
summary: str
executes_process_kill: bool = False
def _mutation_ledger(namespace: str, mode: str) -> tuple[RestartLedgerEntry, ...]:
if mode == MODE_RELOAD:
middle = RestartLedgerEntry(
sequence=2,
step="host_graceful_reload",
summary=(
f"Ask the configured host supervisor to reload {namespace} "
"in place, draining in-flight requests. The console does not "
"signal the process itself."
),
)
else:
middle = RestartLedgerEntry(
sequence=2,
step="host_restart_hook",
summary=(
f"Ask the configured host supervisor to restart {namespace}. "
"The console never sends a signal and never runs a kill."
),
)
return (
RestartLedgerEntry(
sequence=1,
step="quiesce",
summary=(
f"Stop admitting new gated mutations for {namespace} and "
"record the intent before anything restarts."
),
),
middle,
RestartLedgerEntry(
sequence=3,
step="health_recheck",
summary=(
f"Re-probe {namespace} through the live client namespace and "
"prove the required tool is callable again."
),
),
RestartLedgerEntry(
sequence=4,
step="audit",
summary=(
"Append actor, target namespace, mode, and result to the "
"console audit log."
),
),
)
# --- Confirmation -----------------------------------------------------------
def confirmation_phrase(namespace: str, mode: str) -> str:
"""Exact phrase an operator must echo, naming the namespace and mode.
Binding the namespace into the phrase is the point: a confirmation typed
for ``gitea-author`` cannot be replayed against ``gitea-merger``.
"""
return f"{_clean(mode)} {_clean(namespace)}"
def confirmation_matches(
namespace: str, mode: str, confirmation: str | None
) -> bool:
"""Compare *confirmation* to the required phrase (exact, whitespace-trimmed)."""
return _clean(confirmation) == confirmation_phrase(namespace, mode)
# --- Scope validation -------------------------------------------------------
def _validate_scope(namespace: str, mode: str) -> tuple[str, str] | None:
"""Return ``(reason_code, detail)`` when the scope is refused."""
ns = _clean(namespace)
md = _clean(mode)
if md not in MODES:
return (
DENY_UNKNOWN_MODE,
f"Mode {md!r} is not one of {', '.join(MODES)}.",
)
if ns.lower() in _FLEET_TOKENS:
return (
DENY_FLEET_SCOPE,
(
"Fleet-wide restart is an explicit non-goal: it reproduces the "
"blast radius of `pkill -f mcp_server.py` (#630). Restart one "
"namespace at a time."
),
)
if ns not in KNOWN_NAMESPACES:
return (
DENY_UNKNOWN_NAMESPACE,
f"Namespace {ns!r} is not a known MCP namespace.",
)
return None
# --- Host hook --------------------------------------------------------------
def restart_hook(env: dict[str, str] | None = None) -> dict[str, Any]:
"""Report the configured host-managed restart hook.
The hook is a reference the *host* resolves (a supervisor label), not a
command this process runs. ``configured=False`` fails restart closed.
"""
source = env if env is not None else os.environ
reference = _clean(source.get(RESTART_HOOK_ENV))
return {
"configured": bool(reference),
"reference": reference or None,
"source": RESTART_HOOK_ENV if reference else None,
"self_assertable": False,
"console_executes_process": False,
}
# --- Preview ----------------------------------------------------------------
def build_restart_preview(
namespace: str,
mode: str = MODE_RESTART,
*,
principal: console_authz.Principal | None = None,
env: dict[str, str] | None = None,
) -> dict[str, Any]:
"""Render the dry-run preview for a restart/reload request.
Read-only: no authorization is granted, no host is contacted, and the
result never contains a shell command.
"""
ns = _clean(namespace)
md = _clean(mode)
action_id = ACTION_FOR_MODE.get(md, ACTION_RESTART_NAMESPACE)
action = console_authz.get_action(action_id)
decision = console_authz.authorize(action_id, principal)
scope_error = _validate_scope(ns, md)
hook = restart_hook(env)
operator = runtime_recovery_guard.operator_authorization(env)
return {
"action_id": action_id,
"namespace": ns,
"mode": md,
"scope_valid": scope_error is None,
"scope_reason_code": scope_error[0] if scope_error else None,
"scope_detail": scope_error[1] if scope_error else None,
"required_role": action.minimum_role if action else None,
"required_permission": action.mcp_permission if action else None,
"action_class": action.action_class if action else None,
"dual_control": action.dual_control if action else True,
"break_glass": action.break_glass if action else True,
"requires_confirmation": True,
"confirmation_phrase": confirmation_phrase(ns, md),
"mutation_ledger": [asdict(entry) for entry in _mutation_ledger(ns, md)],
"authorization": decision.to_dict(),
"operator_authorization": operator,
"restart_hook": hook,
"execution_enabled": False,
"raw_process_kill_exposed": False,
"known_namespaces": list(KNOWN_NAMESPACES),
"post_restart_verification_required": True,
}
# --- Gate -------------------------------------------------------------------
def assess_restart_request(
namespace: str,
mode: str = MODE_RESTART,
*,
principal: console_authz.Principal | None = None,
confirmation: str | None = None,
contamination_marker: dict[str, Any] | None = None,
env: dict[str, str] | None = None,
) -> dict[str, Any]:
"""Decide whether a restart request may proceed to the host hook.
Every gate must pass. The first failure wins and is reported with a stable
reason code; a pass never means "restarted", only "may be handed to the
configured host hook".
"""
ns = _clean(namespace)
md = _clean(mode)
action_id = ACTION_FOR_MODE.get(md, ACTION_RESTART_NAMESPACE)
preview = build_restart_preview(ns, md, principal=principal, env=env)
def refuse(reason_code: str, detail: str) -> dict[str, Any]:
return {
"allowed": False,
"gates_passed": False,
"reason_code": reason_code,
"detail": detail,
"action_id": action_id,
"namespace": ns,
"mode": md,
"preview": preview,
"execution_enabled": False,
}
scope_error = _validate_scope(ns, md)
if scope_error is not None:
return refuse(*scope_error)
# Authority is checked as an authorization decision, not an execution
# grant. ``for_execution=True`` asks "may the console perform this write?",
# and the answer here is permanently no: step 2 of the ledger is a request
# to the host supervisor, so the console's Phase 2 execution gate is not
# the relevant gate. Every branch below keeps ``execution_enabled`` False
# and :func:`execute_restart` never touches a process.
decision = console_authz.authorize(action_id, principal)
if not decision.allowed:
return refuse(DENY_UNAUTHORIZED, decision.detail)
if not _clean(confirmation):
return refuse(
DENY_CONFIRMATION_MISSING,
(
"Type the confirmation phrase "
f"{preview['confirmation_phrase']!r} to proceed."
),
)
if not confirmation_matches(ns, md, confirmation):
return refuse(
DENY_CONFIRMATION_MISMATCH,
(
"Confirmation does not name this namespace and mode; expected "
f"{preview['confirmation_phrase']!r}."
),
)
operator = preview["operator_authorization"]
if not operator["authorized"]:
return refuse(
DENY_OPERATOR_AUTHORIZATION,
(
"Host daemon maintenance requires out-of-band operator "
"authorization via "
f"{runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV}."
),
)
# #630's task-scoped gate deliberately lets a contaminated worker keep
# commenting and handing off. Restart is stricter and unconditional: a
# runtime already contaminated by a manual kill must be reconciled before
# it is restarted again, or the restart just launders the contamination.
if contamination_marker and not contamination_marker.get(
"cleared_by_reconciler"
):
return refuse(
DENY_CONTAMINATED_RUNTIME,
(
"A live contamination marker is present; clear it through the "
"reconciler path before restarting."
),
)
hook = preview["restart_hook"]
if not hook["configured"]:
return refuse(
DENY_HOOK_NOT_CONFIGURED,
(
"No host-managed restart hook is configured "
f"({RESTART_HOOK_ENV}). The console will not fall back to a "
"process kill."
),
)
return {
"allowed": True,
"gates_passed": True,
"reason_code": ALLOW_HOST_ACTION_REQUIRED,
"detail": (
"Every gate passed. The restart must be performed by the "
"configured host supervisor; the console does not signal the "
"process."
),
"action_id": action_id,
"namespace": ns,
"mode": md,
"preview": preview,
"execution_enabled": False,
"console_executes": False,
"console_active_phase": console_authz.ACTIVE_PHASE,
}
# --- Execution --------------------------------------------------------------
def execute_restart(
namespace: str,
mode: str = MODE_RESTART,
*,
principal: console_authz.Principal | None = None,
confirmation: str | None = None,
contamination_marker: dict[str, Any] | None = None,
env: dict[str, str] | None = None,
request_id: str | None = None,
session_id: str | None = None,
) -> dict[str, Any]:
"""Run every gate, audit the outcome, and hand off to the host.
This function never spawns a process, never sends a signal, and never
builds a command line. ``success`` is False in both directions: a refused
request is refused, and an authorized request still requires the host
supervisor to act.
"""
assessment = assess_restart_request(
namespace,
mode,
principal=principal,
confirmation=confirmation,
contamination_marker=contamination_marker,
env=env,
)
action_id = assessment["action_id"]
allowed = assessment["allowed"]
audit = console_audit.record_event(
action_id=action_id,
result=(
console_audit.RESULT_ALLOWED if allowed
else console_audit.RESULT_DENIED
),
principal=principal,
target={"namespace": assessment["namespace"], "mode": assessment["mode"]},
reason_code=assessment["reason_code"],
detail=assessment["detail"],
request_id=request_id,
session_id=session_id,
metadata={
"gates_passed": assessment["gates_passed"],
"process_kill_executed": False,
"post_restart_verification_required": True,
},
)
return {
"success": False,
"outcome": (
ALLOW_HOST_ACTION_REQUIRED if allowed else assessment["reason_code"]
),
"allowed": allowed,
"detail": assessment["detail"],
"namespace": assessment["namespace"],
"mode": assessment["mode"],
"action_id": action_id,
"process_kill_executed": False,
"host_hook": assessment["preview"]["restart_hook"],
"next_action": (
"Have the host supervisor perform the restart, then call "
"verify_post_restart_health with live client-namespace evidence "
"before claiming a clean session."
if allowed
else assessment["detail"]
),
"assessment": assessment,
"audit": audit,
}
# --- Contamination classification -------------------------------------------
def classify_restart_command(
command: str | None,
*,
mcp_pids: list[Any] | tuple[Any, ...] | None = None,
session_id: str | None = None,
remote: str | None = None,
role: str | None = None,
) -> dict[str, Any]:
"""Classify an operator-proposed recovery command (#630 AC2).
A manual ``pkill``/``kill`` of the MCP daemon is contamination, not a
restart. When contaminating, a durable marker is returned so downstream
gated mutations and clean claims fail closed.
"""
classification = runtime_recovery_guard.classify_recovery_command(
command, mcp_pids=mcp_pids
)
contaminating = bool(classification.get("contamination"))
marker = None
if contaminating:
marker = runtime_recovery_guard.build_contamination_record(
reason_class=(
classification.get("reason_class")
or runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL
),
command_redacted=classification.get("redacted_command"),
session_id=session_id,
remote=remote,
role=role,
detail=(
"Manual daemon kill is forbidden; use the sanctioned "
f"{ACTION_RESTART_NAMESPACE} gated action instead."
),
)
return {
"contamination": contaminating,
"sanctioned": not contaminating and not classification.get("process_kill"),
"clean_claim_allowed": not contaminating,
"reason_class": classification.get("reason_class"),
"redacted_command": classification.get("redacted_command"),
"classification": classification,
"contamination_marker": marker,
"sanctioned_alternative": ACTION_RESTART_NAMESPACE,
}
# --- Post-restart health verification ---------------------------------------
def verify_post_restart_health(
namespace: str,
*,
probe_result: dict[str, Any] | None = None,
probe_source: str | None = None,
registered_tools: list[str] | tuple[str, ...] | None = None,
required_tool: str | None = None,
profile: str | None = None,
) -> dict[str, Any]:
"""Require live proof a namespace is callable before any clean claim (AC3).
Static registration is not proof and neither is an offline subprocess
probe: only ``probe_source=client_namespace`` evidence can clear a
post-restart session for mutations.
"""
ns = _clean(namespace)
health = mcp_namespace_health.classify_namespace_probe(
ns,
required_tool=required_tool,
registered_tools=registered_tools,
probe_result=probe_result,
profile=profile,
probe_source=probe_source,
)
healthy = bool(health.get("healthy"))
proven = bool(health.get("ide_namespace_proven"))
if healthy and proven:
status = HEALTH_CLEAN
elif healthy:
status = HEALTH_UNPROVEN
else:
status = HEALTH_UNHEALTHY
reasons = list(health.get("reasons") or [])
if status == HEALTH_UNPROVEN:
reasons.append(
"Namespace reported healthy without live client-namespace "
"evidence; a post-restart clean claim requires "
f"probe_source={mcp_namespace_health.PROBE_SOURCE_CLIENT!r}."
)
return {
"namespace": ns,
"status": status,
"healthy": healthy,
"ide_namespace_proven": proven,
"clean_claim_allowed": status == HEALTH_CLEAN,
"mutations_allowed": status == HEALTH_CLEAN,
"reasons": reasons,
"health": health,
}
# --- Policy surface ---------------------------------------------------------
def restart_policy() -> dict[str, Any]:
"""Machine-readable description of the sanctioned restart contract."""
return {
"policy_version": 1,
"modes": list(MODES),
"actions": [ACTION_RESTART_NAMESPACE, ACTION_RELOAD_NAMESPACE],
"known_namespaces": list(KNOWN_NAMESPACES),
"fleet_scope_permitted": False,
"console_executes_process_kill": False,
"raw_kill_exposed": False,
"requires_confirmation": True,
"confirmation_binds_namespace": True,
"operator_authorization_env": (
runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV
),
"restart_hook_env": RESTART_HOOK_ENV,
"manual_kill_classified_as": runtime_recovery_guard.CONTAMINATION_KIND,
"post_restart_clean_claim_requires": (
mcp_namespace_health.PROBE_SOURCE_CLIENT
),
"audit_required": True,
"silent_auto_restart_permitted": False,
}
+307
View File
@@ -0,0 +1,307 @@
"""HTML views for the system-health dashboard (#639).
Renders the read-only :class:`~webui.system_health.SystemHealthSnapshot`
produced by the Phase 1 system-health API (#634). The page offers no restart,
reload, or process-kill control: those are Phase 2 work, and manual process
kills are the contamination path #630 exists to prevent.
Every free-text field passes through :func:`webui.system_health.redact` before
it reaches HTML, so a probe detail that captured a token or a credentialed URL
cannot leak through the dashboard even though the API redacts it already.
"""
from __future__ import annotations
import html
from webui.system_health import (
STATUS_DEGRADED,
STATUS_DOWN,
STATUS_OK,
STATUS_SKIPPED,
STATUS_UNPROVEN,
DependencyProbe,
SystemHealthSnapshot,
redact,
)
_STATUS_BADGE_CLASS = {
STATUS_OK: "badge-health-ok",
STATUS_DEGRADED: "badge-health-degraded",
STATUS_DOWN: "badge-health-down",
STATUS_SKIPPED: "badge-health-skipped",
STATUS_UNPROVEN: "badge-health-unproven",
}
def _safe(value: object) -> str:
"""Escape free text for HTML after redacting anything secret-shaped.
Use this for every value that can carry arbitrary text probe details,
reasons, probe errors because those are where a credential could ride
along.
"""
return html.escape(redact(str(value)))
def _esc(value: object) -> str:
"""Escape a structured field for HTML without redacting it.
Commit SHAs, probe names, statuses, and timestamps are enumerated or
machine-generated, never credential-bearing. They must not go through
:func:`redact`: its opaque-token rule matches any 32-plus-character run,
so a 40-character git SHA would render as ``[redacted]`` and the parity
view the one thing an operator reads this page for would be blank.
"""
return html.escape(str(value))
def _status_badge(status: str) -> str:
css = _STATUS_BADGE_CLASS.get(status, "badge-health-unproven")
return f'<span class="badge {css}">{_esc(status)}</span>'
def _reason_list(reasons: tuple[str, ...], *, empty: str) -> str:
if not reasons:
return f"<p class='muted'>{html.escape(empty)}</p>"
items = "".join(f"<li>{_safe(reason)}</li>" for reason in reasons)
return f"<ul class='reasons'>{items}</ul>"
def _readiness_card(snapshot: SystemHealthSnapshot) -> str:
"""Overall readiness.
``ready`` and ``readiness_complete`` are shown separately on purpose: a
snapshot whose required probes never ran is not the same as one that ran
them and passed, and collapsing the two would render an unproven green.
"""
if snapshot.ready and snapshot.readiness_complete:
headline = "Ready"
elif snapshot.ready:
headline = "Ready (incomplete evidence)"
else:
headline = "Not ready"
return (
"<section class='health-card'>"
f"<h3>Readiness {_status_badge(snapshot.status)}</h3>"
f"<p class='health-headline'>{html.escape(headline)}</p>"
"<table class='detail'>"
f"<tr><th>Service</th><td><code>{_esc(snapshot.service)}</code></td></tr>"
f"<tr><th>Mode</th><td>{_esc(snapshot.mode)}</td></tr>"
f"<tr><th>Ready</th><td>{_esc(snapshot.ready)}</td></tr>"
"<tr><th>Readiness evidence complete</th>"
f"<td>{_esc(snapshot.readiness_complete)}</td></tr>"
"<tr><th>Deep probes requested</th>"
f"<td>{_esc(snapshot.deep_probes_requested)}</td></tr>"
f"<tr><th>Observed at</th><td><code>{_esc(snapshot.timestamp)}</code></td></tr>"
"</table>"
"<h4>Readiness reasons</h4>"
f"{_reason_list(snapshot.readiness_reasons, empty='No readiness objections recorded.')}"
"</section>"
)
def _version_card(snapshot: SystemHealthSnapshot) -> str:
version = snapshot.version
uptime_hours = snapshot.uptime_seconds / 3600.0
known = (
"resolved"
if version.known
else "unresolved — version fields could not be read from the checkout"
)
schema = version.control_plane_schema_version
return (
"<section class='health-card'>"
"<h3>Version and uptime</h3>"
"<table class='detail'>"
f"<tr><th>Git SHA</th><td><code>{_esc(version.git_sha or 'unknown')}</code></td></tr>"
"<tr><th>Git describe</th>"
f"<td><code>{_esc(version.git_describe or 'unknown')}</code></td></tr>"
"<tr><th>Control-plane schema</th>"
f"<td>{_esc(schema if schema is not None else 'unknown')}</td></tr>"
f"<tr><th>Python</th><td><code>{_esc(version.python_version)}</code></td></tr>"
f"<tr><th>Version status</th><td>{html.escape(known)}</td></tr>"
f"<tr><th>Started at</th><td><code>{_esc(snapshot.started_at)}</code></td></tr>"
"<tr><th>Uptime</th>"
f"<td>{snapshot.uptime_seconds:.3f}s ({uptime_hours:.2f}h)</td></tr>"
"</table>"
"</section>"
)
def _dependency_rows(probes: tuple[DependencyProbe, ...]) -> str:
if not probes:
return "<p class='muted'>No dependency probes were reported.</p>"
rows = []
for probe in probes:
latency = (
f"{probe.latency_ms:.1f} ms" if probe.latency_ms is not None else "n/a"
)
rows.append(
"<tr>"
f"<td><code>{_esc(probe.name)}</code></td>"
f"<td>{_esc(probe.kind)}</td>"
f"<td>{_status_badge(probe.status)}</td>"
f"<td>{_esc('required' if probe.required else 'optional')}</td>"
f"<td>{html.escape(latency)}</td>"
f"<td>{_safe(probe.detail)}</td>"
"</tr>"
)
return (
"<table class='registry'><thead><tr>"
"<th>Dependency</th><th>Kind</th><th>Status</th><th>Requirement</th>"
"<th>Latency</th><th>Detail</th>"
"</tr></thead><tbody>"
f"{''.join(rows)}</tbody></table>"
)
def _dependency_card(snapshot: SystemHealthSnapshot) -> str:
degraded = [probe for probe in snapshot.dependencies if probe.ran and not probe.healthy]
not_run = [probe for probe in snapshot.dependencies if not probe.ran]
banner = ""
if degraded:
names = ", ".join(sorted(probe.name for probe in degraded))
banner += (
"<div class='stub health-degraded'><p><strong>Degraded dependencies:</strong> "
f"{_esc(names)}</p></div>"
)
if not_run:
names = ", ".join(sorted(probe.name for probe in not_run))
banner += (
"<div class='stub'><p><strong>Not probed:</strong> "
f"{_esc(names)} — these contribute no evidence and are not counted "
"as healthy.</p></div>"
)
return (
"<section class='health-card'>"
"<h3>Dependencies</h3>"
f"{banner}"
f"{_dependency_rows(snapshot.dependencies)}"
"<p class='muted'>Details are redacted at the API boundary and again "
"before rendering; credentials are never displayed.</p>"
"</section>"
)
def _namespace_card(snapshot: SystemHealthSnapshot) -> str:
if not snapshot.mcp_namespaces:
body = "<p class='muted'>No MCP namespaces are declared.</p>"
else:
rows = []
for entry in snapshot.mcp_namespaces:
rows.append(
"<tr>"
f"<td><code>{_esc(entry.get('namespace'))}</code></td>"
f"<td><code>{_esc(entry.get('required_tool'))}</code></td>"
f"<td>{_status_badge(str(entry.get('status') or STATUS_UNPROVEN))}</td>"
f"<td>{_esc(entry.get('ide_namespace_proven'))}</td>"
f"<td>{_safe(entry.get('reason'))}</td>"
"</tr>"
)
body = (
"<table class='registry'><thead><tr>"
"<th>Namespace</th><th>Required tool</th><th>Status</th>"
"<th>IDE-proven</th><th>Reason</th>"
"</tr></thead><tbody>"
f"{''.join(rows)}</tbody></table>"
)
return (
"<section class='health-card'>"
"<h3>MCP namespaces</h3>"
f"{body}"
"<p class='muted'>The web process runs outside the IDE-managed MCP "
"client, so namespace health is reported as unproven rather than "
"guessed (#543).</p>"
"</section>"
)
def _stale_runtime_card(snapshot: SystemHealthSnapshot) -> str:
stale = snapshot.stale_runtime
if stale.stale:
warning = (
"<div class='stub health-stale'><p><strong>Stale runtime:</strong> "
"the running code, the checkout, and the remote-tracking commit "
"disagree. Capability gates may be evaluating obsolete code — "
"do not treat this runtime as mutation-safe.</p></div>"
)
elif not stale.determinable:
warning = (
"<div class='stub health-stale'><p><strong>Staleness "
"indeterminate:</strong> parity could not be proven, so this "
"runtime is not reported as mutation-safe.</p></div>"
)
else:
warning = ""
return (
"<section class='health-card'>"
"<h3>Stale-runtime parity</h3>"
f"{warning}"
"<table class='detail'>"
"<tr><th>Daemon head</th>"
f"<td><code>{_esc(stale.daemon_head or 'unknown')}</code></td></tr>"
"<tr><th>Checkout head</th>"
f"<td><code>{_esc(stale.checkout_head or 'unknown')}</code></td></tr>"
"<tr><th>Remote head</th>"
f"<td><code>{_esc(stale.remote_head or 'unknown')}</code></td></tr>"
f"<tr><th>Stale</th><td>{_esc(stale.stale)}</td></tr>"
f"<tr><th>Determinable</th><td>{_esc(stale.determinable)}</td></tr>"
f"<tr><th>Mutation safe</th><td>{_esc(stale.mutation_safe)}</td></tr>"
"</table>"
f"{_reason_list(stale.reasons, empty='Runtime, checkout, and remote agree.')}"
"</section>"
)
def _probe_error_card(snapshot: SystemHealthSnapshot) -> str:
if not snapshot.probe_errors:
return ""
return (
"<section class='health-card'>"
"<h3>Probe errors</h3>"
f"{_reason_list(snapshot.probe_errors, empty='')}"
"</section>"
)
def _recovery_card() -> str:
"""Sanctioned recovery pointers only — never a manual process kill (#630)."""
return (
"<section class='health-card'>"
"<h3>Recovery</h3>"
"<p class='muted'>This dashboard is read-only. Restart and reload "
"controls arrive in Phase 2 (#642); until then recovery runs through "
"the sanctioned client reconnect / operator restart path.</p>"
"<ul class='reasons'>"
"<li><a href='/runtime'>Runtime and session view</a> — active profile, "
"workflow hashes, and shell health.</li>"
"<li>Reconnect the MCP client from the IDE, then re-run the blocked "
"cycle. Never kill the daemon process manually: unmanaged kills are "
"recorded as runtime contamination (#630).</li>"
"<li>See <code>docs/webui-local-dev.md</code> for the documented "
"recovery sequence.</li>"
"</ul>"
"</section>"
)
def render_system_health_page(snapshot: SystemHealthSnapshot) -> str:
"""Render the full system-health dashboard body."""
return (
"<h2>System health</h2>"
"<p class='meta'>Read-only view of the Phase 1 system-health API "
"(<code>/api/v1/system/health</code>). Reload this page to refresh; "
"nothing here polls or mutates on your behalf.</p>"
f"{_readiness_card(snapshot)}"
f"{_stale_runtime_card(snapshot)}"
f"{_version_card(snapshot)}"
f"{_dependency_card(snapshot)}"
f"{_namespace_card(snapshot)}"
f"{_probe_error_card(snapshot)}"
f"{_recovery_card()}"
)