Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 jcwalker3
|
78e3befbbb | ||
  sysadminandClaude Opus 4.8
|
2d0d8a682b |
@@ -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(
|
||||
|
||||
@@ -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).
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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`
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
@@ -1,189 +0,0 @@
|
||||
"""Integration tests for autonomous canonical handoffs and dependency-aware task orchestration (#628).
|
||||
|
||||
Verifies the 21 acceptance criteria specified in umbrella Issue #628:
|
||||
- Non-terminal stage handoff generation and retrieval
|
||||
- Multi-worker concurrency and exclusive task assignment isolation
|
||||
- Structured dependency graph integration with the work allocator
|
||||
- Head SHA invalidation and stale review decision protection
|
||||
"""
|
||||
|
||||
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):
|
||||
"""AC1 & AC2: Every non-terminal stage stores and retrieves a valid canonical handoff."""
|
||||
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 issue #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):
|
||||
"""AC5 & AC6: Concurrency isolation ensures an exclusive task unit has only one active owner."""
|
||||
candidate = WorkCandidate(
|
||||
kind="issue",
|
||||
number=628,
|
||||
title="Umbrella #628 test 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):
|
||||
"""AC8, AC9, AC10: Durable dependency edges exclude blocked tasks from assignment."""
|
||||
# Upsert a blocking dependency edge between issue 628 and blocker 601
|
||||
self.db.upsert_dependency_edge(
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
source_kind="issue",
|
||||
source_number=628,
|
||||
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=628,
|
||||
)
|
||||
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=628,
|
||||
title="Blocked candidate",
|
||||
state="open",
|
||||
labels=[],
|
||||
blocked=False,
|
||||
dependency_unmet=True,
|
||||
dependency_reason="issue#628 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):
|
||||
"""AC11: Dependency completion updates edge state to MET."""
|
||||
self.db.upsert_dependency_edge(
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
source_kind="issue",
|
||||
source_number=628,
|
||||
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=628,
|
||||
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=628,
|
||||
)
|
||||
self.assertEqual(len(edges), 1)
|
||||
self.assertEqual(edges[0]["state"], "met")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user