From 8a63476787d7616030c43152476ab95fc09988d6 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 23 Jul 2026 03:21:07 -0400 Subject: [PATCH 1/2] fix: conflict-fix lease lifecycle chain termination and TTL handling (Closes #847, Refs #842) --- pr_work_lease.py | 53 ++++++++++++- tests/test_pr_work_lease.py | 153 ++++++++++++++++++++++++++++++++++++ 2 files changed, 204 insertions(+), 2 deletions(-) diff --git a/pr_work_lease.py b/pr_work_lease.py index 2fb6876..b877f10 100644 --- a/pr_work_lease.py +++ b/pr_work_lease.py @@ -228,25 +228,74 @@ def find_active_reviewer_lease( return None +def _conflict_fix_chain_key(lease: dict) -> tuple | None: + """Identity of the lease chain a conflict-fix marker belongs to (#842). + + Keyed by PR number, profile, head_before, and branch. Returns None when any + required component (pr_number, profile, head_before) is missing or malformed. + """ + raw = lease.get("raw_fields") or {} + pr_number = lease.get("pr_number") + profile = (lease.get("profile") or "").strip().lower() + head_before = lease.get("head_before") + branch = (lease.get("branch") or raw.get("branch") or "").strip() + if not (pr_number and profile and head_before): + return None + return (pr_number, profile, head_before, branch) + + +def _conflict_fix_chain_matches(key1: tuple, key2: tuple) -> bool: + """True when two conflict-fix chain keys refer to the same lease chain.""" + pr1, profile1, head1, branch1 = key1 + pr2, profile2, head2, branch2 = key2 + if pr1 != pr2 or profile1 != profile2 or head1 != head2: + return False + if branch1 and branch2 and branch1 != branch2: + return False + return True + + +def _conflict_fix_chain_terminated_after(entries: list[dict], index: int) -> bool: + """True when a later marker terminates the conflict-fix chain of ``entries[index]``. + + Append-only newest-wins: a terminal marker (phase=released/blocked/done) + ends only its matching claim chain (#842). + """ + key = _conflict_fix_chain_key(entries[index]) + if key is None: + return False + for later in entries[index + 1:]: + phase = (later.get("phase") or "").strip().lower() + if phase not in _TERMINAL_CONFLICT_FIX_PHASES: + continue + later_key = _conflict_fix_chain_key(later) + if later_key and _conflict_fix_chain_matches(key, later_key): + return True + return False + + def find_active_conflict_fix_lease( comments: list[dict], *, pr_number: int, now: datetime | None = None, ) -> dict[str, Any] | None: - """Return the newest unexpired conflict-fix lease for *pr_number*, if any.""" + """Return the newest unexpired, non-terminated conflict-fix lease for *pr_number*, if any.""" now = now or datetime.now(timezone.utc) candidates = [ entry for entry in _comment_entries(comments, pr_number=pr_number) if entry.get("lease_kind") == "conflict_fix" ] - for lease in reversed(candidates): + for index in range(len(candidates) - 1, -1, -1): + lease = candidates[index] if _lease_expired(lease, now=now): continue phase = (lease.get("phase") or "").strip().lower() if phase in _TERMINAL_CONFLICT_FIX_PHASES: continue if phase in _ACTIVE_CONFLICT_FIX_PHASES or phase: + if _conflict_fix_chain_terminated_after(candidates, index): + continue return lease return None diff --git a/tests/test_pr_work_lease.py b/tests/test_pr_work_lease.py index 8dbcb33..ddbf2cc 100644 --- a/tests/test_pr_work_lease.py +++ b/tests/test_pr_work_lease.py @@ -19,6 +19,7 @@ from pr_work_lease import ( # noqa: E402 assess_reviewer_mutation_blocked, assess_reviewer_stale_head_final_report, format_conflict_fix_lease_body, + find_active_conflict_fix_lease, parse_conflict_fix_lease_comment, parse_reviewer_lease_comment, ) @@ -203,5 +204,157 @@ class TestFormatLease(unittest.TestCase): self.assertEqual(parsed["pr_number"], 376) +class TestConflictFixLeaseLifecycle(unittest.TestCase): + def test_claim_followed_by_matching_release(self): + claim_body = _conflict_fix_body(phase="claimed", worktree="branches/fix-376") + expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z") + release_body = "\n".join([ + CONFLICT_FIX_LEASE_MARKER, + "pr: #376", + "branch: feat/fix-376", + "worktree: branches/fix-376", + "profile: prgs-author", + "phase: released", + f"head_before: {HEAD_A}", + f"head_after: {HEAD_B}", + f"expires_at: {expires}", + ]) + comments = [{"body": claim_body}, {"body": release_body}] + lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW) + self.assertIsNone(lease) + + def test_expired_claim_without_release(self): + past_expires = (NOW - timedelta(minutes=10)).isoformat().replace("+00:00", "Z") + claim_body = "\n".join([ + CONFLICT_FIX_LEASE_MARKER, + "pr: #376", + "phase: claimed", + f"head_before: {HEAD_A}", + f"expires_at: {past_expires}", + "profile: prgs-author", + ]) + comments = [{"body": claim_body}] + lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW) + self.assertIsNone(lease) + + def test_mismatched_release_different_head(self): + claim_body = _conflict_fix_body(phase="claimed", worktree="branches/fix-376") + expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z") + release_body = "\n".join([ + CONFLICT_FIX_LEASE_MARKER, + "pr: #376", + "profile: prgs-author", + "phase: released", + f"head_before: {HEAD_B}", + f"expires_at: {expires}", + ]) + comments = [{"body": claim_body}, {"body": release_body}] + lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW) + self.assertIsNotNone(lease) + self.assertEqual(lease["phase"], "claimed") + + def test_mismatched_release_different_branch(self): + claim_body = "\n".join([ + CONFLICT_FIX_LEASE_MARKER, + "pr: #376", + "branch: feat/branch-A", + "phase: claimed", + f"head_before: {HEAD_A}", + f"expires_at: {(NOW + timedelta(minutes=60)).isoformat().replace('+00:00', 'Z')}", + "profile: prgs-author", + ]) + release_body = "\n".join([ + CONFLICT_FIX_LEASE_MARKER, + "pr: #376", + "branch: feat/branch-B", + "phase: released", + f"head_before: {HEAD_A}", + f"expires_at: {(NOW + timedelta(minutes=60)).isoformat().replace('+00:00', 'Z')}", + "profile: prgs-author", + ]) + comments = [{"body": claim_body}, {"body": release_body}] + lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW) + self.assertIsNotNone(lease) + self.assertEqual(lease["phase"], "claimed") + + def test_release_followed_by_newer_claim(self): + claim_1 = _conflict_fix_body(phase="claimed", worktree="branches/fix-376") + expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z") + release_1 = "\n".join([ + CONFLICT_FIX_LEASE_MARKER, + "pr: #376", + "profile: prgs-author", + "phase: released", + f"head_before: {HEAD_A}", + f"head_after: {HEAD_B}", + f"expires_at: {expires}", + ]) + claim_2 = "\n".join([ + CONFLICT_FIX_LEASE_MARKER, + "pr: #376", + "profile: prgs-author", + "phase: claimed", + f"head_before: {HEAD_B}", + f"expires_at: {expires}", + ]) + comments = [{"body": claim_1}, {"body": release_1}, {"body": claim_2}] + lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW) + self.assertIsNotNone(lease) + self.assertEqual(lease["head_before"], HEAD_B) + + def test_malformed_or_ambiguous_markers(self): + malformed_release = "\n".join([ + CONFLICT_FIX_LEASE_MARKER, + "pr: #376", + "phase: released", + # missing head_before and profile + ]) + claim_body = _conflict_fix_body(phase="claimed") + comments = [{"body": claim_body}, {"body": malformed_release}] + lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW) + self.assertIsNotNone(lease) + + def test_pr818_historical_sequence(self): + comment_14696 = "\n".join([ + "", + "pr: #818", + "branch: feat/issue-638-webui-app-shell-phase1", + "worktree: /Users/jasonwalker/Development/Gitea-Tools/branches/issue-638-webui-app-shell-phase1", + "profile: prgs-author", + "session_id: unknown", + "phase: claimed", + "head_before: 08061b7b8aebdd099a37d1abf5dafcf38e4fd3fb", + "expires_at: 2026-07-23T07:12:13Z", + "reviewer_active: no", + ]) + comment_14730 = "\n".join([ + "", + "pr: #818", + "branch: feat/issue-638-webui-app-shell-phase1", + "worktree: /Users/jasonwalker/Development/Gitea-Tools/branches/issue-638-webui-app-shell-phase1", + "profile: prgs-author", + "session_id: prgs-author-61241-e5129c60", + "phase: released", + "head_before: 08061b7b8aebdd099a37d1abf5dafcf38e4fd3fb", + "head_after: 64b6eb5d5402663098de5ded3b0617cc3b3df98f", + "expires_at: 2026-07-23T06:05:00Z", + "reviewer_active: no", + ]) + comments = [{"body": comment_14696}, {"body": comment_14730}] + check_now = datetime(2026, 7, 23, 6, 30, tzinfo=timezone.utc) + lease = find_active_conflict_fix_lease(comments, pr_number=818, now=check_now) + self.assertIsNone(lease) + + reviewer_gate = assess_reviewer_mutation_blocked( + pr_number=818, + comments=comments, + reviewed_head_sha="64b6eb5d5402663098de5ded3b0617cc3b3df98f", + live_head_sha="64b6eb5d5402663098de5ded3b0617cc3b3df98f", + mutation="approve", + now=check_now, + ) + self.assertTrue(reviewer_gate["mutation_allowed"]) + + if __name__ == "__main__": unittest.main() \ No newline at end of file From 2d0d8a682b2fa26660fec90c0515e46fd39922d3 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 24 Jul 2026 02:14:27 -0400 Subject: [PATCH 2/2] feat(mcp-health): add MCP restart coordinator and impact analysis (Closes #658) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Child of umbrella #655 (governed MCP restart coordination); builds on the #657 restart-path inventory. Adds a central coordinator that evaluates live control-plane state before a restart and returns a blast-radius impact preview, so operators and the web console (#642/#652) can see what a restart would disrupt before concurrent LLM work is destroyed. Changes - restart_coordinator.py (new) — pure classification: inventory -> impact report DTO (RestartImpactReport/SessionImpact/LeaseImpact). Verdicts: safe / unsafe / override. Never restarts anything; fails closed on an incomplete inventory. - control_plane_db.py — additive ControlPlaneDB.list_sessions() read-only session inventory (the process-level unit a restart kills). - gitea_mcp_server.py — new dry-run MCP tool gitea_request_mcp_restart: gathers sessions/leases/terminal-lock from the #613 DB, calls the coordinator, returns the report. Override authority is read from the environment, never self-asserted (#630/#710 F1 pattern). Apply is gated by a later drain proof (non-goal here). - docs/mcp-restart-coordinator.md + docs/mcp-restart-impact-sample.json — doc and a real dry-run sample report. - docs/mcp-tool-inventory.md — register the new tool (inventory sync). - tests/test_restart_coordinator.py (new) — 15 tests: multi-session fixtures, deny-when-critical-section-open, fail-closed deny, override, terminal lock, stale heartbeat, JSON-serializable DTO, list_sessions. Tests: pytest tests/test_restart_coordinator.py -> 15 passed. Full suite: 13 failed / 4753 passed; all 13 reproduce identically on clean master @ef14622 (0 regressions). The residual test_issue_781 doc-registry failure is a pre-existing baseline gap for gitea_rebind_dirty_same_claimant_author_session (merged #864, undocumented on master) — out of scope for #658. Co-Authored-By: Claude Opus 4.8 (1M context) --- control_plane_db.py | 29 ++ docs/mcp-restart-coordinator.md | 95 ++++++ docs/mcp-restart-impact-sample.json | 148 +++++++++ docs/mcp-tool-inventory.md | 1 + gitea_mcp_server.py | 151 ++++++++++ restart_coordinator.py | 451 ++++++++++++++++++++++++++++ tests/test_restart_coordinator.py | 340 +++++++++++++++++++++ 7 files changed, 1215 insertions(+) create mode 100644 docs/mcp-restart-coordinator.md create mode 100644 docs/mcp-restart-impact-sample.json create mode 100644 restart_coordinator.py create mode 100644 tests/test_restart_coordinator.py diff --git a/control_plane_db.py b/control_plane_db.py index 75af7c9..4616cb3 100644 --- a/control_plane_db.py +++ b/control_plane_db.py @@ -599,6 +599,35 @@ class ControlPlaneDB: (_ts(), session_id), ) + def list_sessions( + self, + *, + statuses: Sequence[str] | None = None, + limit: int = 500, + ) -> list[dict[str, Any]]: + """List session rows for restart / impact analysis (#658). + + Read-only. Sessions are the process-level unit an MCP restart + disrupts, so the restart coordinator inventories them to compute blast + radius. Optional ``statuses`` filter (e.g. ``('active',)``) narrows to + live rows. Never returns secrets — only operational metadata. + """ + clauses: list[str] = [] + params: list[Any] = [] + if statuses: + placeholders = ", ".join("?" for _ in statuses) + clauses.append(f"status IN ({placeholders})") + params.extend(statuses) + where = ("WHERE " + " AND ".join(clauses)) if clauses else "" + sql = ( + f"SELECT * FROM sessions {where} " + "ORDER BY last_heartbeat_at DESC LIMIT ?" + ) + params.append(max(1, int(limit))) + with self._tx(immediate=False) as conn: + rows = conn.execute(sql, params).fetchall() + return [dict(r) for r in rows] + # ── work items ──────────────────────────────────────────────────────── def upsert_work_item( diff --git a/docs/mcp-restart-coordinator.md b/docs/mcp-restart-coordinator.md new file mode 100644 index 0000000..b1368c0 --- /dev/null +++ b/docs/mcp-restart-coordinator.md @@ -0,0 +1,95 @@ +# MCP restart coordinator and impact analysis (#658) + +Before any sanctioned MCP restart, a central coordinator evaluates the live +control-plane state and produces an **impact preview** so operators and the web +console (#642 / #652) can see the blast radius *before* concurrent LLM work is +disrupted. Uncoordinated restarts destroy in-flight author/reviewer/merger work +and give operators no way to see what they are about to break. + +This lands the coordinator + impact DTO + a dry-run MCP tool. It is the single +sanctioned entry point for restart evaluation post-#657 (which inventoried the +restart/reload/kill paths). The **mutative apply** path — actually performing a +restart — is a later child gated by a drain proof and is explicitly out of +scope here. + +## Components + +| Piece | Where | Responsibility | +|-------|-------|----------------| +| `restart_coordinator.evaluate_restart_impact` | `restart_coordinator.py` | Pure classification: inventory → impact report DTO. No I/O, no restart. | +| `RestartImpactReport` / `SessionImpact` / `LeaseImpact` | `restart_coordinator.py` | Console-facing DTO (`.as_dict()` is JSON-serializable). | +| `ControlPlaneDB.list_sessions` | `control_plane_db.py` | Read-only session inventory (the process-level unit a restart kills). | +| `gitea_request_mcp_restart` | `gitea_mcp_server.py` | MCP tool: gathers inventory from the #613 DB, calls the coordinator, returns the report. Dry-run only. | + +## Dimensions evaluated + +The coordinator classifies the inventory across the dimensions #658 requires: + +- **Sessions** — every active MCP session; a restart terminates all of them. + Liveness = `status == active` **and** the owner pid is alive **and** the + heartbeat is fresh (default window 15 min). Dead/stale sessions do not count + toward blast radius. +- **Leases / locks** — control-plane leases joined with work items and their + freshness (`lease_lifecycle.classify_lease_freshness`). Only `active` (live + owner) leases are *disruptive*; expired / released / dead-process leases never + withhold a restart. +- **Issue / PR work** — the issues and PRs behind disruptive leases. +- **Mutations / critical sections** — a live lease carrying an author worktree + or a mutating phase (`implementing`, `publishing`, `merging`, …) is a + critical section a restart must not sever. +- **Terminal (merge) lock** — an active terminal lock always makes a restart + unsafe. +- **Prior recovery attempts** — narrower recovery already tried (e.g. sanctioned + client reconnects) is echoed so the operator sees the escalation history. + +## Verdict + +Exactly three verdicts, matching the acceptance criteria: + +| Verdict | `allow_restart` | Meaning | +|---------|-----------------|---------| +| `safe` | `true` | No other live sessions, no live leases, no terminal lock. | +| `unsafe` | `false` | Live work would be disrupted and no operator override is present — **or** the inventory could not be completed (fail closed). | +| `override` | `true` | Live work present, but an operator override accepts the blast radius. | + +`override_would_allow` tells the console whether an override path exists for the +current state. `blast_radius` is a `none` / `low` / `medium` / `high` severity +band derived from the affected session and work counts. + +### Fail closed + +If the control-plane inventory cannot be completed (DB unavailable, a listing +failed), `inventory_complete` is `false` and the verdict is `unsafe` / deny. An +incomplete evaluation must never green-light a restart. + +### Operator override authority + +Override authority is read from the environment variable +`GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION` and **never** from a tool +argument. A worker session cannot set an environment variable on an +already-running daemon, so override cannot be self-asserted (same pattern as the +#630 daemon-maintenance authorization). The `request_override` tool argument only +expresses caller intent; it takes effect solely when the environment +authorization is present. + +## The tool + +```text +gitea_request_mcp_restart(remote, host, org, repo, + dry_run=True, request_override=False, + session_id=None, limit=200) +``` + +Read-only, dry-run, and it **never restarts anything**. `apply_supported` is +always `false`; passing `dry_run=False` performs no restart and reports that +apply is gated by a drain proof (a separate child). + +## Audit + +Every evaluation carries an `audit_record` (event, coordinator version, verdict, +allow decision, blast radius, counts, timestamp) so restart decisions are +auditable. No secrets flow through the coordinator — session ids, pids, and +profiles are operational metadata only. + +A representative dry-run report is in +[`mcp-restart-impact-sample.json`](./mcp-restart-impact-sample.json). diff --git a/docs/mcp-restart-impact-sample.json b/docs/mcp-restart-impact-sample.json new file mode 100644 index 0000000..e0136ee --- /dev/null +++ b/docs/mcp-restart-impact-sample.json @@ -0,0 +1,148 @@ +{ + "coordinator_version": "1.0.0-issue-658", + "evaluated_at": "2026-07-24T06:00:00+00:00", + "dry_run": true, + "restart_performed": false, + "inventory_complete": true, + "incomplete_reasons": [], + "verdict": "unsafe", + "allow_restart": false, + "override_would_allow": true, + "operator_override": false, + "blast_radius": "high", + "reasons": [ + "live work would be disrupted; restart denied without operator override", + "1 critical section(s) in flight (active lease with a live owner)" + ], + "affected_sessions": [ + { + "session_id": "prgs-author-30988-d6f43c25", + "role": "author", + "profile": "prgs-author", + "pid": 1, + "status": "active", + "alive": true, + "heartbeat_stale": false, + "is_requester": false, + "live": true + }, + { + "session_id": "prgs-reviewer-4157-0ce9", + "role": "reviewer", + "profile": "prgs-reviewer", + "pid": 1, + "status": "active", + "alive": true, + "heartbeat_stale": false, + "is_requester": true, + "live": true + } + ], + "affected_leases": [ + { + "lease_id": "lease-abc", + "session_id": "prgs-author-30988-d6f43c25", + "role": "author", + "phase": "implementing", + "freshness": "active", + "work_kind": "issue", + "work_number": 658, + "worktree_path": "/repo/branches/feat-issue-658", + "disruptive": true, + "is_mutation": true, + "is_critical_section": true + }, + { + "lease_id": "lease-dead", + "session_id": "prgs-author-91485", + "role": "author", + "phase": "allocated", + "freshness": "stale_dead_process", + "work_kind": "issue", + "work_number": 651, + "worktree_path": null, + "disruptive": false, + "is_mutation": false, + "is_critical_section": false + } + ], + "critical_sections": [ + { + "lease_id": "lease-abc", + "session_id": "prgs-author-30988-d6f43c25", + "role": "author", + "phase": "implementing", + "freshness": "active", + "work_kind": "issue", + "work_number": 658, + "worktree_path": "/repo/branches/feat-issue-658", + "disruptive": true, + "is_mutation": true, + "is_critical_section": true + } + ], + "affected_issues": [ + 658 + ], + "affected_prs": [], + "mutations": [ + { + "lease_id": "lease-abc", + "session_id": "prgs-author-30988-d6f43c25", + "role": "author", + "phase": "implementing", + "freshness": "active", + "work_kind": "issue", + "work_number": 658, + "worktree_path": "/repo/branches/feat-issue-658", + "disruptive": true, + "is_mutation": true, + "is_critical_section": true + } + ], + "terminal_lock": null, + "ack_state": { + "prgs-author-30988-d6f43c25": "pending" + }, + "prior_recovery_attempts": [ + { + "kind": "client_reconnect", + "at": "2026-07-24T06:00:00+00:00", + "outcome": "insufficient" + } + ], + "counts": { + "sessions_total": 2, + "sessions_live_other": 1, + "leases_total": 2, + "leases_disruptive": 1, + "critical_sections": 1, + "mutations": 1, + "affected_issues": 1, + "affected_prs": 0, + "prior_recovery_attempts": 1 + }, + "audit_record": { + "event": "restart_impact_evaluated", + "coordinator_version": "1.0.0-issue-658", + "evaluated_at": "2026-07-24T06:00:00+00:00", + "dry_run": true, + "operator_override": false, + "requesting_session_id": "prgs-reviewer-4157-0ce9", + "inventory_complete": true, + "verdict": "unsafe", + "allow_restart": false, + "blast_radius": "high", + "counts": { + "sessions_total": 2, + "sessions_live_other": 1, + "leases_total": 2, + "leases_disruptive": 1, + "critical_sections": 1, + "mutations": 1, + "affected_issues": 1, + "affected_prs": 0, + "prior_recovery_attempts": 1 + } + } +} diff --git a/docs/mcp-tool-inventory.md b/docs/mcp-tool-inventory.md index 8cb865b..b1aed12 100644 --- a/docs/mcp-tool-inventory.md +++ b/docs/mcp-tool-inventory.md @@ -135,6 +135,7 @@ that gates each call, not which tools exist. - `gitea_release_merger_pr_lease` - `gitea_release_reviewer_pr_lease` - `gitea_release_workflow_lease` +- `gitea_request_mcp_restart` - `gitea_resolve_task_capability` - `gitea_resume_review_draft` - `gitea_review_pr` diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 8e1661a..ed5fc11 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2040,6 +2040,7 @@ import dependency_graph # noqa: E402 # #784 durable dependency edges import control_plane_db # noqa: E402 import lease_lifecycle # noqa: E402 import workflow_dashboard # noqa: E402 # #605 live queue/lease dashboard +import restart_coordinator # noqa: E402 # #658 MCP restart coordinator/impact import incident_bridge # noqa: E402 import sentry_observability # noqa: E402 (#606 optional Sentry observability) import sentry_incident_bridge # noqa: E402 (#607 Sentry→Gitea incident bridge) @@ -21489,6 +21490,156 @@ def gitea_workflow_dashboard( return payload +@mcp.tool() +def gitea_request_mcp_restart( + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + dry_run: bool = True, + request_override: bool = False, + session_id: str | None = None, + limit: int = 200, +) -> dict: + """Evaluate a proposed MCP restart and return an impact preview (#658). + + Central restart coordinator: gathers live control-plane state (sessions, + leases/locks, in-flight issue/PR work, mutations, worktrees) and returns a + blast-radius impact report with a ``safe`` / ``unsafe`` / ``override`` + verdict, so the console (#642/#652) and operators can see what a restart + would disrupt *before* any concurrent LLM work is destroyed. + + This tool is **dry-run and never restarts anything.** The mutative apply + path is a separate child gated by a drain proof (non-goal here); calling + with ``dry_run=False`` still performs no restart and reports that apply is + not yet available. + + Operator override authority is read from the process environment + (``GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION``), never self-asserted by + the requesting session: ``request_override`` only expresses caller intent + and takes effect solely when that environment authorization is present. + + Fails closed: if the control-plane inventory cannot be completed, the + verdict is ``unsafe`` / deny (an incomplete evaluation must never green-light + a restart). + """ + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "read_only": True, + "dry_run": True, + "restart_performed": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + + try: + h, o, r = _resolve(remote, host, org, repo) + except ValueError as exc: + return { + "success": False, + "read_only": True, + "dry_run": True, + "restart_performed": False, + "reasons": [str(exc)], + } + + inventory_complete = True + incomplete_reasons: list[str] = [] + sessions: list[dict] = [] + leases: list[dict] = [] + terminal_lock: dict | None = None + + 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 evaluate 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=o, + repo=r, + role=None, + include_non_active=False, + 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))}" + ) + try: + terminal = db.get_active_terminal_lock( + remote=remote if remote in REMOTES else remote, + org=o, + repo=r, + ) + if terminal: + terminal_lock = dict(terminal) + except Exception as exc: # noqa: BLE001 + inventory_complete = False + incomplete_reasons.append( + f"terminal lock lookup failed: {_redact(str(exc))}" + ) + + profile = get_profile() + profile_name = (profile.get("profile_name") or "").strip() or "session" + sid = (session_id or "").strip() or f"{profile_name}-{os.getpid()}" + + # Override authority is read from the environment only — a worker session + # cannot set an env var for an already-running daemon, so it cannot be + # self-asserted the way a tool argument could (#630/#710 F1 pattern). + operator_authorized = bool( + (os.environ.get("GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION") or "").strip() + ) + operator_override = bool(request_override and operator_authorized) + + inventory = { + "sessions": sessions, + "leases": leases, + "terminal_lock": terminal_lock, + "inventory_complete": inventory_complete, + "incomplete_reasons": incomplete_reasons, + } + + report = restart_coordinator.evaluate_restart_impact( + inventory, + operator_override=operator_override, + requesting_session_id=sid, + dry_run=True, # coordinator is always analysis-only (#658) + ) + + payload = report.as_dict() + payload["success"] = True + payload["read_only"] = True + payload["remote"] = remote + payload["org"] = o + payload["repo"] = r + payload["requesting_session_id"] = sid + payload["operator_override_requested"] = bool(request_override) + payload["operator_override_authorized"] = operator_authorized + payload["apply_supported"] = False + if not dry_run: + payload["reasons"] = list(payload.get("reasons") or []) + [ + "apply requested but not supported: sanctioned restart apply is " + "gated by a drain proof (separate child); no restart performed (#658)" + ] + return payload + + @mcp.tool() def gitea_inspect_workflow_lease( lease_id: str, diff --git a/restart_coordinator.py b/restart_coordinator.py new file mode 100644 index 0000000..5db7de0 --- /dev/null +++ b/restart_coordinator.py @@ -0,0 +1,451 @@ +"""MCP restart coordinator and impact analysis (#658). + +Before any sanctioned MCP restart, a central coordinator must evaluate the +live control-plane state — active sessions, leases/locks, in-flight issue/PR +work, mutations, worktrees, and recovery history — and produce an *impact +preview* so operators (and the web console, #642/#652) can see the blast +radius **before** concurrent LLM work is disrupted. + +Design rules (mirrors the read-only posture of ``workflow_dashboard`` / +``lease_lifecycle``): + +* **Pure classification.** :func:`evaluate_restart_impact` takes an already + gathered inventory and returns a structured report. It never touches the + network, the filesystem, or a live process, so multi-session fixtures can + drive every branch in unit tests. The coordinator *never restarts anything*; + a mutative apply path is a later child gated by a drain proof (non-goal here). +* **Fail closed.** If the inventory is not explicitly complete, the verdict is + ``unsafe`` / deny — an incomplete evaluation must never green-light a restart. +* **No secrets.** Session ids, pids, and profiles are operational metadata, not + credentials; nothing secret flows through this module. + +The single sanctioned entry point post-#657 is the MCP tool +``gitea_request_mcp_restart`` (dry-run by default), which gathers the inventory +from the #613 control-plane DB and calls :func:`evaluate_restart_impact`. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Mapping, Sequence + +import lease_lifecycle + +COORDINATOR_VERSION = "1.0.0-issue-658" + +# Restart verdicts. Exactly the three the acceptance criteria name. +VERDICT_SAFE = "safe" +VERDICT_UNSAFE = "unsafe" +VERDICT_OVERRIDE = "override" + +# Blast-radius severity bands. +BLAST_NONE = "none" +BLAST_LOW = "low" +BLAST_MEDIUM = "medium" +BLAST_HIGH = "high" + +# A live lease with a live owner process is treated as active in-flight work. +LEASE_FRESHNESS_LIVE = "active" + +# Default staleness window for a session heartbeat (seconds). A session whose +# last heartbeat is older than this is not counted as live even if its row is +# still marked ``active`` — it is assumed dead/detached. +DEFAULT_SESSION_HEARTBEAT_STALE_SECONDS = 900 + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _parse_ts(value: str | None) -> datetime | None: + return lease_lifecycle._parse_ts(value) + + +@dataclass(frozen=True) +class SessionImpact: + """One MCP session a restart would terminate.""" + + session_id: str + role: str | None + profile: str | None + pid: int | None + status: str | None + alive: bool | None + heartbeat_stale: bool + is_requester: bool + live: bool + + def as_dict(self) -> dict[str, Any]: + return { + "session_id": self.session_id, + "role": self.role, + "profile": self.profile, + "pid": self.pid, + "status": self.status, + "alive": self.alive, + "heartbeat_stale": self.heartbeat_stale, + "is_requester": self.is_requester, + "live": self.live, + } + + +@dataclass(frozen=True) +class LeaseImpact: + """One control-plane lease a restart would disrupt.""" + + lease_id: str | None + session_id: str | None + role: str | None + phase: str | None + freshness: str | None + work_kind: str | None + work_number: int | None + worktree_path: str | None + disruptive: bool + is_mutation: bool + is_critical_section: bool + + def as_dict(self) -> dict[str, Any]: + return { + "lease_id": self.lease_id, + "session_id": self.session_id, + "role": self.role, + "phase": self.phase, + "freshness": self.freshness, + "work_kind": self.work_kind, + "work_number": self.work_number, + "worktree_path": self.worktree_path, + "disruptive": self.disruptive, + "is_mutation": self.is_mutation, + "is_critical_section": self.is_critical_section, + } + + +@dataclass(frozen=True) +class RestartImpactReport: + """Impact preview DTO returned to the console / operator (#642/#652).""" + + coordinator_version: str + evaluated_at: str + dry_run: bool + restart_performed: bool + inventory_complete: bool + verdict: str + allow_restart: bool + override_would_allow: bool + operator_override: bool + blast_radius: str + reasons: list[str] + affected_sessions: list[SessionImpact] + affected_leases: list[LeaseImpact] + critical_sections: list[LeaseImpact] + affected_issues: list[int] + affected_prs: list[int] + mutations: list[LeaseImpact] + terminal_lock: dict[str, Any] | None + ack_state: dict[str, str] + prior_recovery_attempts: list[dict[str, Any]] + counts: dict[str, int] + audit_record: dict[str, Any] + incomplete_reasons: list[str] = field(default_factory=list) + + def as_dict(self) -> dict[str, Any]: + return { + "coordinator_version": self.coordinator_version, + "evaluated_at": self.evaluated_at, + "dry_run": self.dry_run, + "restart_performed": self.restart_performed, + "inventory_complete": self.inventory_complete, + "incomplete_reasons": list(self.incomplete_reasons), + "verdict": self.verdict, + "allow_restart": self.allow_restart, + "override_would_allow": self.override_would_allow, + "operator_override": self.operator_override, + "blast_radius": self.blast_radius, + "reasons": list(self.reasons), + "affected_sessions": [s.as_dict() for s in self.affected_sessions], + "affected_leases": [l.as_dict() for l in self.affected_leases], + "critical_sections": [l.as_dict() for l in self.critical_sections], + "affected_issues": list(self.affected_issues), + "affected_prs": list(self.affected_prs), + "mutations": [l.as_dict() for l in self.mutations], + "terminal_lock": self.terminal_lock, + "ack_state": dict(self.ack_state), + "prior_recovery_attempts": list(self.prior_recovery_attempts), + "counts": dict(self.counts), + "audit_record": dict(self.audit_record), + } + + +def _classify_session( + row: Mapping[str, Any], + *, + now: datetime, + requesting_session_id: str | None, + heartbeat_stale_seconds: int, +) -> SessionImpact: + session_id = str(row.get("session_id") or "") + pid = row.get("pid") + status = (row.get("status") or "").strip().lower() or None + alive = lease_lifecycle.is_process_alive(pid) if pid is not None else None + hb = _parse_ts(row.get("last_heartbeat_at")) + heartbeat_stale = bool( + hb is not None and (now - hb).total_seconds() > heartbeat_stale_seconds + ) + live = bool(status == "active" and alive is not False and not heartbeat_stale) + return SessionImpact( + session_id=session_id, + role=row.get("role"), + profile=row.get("profile"), + pid=pid, + status=status, + alive=alive, + heartbeat_stale=heartbeat_stale, + is_requester=bool( + requesting_session_id and session_id == requesting_session_id + ), + live=live, + ) + + +# Lease phases that represent an active mutation in flight (as opposed to a +# mere allocation/claim with no work committed yet). An active lease in any of +# these phases is a critical section a restart must not sever. +_MUTATING_PHASES = frozenset( + { + "implementing", + "publishing", + "pushing", + "committing", + "reviewing", + "merging", + "reconciling", + "conflict_fix", + } +) + + +def _classify_lease(row: Mapping[str, Any]) -> LeaseImpact: + freshness_obj = row.get("freshness") + if isinstance(freshness_obj, Mapping): + freshness = str(freshness_obj.get("freshness") or "").strip().lower() or None + else: + freshness = str(freshness_obj or "").strip().lower() or None + phase = (row.get("phase") or "").strip().lower() or None + worktree = row.get("worktree_path") + disruptive = freshness == LEASE_FRESHNESS_LIVE + # A live lease is a mutation-in-flight if it carries an author worktree or + # its phase names a mutating step. All disruptive leases are critical + # sections a restart would sever regardless. + is_mutation = bool( + disruptive and (bool(worktree) or (phase in _MUTATING_PHASES)) + ) + number = row.get("work_number") + try: + number = int(number) if number is not None else None + except (TypeError, ValueError): + number = None + return LeaseImpact( + lease_id=row.get("lease_id"), + session_id=row.get("session_id"), + role=row.get("role"), + phase=phase, + freshness=freshness, + work_kind=(str(row.get("work_kind") or "").strip().lower() or None), + work_number=number, + worktree_path=worktree, + disruptive=disruptive, + is_mutation=is_mutation, + is_critical_section=disruptive, + ) + + +def _blast_radius(*, session_count: int, work_count: int, mutation_count: int) -> str: + if mutation_count > 0 or work_count >= 3 or session_count >= 3: + return BLAST_HIGH + if work_count > 0 or session_count == 2: + return BLAST_MEDIUM + if session_count == 1: + return BLAST_LOW + return BLAST_NONE + + +def evaluate_restart_impact( + inventory: Mapping[str, Any], + *, + now: datetime | None = None, + operator_override: bool = False, + requesting_session_id: str | None = None, + dry_run: bool = True, + session_heartbeat_stale_seconds: int = DEFAULT_SESSION_HEARTBEAT_STALE_SECONDS, +) -> RestartImpactReport: + """Evaluate a proposed MCP restart and return an impact preview. + + ``inventory`` is a mapping with: + + * ``sessions`` — session rows (session_id, role, profile, pid, status, + last_heartbeat_at). + * ``leases`` — control-plane lease rows, each ideally carrying an enriched + ``freshness`` dict (as :func:`lease_lifecycle.list_active_leases` returns); + a bare string freshness is also accepted. + * ``terminal_lock`` — the active terminal (merge) lock, if any. + * ``prior_recovery_attempts`` — narrower recovery attempts already tried + (e.g. sanctioned reconnects) so the operator sees escalation history. + * ``inventory_complete`` — bool. **Must** be explicitly True; a missing or + falsy value forces a deny (fail closed). + * ``incomplete_reasons`` — optional reasons the inventory is incomplete. + + The coordinator never restarts anything: ``restart_performed`` is always + False and the mutative apply path is a later drain-gated child. + """ + moment = now or _utc_now() + reasons: list[str] = [] + + inventory_complete = bool(inventory.get("inventory_complete", False)) + incomplete_reasons = [str(r) for r in (inventory.get("incomplete_reasons") or [])] + + sessions_raw: Sequence[Mapping[str, Any]] = inventory.get("sessions") or [] + leases_raw: Sequence[Mapping[str, Any]] = inventory.get("leases") or [] + terminal_lock = inventory.get("terminal_lock") or None + prior_recovery_attempts = [ + dict(a) for a in (inventory.get("prior_recovery_attempts") or []) + ] + + session_impacts = [ + _classify_session( + s, + now=moment, + requesting_session_id=requesting_session_id, + heartbeat_stale_seconds=session_heartbeat_stale_seconds, + ) + for s in sessions_raw + ] + lease_impacts = [_classify_lease(l) for l in leases_raw] + + # Only *other* live sessions and live leases constitute blast radius: a + # restart that would kill only the requesting session with no other work in + # flight is safe. + other_live_sessions = [ + s for s in session_impacts if s.live and not s.is_requester + ] + disruptive_leases = [l for l in lease_impacts if l.disruptive] + critical_sections = [l for l in lease_impacts if l.is_critical_section] + mutations = [l for l in lease_impacts if l.is_mutation] + + affected_issues = sorted( + { + l.work_number + for l in disruptive_leases + if l.work_kind == "issue" and l.work_number is not None + } + ) + affected_prs = sorted( + { + l.work_number + for l in disruptive_leases + if l.work_kind == "pr" and l.work_number is not None + } + ) + + disruptive = bool(disruptive_leases or other_live_sessions or terminal_lock) + + if not inventory_complete: + verdict = VERDICT_UNSAFE + allow_restart = False + reasons.append( + "inventory incomplete: restart evaluation cannot confirm blast " + "radius — deny (fail closed, #658)" + ) + reasons.extend(incomplete_reasons) + elif not disruptive: + verdict = VERDICT_SAFE + allow_restart = True + reasons.append("no other live sessions, live leases, or terminal lock") + elif operator_override: + verdict = VERDICT_OVERRIDE + allow_restart = True + reasons.append( + "live work present; operator override accepts the blast radius" + ) + else: + verdict = VERDICT_UNSAFE + allow_restart = False + reasons.append( + "live work would be disrupted; restart denied without operator " + "override" + ) + + if critical_sections and inventory_complete: + reasons.append( + f"{len(critical_sections)} critical section(s) in flight " + "(active lease with a live owner)" + ) + if terminal_lock: + reasons.append("active terminal (merge) lock present") + + override_would_allow = bool(inventory_complete and disruptive) + + blast_radius = _blast_radius( + session_count=len(other_live_sessions), + work_count=len(affected_issues) + len(affected_prs), + mutation_count=len(mutations), + ) + + # Acknowledgement is a later child (drain protocol); expose per-session + # placeholders so the console can render the ack column now. + ack_state = {s.session_id: "pending" for s in other_live_sessions} + + counts = { + "sessions_total": len(session_impacts), + "sessions_live_other": len(other_live_sessions), + "leases_total": len(lease_impacts), + "leases_disruptive": len(disruptive_leases), + "critical_sections": len(critical_sections), + "mutations": len(mutations), + "affected_issues": len(affected_issues), + "affected_prs": len(affected_prs), + "prior_recovery_attempts": len(prior_recovery_attempts), + } + + audit_record = { + "event": "restart_impact_evaluated", + "coordinator_version": COORDINATOR_VERSION, + "evaluated_at": moment.isoformat(), + "dry_run": dry_run, + "operator_override": bool(operator_override), + "requesting_session_id": requesting_session_id, + "inventory_complete": inventory_complete, + "verdict": verdict, + "allow_restart": allow_restart, + "blast_radius": blast_radius, + "counts": counts, + } + + return RestartImpactReport( + coordinator_version=COORDINATOR_VERSION, + evaluated_at=moment.isoformat(), + dry_run=dry_run, + restart_performed=False, + inventory_complete=inventory_complete, + verdict=verdict, + allow_restart=allow_restart, + override_would_allow=override_would_allow, + operator_override=bool(operator_override), + blast_radius=blast_radius, + reasons=reasons, + affected_sessions=session_impacts, + affected_leases=lease_impacts, + critical_sections=critical_sections, + affected_issues=affected_issues, + affected_prs=affected_prs, + mutations=mutations, + terminal_lock=dict(terminal_lock) + if isinstance(terminal_lock, Mapping) + else terminal_lock, + ack_state=ack_state, + prior_recovery_attempts=prior_recovery_attempts, + counts=counts, + audit_record=audit_record, + incomplete_reasons=incomplete_reasons, + ) diff --git a/tests/test_restart_coordinator.py b/tests/test_restart_coordinator.py new file mode 100644 index 0000000..558aaa0 --- /dev/null +++ b/tests/test_restart_coordinator.py @@ -0,0 +1,340 @@ +"""Tests for the MCP restart coordinator and impact analysis (#658). + +Multi-session fixtures exercise every verdict branch: safe, unsafe (live work), +override, and the fail-closed deny on incomplete inventory. Also covers the +critical-section deny path and the new ``ControlPlaneDB.list_sessions``. +""" + +from __future__ import annotations + +import os +import tempfile +import unittest +from datetime import datetime, timedelta, timezone + +import restart_coordinator as rc +from control_plane_db import ControlPlaneDB + + +NOW = datetime(2026, 7, 24, 6, 0, 0, tzinfo=timezone.utc) + + +def _ts(dt: datetime) -> str: + return dt.isoformat() + + +def _live_pid() -> int: + return os.getpid() + + +def _dead_pid() -> int: + # A pid that is essentially never alive. os.kill(0) on it raises + # ProcessLookupError → is_process_alive False. + return 2_000_000_000 + + +def _session(session_id, *, pid, status="active", heartbeat=None, role="author"): + return { + "session_id": session_id, + "role": role, + "profile": "prgs-author", + "pid": pid, + "status": status, + "last_heartbeat_at": _ts(heartbeat or NOW), + } + + +def _lease( + lease_id, + *, + session_id, + freshness, + kind="issue", + number=658, + phase="allocated", + worktree=None, + role="author", +): + return { + "lease_id": lease_id, + "session_id": session_id, + "role": role, + "phase": phase, + "work_kind": kind, + "work_number": number, + "worktree_path": worktree, + "freshness": {"freshness": freshness}, + } + + +class EvaluateRestartImpactTest(unittest.TestCase): + def test_incomplete_inventory_denies_fail_closed(self) -> None: + report = rc.evaluate_restart_impact( + {"inventory_complete": False, "incomplete_reasons": ["db down"]}, + now=NOW, + ) + self.assertEqual(report.verdict, rc.VERDICT_UNSAFE) + self.assertFalse(report.allow_restart) + self.assertFalse(report.restart_performed) + self.assertIn("db down", report.incomplete_reasons) + self.assertTrue( + any("fail closed" in reasoning for reasoning in report.reasons) + ) + + def test_missing_completeness_flag_denies(self) -> None: + # No inventory_complete key at all → treated as incomplete. + report = rc.evaluate_restart_impact({}, now=NOW) + self.assertEqual(report.verdict, rc.VERDICT_UNSAFE) + self.assertFalse(report.allow_restart) + + def test_no_other_work_is_safe(self) -> None: + report = rc.evaluate_restart_impact( + { + "inventory_complete": True, + "sessions": [_session("requester", pid=_live_pid())], + "leases": [], + }, + now=NOW, + requesting_session_id="requester", + ) + self.assertEqual(report.verdict, rc.VERDICT_SAFE) + self.assertTrue(report.allow_restart) + self.assertEqual(report.blast_radius, rc.BLAST_NONE) + self.assertEqual(report.affected_issues, []) + + def test_dead_foreign_session_and_lease_are_not_disruptive(self) -> None: + report = rc.evaluate_restart_impact( + { + "inventory_complete": True, + "sessions": [ + _session("requester", pid=_live_pid()), + _session("dead", pid=_dead_pid()), + ], + "leases": [ + _lease("l-dead", session_id="dead", freshness="stale_dead_process") + ], + }, + now=NOW, + requesting_session_id="requester", + ) + self.assertEqual(report.verdict, rc.VERDICT_SAFE) + self.assertTrue(report.allow_restart) + self.assertEqual(report.counts["leases_disruptive"], 0) + self.assertEqual(report.counts["sessions_live_other"], 0) + + def test_live_foreign_lease_denies_without_override(self) -> None: + report = rc.evaluate_restart_impact( + { + "inventory_complete": True, + "sessions": [ + _session("requester", pid=_live_pid()), + _session("worker", pid=_live_pid()), + ], + "leases": [ + _lease( + "l1", + session_id="worker", + freshness="active", + worktree="/tmp/wt-658", + phase="implementing", + ) + ], + }, + now=NOW, + requesting_session_id="requester", + ) + self.assertEqual(report.verdict, rc.VERDICT_UNSAFE) + self.assertFalse(report.allow_restart) + # Critical section detected: active lease with a live owner. + self.assertEqual(len(report.critical_sections), 1) + self.assertEqual(report.affected_issues, [658]) + self.assertEqual(report.counts["mutations"], 1) + self.assertTrue(report.override_would_allow) + self.assertEqual(report.blast_radius, rc.BLAST_HIGH) + # Placeholder ack state for the affected session. + self.assertEqual(report.ack_state.get("worker"), "pending") + + def test_operator_override_allows_despite_live_work(self) -> None: + inv = { + "inventory_complete": True, + "sessions": [ + _session("requester", pid=_live_pid()), + _session("worker", pid=_live_pid()), + ], + "leases": [_lease("l1", session_id="worker", freshness="active")], + } + report = rc.evaluate_restart_impact( + inv, + now=NOW, + requesting_session_id="requester", + operator_override=True, + ) + self.assertEqual(report.verdict, rc.VERDICT_OVERRIDE) + self.assertTrue(report.allow_restart) + self.assertFalse(report.restart_performed) + + def test_deny_when_critical_section_open(self) -> None: + # A single live author lease in a mutating phase is a critical section + # that must deny an un-overridden restart. + report = rc.evaluate_restart_impact( + { + "inventory_complete": True, + "sessions": [_session("worker", pid=_live_pid())], + "leases": [ + _lease( + "l1", + session_id="worker", + freshness="active", + phase="merging", + kind="pr", + number=900, + ) + ], + }, + now=NOW, + requesting_session_id="requester", + ) + self.assertEqual(report.verdict, rc.VERDICT_UNSAFE) + self.assertFalse(report.allow_restart) + self.assertEqual(report.affected_prs, [900]) + self.assertEqual(len(report.critical_sections), 1) + + def test_terminal_lock_makes_restart_unsafe(self) -> None: + report = rc.evaluate_restart_impact( + { + "inventory_complete": True, + "sessions": [_session("requester", pid=_live_pid())], + "leases": [], + "terminal_lock": {"terminal_pr": 812}, + }, + now=NOW, + requesting_session_id="requester", + ) + self.assertEqual(report.verdict, rc.VERDICT_UNSAFE) + self.assertFalse(report.allow_restart) + self.assertIsNotNone(report.terminal_lock) + self.assertTrue( + any("terminal" in reasoning for reasoning in report.reasons) + ) + + def test_other_live_session_without_lease_is_disruptive(self) -> None: + report = rc.evaluate_restart_impact( + { + "inventory_complete": True, + "sessions": [ + _session("requester", pid=_live_pid()), + _session("idle-but-live", pid=_live_pid()), + ], + "leases": [], + }, + now=NOW, + requesting_session_id="requester", + ) + self.assertEqual(report.verdict, rc.VERDICT_UNSAFE) + self.assertEqual(report.counts["sessions_live_other"], 1) + + def test_stale_heartbeat_session_not_counted_live(self) -> None: + stale = NOW - timedelta(hours=2) + report = rc.evaluate_restart_impact( + { + "inventory_complete": True, + "sessions": [ + _session("requester", pid=_live_pid()), + _session("stale", pid=_live_pid(), heartbeat=stale), + ], + "leases": [], + }, + now=NOW, + requesting_session_id="requester", + ) + self.assertEqual(report.verdict, rc.VERDICT_SAFE) + self.assertEqual(report.counts["sessions_live_other"], 0) + + def test_prior_recovery_attempts_echoed(self) -> None: + report = rc.evaluate_restart_impact( + { + "inventory_complete": True, + "sessions": [_session("requester", pid=_live_pid())], + "leases": [], + "prior_recovery_attempts": [ + {"kind": "client_reconnect", "at": _ts(NOW)} + ], + }, + now=NOW, + requesting_session_id="requester", + ) + self.assertEqual(len(report.prior_recovery_attempts), 1) + self.assertEqual(report.counts["prior_recovery_attempts"], 1) + + def test_bare_string_freshness_accepted(self) -> None: + lease = _lease("l1", session_id="worker", freshness="active") + lease["freshness"] = "active" # bare string, not a dict + report = rc.evaluate_restart_impact( + { + "inventory_complete": True, + "sessions": [_session("worker", pid=_live_pid())], + "leases": [lease], + }, + now=NOW, + requesting_session_id="requester", + ) + self.assertEqual(report.counts["leases_disruptive"], 1) + + def test_as_dict_is_serializable_dto(self) -> None: + import json + + report = rc.evaluate_restart_impact( + { + "inventory_complete": True, + "sessions": [_session("requester", pid=_live_pid())], + "leases": [], + }, + now=NOW, + requesting_session_id="requester", + ) + payload = report.as_dict() + # Round-trips through JSON — safe for the console DTO. + encoded = json.dumps(payload) + decoded = json.loads(encoded) + self.assertEqual(decoded["verdict"], rc.VERDICT_SAFE) + self.assertIn("audit_record", decoded) + self.assertEqual(decoded["audit_record"]["event"], "restart_impact_evaluated") + self.assertFalse(decoded["restart_performed"]) + self.assertIn("coordinator_version", decoded) + + +class ListSessionsTest(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3")) + + def tearDown(self) -> None: + self._tmp.cleanup() + + def test_list_sessions_filters_by_status(self) -> None: + self.db.upsert_session(session_id="a", role="author", pid=1, status="active") + self.db.upsert_session(session_id="b", role="author", pid=2, status="ended") + active = self.db.list_sessions(statuses=("active",)) + ids = {row["session_id"] for row in active} + self.assertEqual(ids, {"a"}) + every = self.db.list_sessions() + self.assertEqual({row["session_id"] for row in every}, {"a", "b"}) + + def test_list_sessions_feeds_coordinator(self) -> None: + self.db.upsert_session( + session_id="requester", role="author", pid=os.getpid(), status="active" + ) + report = rc.evaluate_restart_impact( + { + "inventory_complete": True, + "sessions": self.db.list_sessions(statuses=("active",)), + "leases": [], + }, + now=NOW, + requesting_session_id="requester", + ) + self.assertEqual(report.counts["sessions_total"], 1) + + +if __name__ == "__main__": # pragma: no cover + unittest.main()