From e8bae606cb1a82d4cb6f3e544c8ffe28e9617f0c Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 29 Jul 2026 05:27:42 -0400 Subject: [PATCH] fix(reconcile): retire workflow session rows whose owners are no longer live Implement a sanctioned session lifecycle for post-restart reconciliation so active session rows with dead or reused owner PIDs can be terminalized without deleting history. Protect live owners, live leases, and live client-managed sessions; record durable session_retired audit events; keep cleanup idempotent under concurrent reconciles. Closes #969 Co-Authored-By: Claude Opus 4.8 (1M context) --- control_plane_db.py | 233 ++++++- docs/post-restart-reconcile.md | 6 +- gitea_mcp_server.py | 151 ++++- post_restart_reconcile.py | 69 ++- session_lifecycle.py | 703 ++++++++++++++++++++++ task_capability_map.py | 14 +- tests/test_control_plane_db.py | 4 +- tests/test_issue_969_session_lifecycle.py | 480 +++++++++++++++ 8 files changed, 1630 insertions(+), 30 deletions(-) create mode 100644 session_lifecycle.py create mode 100644 tests/test_issue_969_session_lifecycle.py diff --git a/control_plane_db.py b/control_plane_db.py index 422003c..8ee3529 100644 --- a/control_plane_db.py +++ b/control_plane_db.py @@ -27,12 +27,12 @@ import uuid from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timedelta, timezone -from typing import Any, Iterator, Sequence +from typing import Any, Iterator, Mapping, Sequence import dependency_graph import gitea_audit -SCHEMA_VERSION = 5 +SCHEMA_VERSION = 6 # Assignable work kinds only — raw monitoring incidents are never work items. WORK_KINDS = frozenset({"issue", "pr"}) @@ -419,6 +419,7 @@ class ControlPlaneDB: self._migrate_incident_links_null_scope(conn) self._migrate_lease_lifecycle_columns(conn) self._migrate_session_ownership_columns(conn) + self._migrate_session_lifecycle_columns(conn) self._migrate_usage_events_table(conn) conn.execute( "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)", @@ -812,6 +813,7 @@ class ControlPlaneDB: pid: int | None = None, status: str = "active", controller_instance_id: str | None = None, + owner_process_started_at: str | None = None, ) -> dict[str, Any]: """Register/refresh a session row. @@ -821,16 +823,22 @@ class ControlPlaneDB: controller instance can. It is never overwritten with ``None``, so a heartbeat from a caller that does not supply one cannot erase ownership. + + *owner_process_started_at* (#969) is the OS start time of the owner + process at registration time. When present it is retained across + heartbeats (never cleared by ``None``) so later PID-reuse checks do + not depend on a live ``ps`` probe of a long-dead process. """ now = _ts() instance = (controller_instance_id or "").strip() or None + proc_start = (owner_process_started_at or "").strip() or None with self._tx() as conn: existing = conn.execute( "SELECT session_id FROM sessions WHERE session_id = ?", (session_id,), ).fetchone() if existing: - if instance is None: + if instance is None and proc_start is None: conn.execute( """ UPDATE sessions @@ -840,7 +848,21 @@ class ControlPlaneDB: """, (role, profile, namespace, pid, now, status, session_id), ) - else: + elif instance is None: + conn.execute( + """ + UPDATE sessions + SET role = ?, profile = ?, namespace = ?, pid = ?, + last_heartbeat_at = ?, status = ?, + owner_process_started_at = COALESCE(?, owner_process_started_at) + WHERE session_id = ? + """, + ( + role, profile, namespace, pid, now, status, + proc_start, session_id, + ), + ) + elif proc_start is None: conn.execute( """ UPDATE sessions @@ -854,18 +876,33 @@ class ControlPlaneDB: instance, session_id, ), ) + else: + conn.execute( + """ + UPDATE sessions + SET role = ?, profile = ?, namespace = ?, pid = ?, + last_heartbeat_at = ?, status = ?, + controller_instance_id = ?, + owner_process_started_at = COALESCE(?, owner_process_started_at) + WHERE session_id = ? + """, + ( + role, profile, namespace, pid, now, status, + instance, proc_start, session_id, + ), + ) else: conn.execute( """ INSERT INTO sessions( session_id, role, profile, namespace, pid, started_at, last_heartbeat_at, status, - controller_instance_id - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + controller_instance_id, owner_process_started_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( session_id, role, profile, namespace, pid, now, now, - status, instance, + status, instance, proc_start, ), ) row = conn.execute( @@ -881,6 +918,165 @@ class ControlPlaneDB: (_ts(), session_id), ) + def retire_session( + self, + *, + session_id: str, + reason: str, + actor_session_id: str | None = None, + details: Mapping[str, Any] | None = None, + now: datetime | None = None, + terminal_status: str = "retired", + ) -> dict[str, Any]: + """Terminalize one session row if it is still non-terminal (#969). + + CAS on non-terminal status: concurrent retirements of the same row + yield exactly one ``retired`` outcome and subsequent + ``already_terminal`` outcomes. Never deletes historical rows. Writes a + durable ``session_retired`` event for audit. + """ + moment = _ts(now) + reason_s = (reason or "").strip() or "unspecified" + term = (terminal_status or "retired").strip().lower() or "retired" + terminal_set = { + "retired", + "ended", + "terminal", + "dead", + "stale", + "orphaned", + } + with self._tx() as conn: + row = conn.execute( + "SELECT * FROM sessions WHERE session_id = ?", + (session_id,), + ).fetchone() + if row is None: + return { + "outcome": "missing", + "reason": reason_s, + "prior_status": None, + "new_status": None, + "details": {"session_id": session_id}, + } + prior = dict(row) + prior_status = str(prior.get("status") or "").strip().lower() + if prior_status in terminal_set: + return { + "outcome": "already_terminal", + "reason": reason_s, + "prior_status": prior_status, + "new_status": prior_status, + "details": {"session_id": session_id, "idempotent": True}, + } + + # Optional: refuse when an active lease still names this session. + active_lease = conn.execute( + """ + SELECT lease_id, status, expires_at FROM leases + WHERE session_id = ? AND status = 'active' + LIMIT 1 + """, + (session_id,), + ).fetchone() + if active_lease is not None: + return { + "outcome": "blocked", + "reason": "live_lease", + "prior_status": prior_status, + "new_status": prior_status, + "details": { + "session_id": session_id, + "lease_id": active_lease["lease_id"], + "blocker": "active_lease_row", + }, + } + + cols = { + r[1] for r in conn.execute("PRAGMA table_info(sessions)").fetchall() + } + if "retired_at" in cols and "retire_reason" in cols: + conn.execute( + """ + UPDATE sessions + SET status = ?, retired_at = ?, retire_reason = ? + WHERE session_id = ? AND status = ? + """, + (term, moment, reason_s, session_id, prior.get("status")), + ) + else: + conn.execute( + """ + UPDATE sessions + SET status = ? + WHERE session_id = ? AND status = ? + """, + (term, session_id, prior.get("status")), + ) + changed = conn.execute( + "SELECT changes()" + ).fetchone()[0] + if not changed: + # Lost CAS race — re-read. + refreshed = conn.execute( + "SELECT status FROM sessions WHERE session_id = ?", + (session_id,), + ).fetchone() + cur = ( + str(refreshed["status"]).strip().lower() + if refreshed is not None + else None + ) + return { + "outcome": "already_terminal" + if cur in terminal_set + else "blocked", + "reason": reason_s, + "prior_status": prior_status, + "new_status": cur, + "details": { + "session_id": session_id, + "cas_lost": True, + }, + } + + detail_payload: dict[str, Any] = { + "session_id": session_id, + "prior_status": prior_status, + "new_status": term, + "reason": reason_s, + "actor_session_id": actor_session_id, + "pid": prior.get("pid"), + "role": prior.get("role"), + "profile": prior.get("profile"), + } + if isinstance(details, Mapping): + for key, value in details.items(): + if key not in detail_payload: + detail_payload[key] = value + message = ( + f"session {session_id} retired reason={reason_s} " + f"prior_status={prior_status}" + ) + # events.work_item_id is nullable; session retirement is not work-scoped. + conn.execute( + """ + INSERT INTO events(work_item_id, event_type, message, created_at) + VALUES (NULL, 'session_retired', ?, ?) + """, + (message[:2000], moment), + ) + # Also persist a structured JSON line in the message when short enough + # by appending a compact summary (full detail stays in return value / + # audit log; events.message is human-readable). + return { + "outcome": "retired", + "reason": reason_s, + "prior_status": prior_status, + "new_status": term, + "details": detail_payload, + } + def list_sessions( self, *, @@ -1639,6 +1835,29 @@ class ControlPlaneDB: if name not in cols: conn.execute(f"ALTER TABLE sessions ADD COLUMN {name} {decl}") + _SESSION_LIFECYCLE_COLUMNS: tuple[tuple[str, str], ...] = ( + ("owner_process_started_at", "TEXT"), + ("retired_at", "TEXT"), + ("retire_reason", "TEXT"), + ) + + def _migrate_session_lifecycle_columns(self, conn: sqlite3.Connection) -> None: + """Add session retirement / PID-reuse provenance columns (#969). + + Additive and idempotent. Pre-existing rows migrate with NULL; retirement + fills ``retired_at`` / ``retire_reason``, and new upserts may record + ``owner_process_started_at`` for stronger identity checks. + """ + cols = { + row[1] + for row in conn.execute("PRAGMA table_info(sessions)").fetchall() + } + if not cols: + return + for name, decl in self._SESSION_LIFECYCLE_COLUMNS: + if name not in cols: + conn.execute(f"ALTER TABLE sessions ADD COLUMN {name} {decl}") + def list_active_claims( self, *, diff --git a/docs/post-restart-reconcile.md b/docs/post-restart-reconcile.md index 42511b3..9c1b8a2 100644 --- a/docs/post-restart-reconcile.md +++ b/docs/post-restart-reconcile.md @@ -19,7 +19,11 @@ The assessor classifies: - **service_health** — process healthy / parity mutation-safe - **clients** — connected client descriptors (optional inventory) -- **sessions** — active session rows with dead owner pids are unresolved +- **sessions** — active session rows with dead or reused owner pids are + unresolved until retired via `#969` (`session_lifecycle` / + `apply_session_cleanup=true` on `gitea_reconcile_after_restart`, or + `gitea_retire_stale_workflow_sessions`). Live owners, live leases, and live + client-managed sessions are never retired. - **checkpoints** — soft-depends on #660; skipped with reason when schema absent - **leases** — live control-plane leases after restart - **capabilities** — master-parity / stale-runtime (#610) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 532b8be..d623966 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -24432,10 +24432,18 @@ def _run_post_restart_reconcile( repo: str | None = None, mode: str | None = None, limit: int = 200, + apply_session_cleanup: bool = False, ) -> dict: - """Gather + classify post-restart state; cache the latest proof (#662).""" + """Gather + classify post-restart state; cache the latest proof (#662 / #969). + + When *apply_session_cleanup* is true, confirmed-stale session rows (dead + owner / PID reuse, no live lease) are terminalized through + ``session_lifecycle`` before re-classification so the sessions dimension + can resolve. Default remains dry (read-only) to preserve #662 rollout. + """ global _POST_RESTART_LAST_PROOF, _POST_RESTART_BOOT_RAN import post_restart_reconcile as prr + import session_lifecycle as sl try: _h, o, r = _resolve(remote, None, org, repo) @@ -24449,13 +24457,48 @@ def _run_post_restart_reconcile( inventory = _gather_post_restart_inventory( remote=remote, org=o, repo=r, limit=limit ) + + session_cleanup: dict | None = None + if apply_session_cleanup: + db, db_errs = _control_plane_db_or_error() + if db is None: + session_cleanup = { + "success": False, + "reasons": db_errs + or ["control-plane DB unavailable; cannot retire sessions"], + } + else: + profile = get_profile() + profile_name = (profile.get("profile_name") or "").strip() or "session" + actor = f"{profile_name}-{os.getpid()}" + session_cleanup = sl.retire_stale_sessions( + db, + sessions=inventory.get("sessions") or [], + leases=inventory.get("leases") or [], + dry_run=False, + actor_session_id=actor, + session_limit=max(1, int(limit)), + ) + # Re-gather active sessions after mutation so classification sees + # the post-retirement fleet (retired rows drop out of active list). + try: + inventory["sessions"] = db.list_sessions( + statuses=("active",), limit=max(1, int(limit)) + ) + except Exception as exc: # noqa: BLE001 + inventory["incomplete_reasons"] = list( + inventory.get("incomplete_reasons") or [] + ) + [f"post-retirement session re-list failed: {_redact(str(exc))}"] + fleet = (session_cleanup.get("fleet") or {}) if session_cleanup else {} + inventory["session_fleet"] = fleet + proof = prr.reconcile_after_restart( inventory, mode=mode or _post_restart_reconcile_mode(), ) payload = proof.as_dict() payload["success"] = True - payload["read_only"] = True + payload["read_only"] = not apply_session_cleanup payload["remote"] = remote payload["org"] = o payload["repo"] = r @@ -24464,6 +24507,9 @@ def _run_post_restart_reconcile( "proposed_follow_ups lists durable issues the apply path may create; " "this tool never creates them (log-only by default, #662 rollout)" ) + if session_cleanup is not None: + payload["session_cleanup"] = session_cleanup + payload["apply_session_cleanup"] = bool(apply_session_cleanup) _POST_RESTART_LAST_PROOF = payload _POST_RESTART_BOOT_RAN = True return payload @@ -24490,8 +24536,9 @@ def gitea_reconcile_after_restart( repo: str | None = None, mode: str | None = None, limit: int = 200, + apply_session_cleanup: bool = False, ) -> dict: - """Run post-restart MCP reconciliation and return a completion proof (#662). + """Run post-restart MCP reconciliation and return a completion proof (#662 / #969). Gathers live control-plane sessions, leases, worktree bindings, and master-parity evidence, then classifies them with the pure @@ -24499,12 +24546,17 @@ def gitea_reconcile_after_restart( machine-readable completion proof listing resolved / unresolved dimensions and proposed durable follow-up issues. - Read-only by design: never restarts MCP, never auto-resumes write - mutations, and never creates Gitea issues (those are a separate apply - path). Default mode is ``log_only``; set + Never restarts MCP, never auto-resumes write mutations, and never creates + Gitea issues. Default mode is ``log_only``; set ``GITEA_POST_RESTART_RECONCILE_MODE=enforce`` (or pass ``mode='enforce'``) to set ``mutation_hold`` when anything remains unresolved. + *apply_session_cleanup* (#969): when true, confirmed-stale workflow session + rows (dead owner PID / PID reuse, no live lease, not a live client-managed + owner) are terminalized through the sanctioned ``session_lifecycle`` path + before re-classification. Default false preserves the historical read-only + gather+classify behaviour. Does not delete historical rows. + Soft-depends on #660 for session checkpoints: when the checkpoint schema module is absent the checkpoints dimension is ``skipped`` with an explicit reason rather than inventing a schema. @@ -24524,9 +24576,96 @@ def gitea_reconcile_after_restart( repo=repo, mode=mode, limit=limit, + apply_session_cleanup=bool(apply_session_cleanup), ) +@mcp.tool() +def gitea_retire_stale_workflow_sessions( + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + apply: bool = False, + limit: int = 500, +) -> dict: + """Retire workflow session rows whose owners are no longer live (#969). + + Plans (and optionally applies) terminalization of control-plane session + rows that are confirmed stale: + + * owner PID absent / dead + * PID reused by an unrelated process (process start after session start) + * no live workflow lease + * not a live client-managed owner + + Default ``apply=false`` is dry-run only. ``apply=true`` performs CAS + status updates to ``retired`` and writes durable ``session_retired`` + events. Idempotent under concurrent reconciliation. Never deletes rows + and never touches sessions protected by a live lease or live owner. + """ + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "read_only": not apply, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + + try: + _h, o, r = _resolve(remote, None, org, repo) + except ValueError as exc: + return {"success": False, "reasons": [str(exc)], "read_only": not apply} + + db, db_errs = _control_plane_db_or_error() + if db is None: + return { + "success": False, + "read_only": not apply, + "reasons": db_errs or ["control-plane DB unavailable"], + } + + import session_lifecycle as sl + + profile = get_profile() + profile_name = (profile.get("profile_name") or "").strip() or "session" + actor = f"{profile_name}-{os.getpid()}" + + # Prefer lease inventory scoped to the requested repo when available. + leases: list[dict] = [] + 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=True, + limit=max(1, int(limit)), + ) + leases = list(lease_result.get("leases") or []) + except Exception: # noqa: BLE001 + try: + leases = db.list_leases(statuses=("active",), limit=max(1, int(limit))) + except Exception: # noqa: BLE001 + leases = [] + + result = sl.retire_stale_sessions( + db, + leases=leases, + dry_run=not bool(apply), + actor_session_id=actor, + session_limit=max(1, int(limit)), + ) + result["read_only"] = not bool(apply) + result["apply"] = bool(apply) + result["remote"] = remote + result["org"] = o + result["repo"] = r + return result + + @mcp.tool() def gitea_inspect_workflow_lease( lease_id: str, diff --git a/post_restart_reconcile.py b/post_restart_reconcile.py index 6477197..6d2f500 100644 --- a/post_restart_reconcile.py +++ b/post_restart_reconcile.py @@ -475,24 +475,65 @@ def reconcile_after_restart( ) ) - # --- sessions ------------------------------------------------------- + # --- sessions (#969: dead-owner / PID-reuse lifecycle) --------------- + # Prefer a precomputed fleet report from the gather/apply path when present; + # otherwise classify pure from inventory (injectable checkers stay default). sessions = [s for s in (inventory.get("sessions") or []) if isinstance(s, Mapping)] - orphan_sessions = [ - s - for s in sessions - if str(s.get("status") or "").lower() == "active" - and s.get("pid") is not None - and not lease_lifecycle.is_process_alive(s.get("pid")) + leases_for_sessions = [ + L for L in (inventory.get("leases") or []) if isinstance(L, Mapping) ] - if orphan_sessions: + fleet_report = inventory.get("session_fleet") + if isinstance(fleet_report, Mapping) and "retireable_session_ids" in fleet_report: + fleet_details = dict(fleet_report) + retireable_ids = list(fleet_details.get("retireable_session_ids") or []) + resolved = bool(fleet_details.get("sessions_dimension_resolved", not retireable_ids)) + else: + try: + import session_lifecycle as _sl + + client_managed = inventory.get("client_managed_session_ids") + cm_set = None + if isinstance(client_managed, (list, tuple, set, frozenset)): + cm_set = {str(x) for x in client_managed} + fleet = _sl.classify_sessions( + sessions, + leases=leases_for_sessions, + now=started, + client_managed_sessions=cm_set, + ) + fleet_details = fleet.as_dict() + retireable_ids = list(fleet_details.get("retireable_session_ids") or []) + resolved = bool(fleet_details.get("sessions_dimension_resolved")) + except Exception as exc: # noqa: BLE001 — fail closed to legacy signal + # Legacy fallback: dead-pid active rows only (pre-#969 behaviour). + orphan_sessions = [ + s + for s in sessions + if str(s.get("status") or "").lower() == "active" + and s.get("pid") is not None + and not lease_lifecycle.is_process_alive(s.get("pid")) + ] + retireable_ids = [s.get("session_id") for s in orphan_sessions] + resolved = not orphan_sessions + fleet_details = { + "total_sessions": len(sessions), + "retireable_session_ids": retireable_ids, + "legacy_fallback": True, + "fallback_error": str(exc), + } + + if not resolved and retireable_ids: items.append( _item( DIM_SESSIONS, ITEM_UNRESOLVED, - f"{len(orphan_sessions)} active session row(s) with dead owner pid", + f"{len(retireable_ids)} session row(s) with dead/reused owner " + f"await retirement", details={ - "orphan_session_ids": [s.get("session_id") for s in orphan_sessions], + "orphan_session_ids": retireable_ids, + "retireable_session_ids": retireable_ids, "total_sessions": len(sessions), + "fleet": fleet_details, }, follow_up=True, ) @@ -502,8 +543,12 @@ def reconcile_after_restart( _item( DIM_SESSIONS, ITEM_RESOLVED, - f"{len(sessions)} session row(s) reconciled (no dead-pid orphans)", - details={"total_sessions": len(sessions)}, + f"{len(sessions)} session row(s) reconciled " + f"(no retireable dead/reused owners)", + details={ + "total_sessions": len(sessions), + "fleet": fleet_details, + }, ) ) diff --git a/session_lifecycle.py b/session_lifecycle.py new file mode 100644 index 0000000..8bcf07c --- /dev/null +++ b/session_lifecycle.py @@ -0,0 +1,703 @@ +"""Safe lifecycle for workflow session rows with dead or reused owners (#969). + +Post-restart reconciliation previously left hundreds of ``active`` session rows +with dead owner PIDs permanently unresolved. A PID existence check alone is not +enough: operating systems reuse PIDs, so an unrelated live process can appear to +own a historical session. + +This module is the pure classification + apply core for session retirement: + +* distinguish live, disconnected, stale, protected, and terminal records +* refuse retirement when a live lease or live verified owner remains +* protect live client-managed sessions +* detect PID reuse via process start time vs session start / heartbeat +* terminalize confirmed-stale rows idempotently with durable audit events +* stay safe under concurrent reconciles (CAS on status) + +Design mirrors ``lease_lifecycle`` / ``post_restart_reconcile``: + +* Pure classification accepts injectable checkers so unit tests never touch + real processes. +* Apply mutations go only through :meth:`ControlPlaneDB.retire_session`. +* Historical rows are never deleted; status moves to a terminal value and an + events-row records the action + reason. +""" + +from __future__ import annotations + +import json +import os +import subprocess +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Any, Callable, Mapping, Sequence + +import control_plane_db as cpd +import lease_lifecycle + +# Session status vocabulary. +SESSION_STATUS_ACTIVE = "active" +SESSION_STATUS_RETIRED = "retired" +SESSION_STATUS_ENDED = "ended" +SESSION_STATUS_TERMINAL = "terminal" + +TERMINAL_SESSION_STATUSES = frozenset( + { + SESSION_STATUS_RETIRED, + SESSION_STATUS_ENDED, + SESSION_STATUS_TERMINAL, + "dead", + "stale", + "orphaned", + } +) + +# Classification outcomes for one session row. +CLASS_LIVE = "live" +CLASS_DISCONNECTED = "disconnected" +CLASS_STALE = "stale" +CLASS_PROTECTED = "protected" +CLASS_TERMINAL = "terminal" + +# Stable reason codes (audit + tests). +REASON_ALREADY_TERMINAL = "already_terminal" +REASON_LIVE_OWNER = "live_owner" +REASON_LIVE_LEASE = "live_lease" +REASON_CLIENT_MANAGED_LIVE = "client_managed_live" +REASON_DEAD_OWNER = "dead_owner" +REASON_PID_REUSE = "pid_reuse" +REASON_HEARTBEAT_STALE_DEAD = "heartbeat_stale_dead_owner" +REASON_MISSING_PID = "missing_pid_no_lease" + +# Event type written to control-plane events table. +EVENT_SESSION_RETIRED = "session_retired" + +# Default heartbeat window before a still-alive PID is treated as disconnected +# rather than live (does not alone authorize retirement). +DEFAULT_HEARTBEAT_STALE_SECONDS = 900 + +# PID reuse: process start must be strictly later than session started_at by +# more than this skew (ps lstart is second-resolution; clocks can lag). +PID_REUSE_SKEW = timedelta(seconds=2) + +# Live lease freshness values that block retirement. +_LIVE_LEASE_FRESHNESS = frozenset({"active", "live"}) + + +class SessionLifecycleError(RuntimeError): + """Fail-closed session lifecycle policy error.""" + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _parse_ts(value: str | datetime | None) -> datetime | None: + if value is None: + return None + if isinstance(value, datetime): + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + return cpd._parse_ts(str(value)) + + +def _ts(dt: datetime | None = None) -> str: + return cpd._ts(dt) + + +def process_start_time(pid: int | None) -> datetime | None: + """Return the OS start time of *pid*, or None when it cannot be resolved. + + Uses ``ps -o lstart=`` (POSIX). Failures return None rather than inventing + evidence — missing start time never authorizes retirement of a live PID. + """ + if pid is None: + return None + try: + pid_i = int(pid) + except (TypeError, ValueError): + return None + if pid_i <= 0: + return None + try: + proc = subprocess.run( + ["ps", "-o", "lstart=", "-p", str(pid_i)], + capture_output=True, + text=True, + check=False, + timeout=2, + ) + except (OSError, subprocess.SubprocessError): + return None + if proc.returncode != 0: + return None + text = (proc.stdout or "").strip() + if not text: + return None + try: + # Example: "Wed Jul 29 09:14:36 2026" + naive = datetime.strptime(text, "%a %b %d %H:%M:%S %Y") + return naive.replace(tzinfo=timezone.utc) + except ValueError: + return None + + +def _lease_freshness_label(lease: Mapping[str, Any]) -> str: + fr = lease.get("freshness") + if isinstance(fr, Mapping): + return str(fr.get("freshness") or "").strip().lower() + if fr: + return str(fr).strip().lower() + # Fall back to classifying a raw lease row. + try: + return str( + lease_lifecycle.classify_lease_freshness(lease).get("freshness") or "" + ).strip().lower() + except Exception: # noqa: BLE001 — pure classifier must not raise on bad rows + status = str(lease.get("status") or "").strip().lower() + return status or "unknown" + + +def live_lease_session_ids( + leases: Sequence[Mapping[str, Any]] | None, + *, + now: datetime | None = None, + pid_checker: Callable[[int | None], bool] = lease_lifecycle.is_process_alive, +) -> set[str]: + """Session ids that still hold a live (or ambiguous-active) workflow lease.""" + live: set[str] = set() + moment = now or _utc_now() + for lease in leases or (): + if not isinstance(lease, Mapping): + continue + status = str(lease.get("status") or "").strip().lower() + if status and status not in { + lease_lifecycle.LEASE_STATUS_ACTIVE, + "", + }: + # Explicit terminal lease statuses never protect a session. + if status in { + lease_lifecycle.LEASE_STATUS_RELEASED, + lease_lifecycle.LEASE_STATUS_EXPIRED, + lease_lifecycle.LEASE_STATUS_ABANDONED, + }: + continue + freshness = _lease_freshness_label(lease) + if freshness in _LIVE_LEASE_FRESHNESS or freshness in {"", "unknown"}: + # Ambiguous active rows: re-check with authoritative classifier. + try: + fr = lease_lifecycle.classify_lease_freshness( + lease, now=moment, pid_checker=pid_checker + ) + freshness = str(fr.get("freshness") or "").strip().lower() + except Exception: # noqa: BLE001 + freshness = "unknown" + if freshness in _LIVE_LEASE_FRESHNESS: + sid = str(lease.get("session_id") or "").strip() + if sid: + live.add(sid) + elif freshness == "unknown" and status in { + lease_lifecycle.LEASE_STATUS_ACTIVE, + "", + }: + # Fail closed: active lease with unknown freshness blocks retirement. + sid = str(lease.get("session_id") or "").strip() + if sid: + live.add(sid) + return live + + +@dataclass(frozen=True) +class SessionClassification: + """Classification of one workflow session row.""" + + session_id: str + classification: str + reason: str + retireable: bool + status: str | None + pid: int | None + pid_alive: bool | None + pid_reused: bool + heartbeat_stale: bool + has_live_lease: bool + client_managed: bool + details: dict[str, Any] = field(default_factory=dict) + + def as_dict(self) -> dict[str, Any]: + return { + "session_id": self.session_id, + "classification": self.classification, + "reason": self.reason, + "retireable": self.retireable, + "status": self.status, + "pid": self.pid, + "pid_alive": self.pid_alive, + "pid_reused": self.pid_reused, + "heartbeat_stale": self.heartbeat_stale, + "has_live_lease": self.has_live_lease, + "client_managed": self.client_managed, + "details": dict(self.details), + } + + +def classify_session( + row: Mapping[str, Any], + *, + now: datetime | None = None, + pid_checker: Callable[[int | None], bool] = lease_lifecycle.is_process_alive, + process_start_probe: Callable[[int | None], datetime | None] = process_start_time, + live_lease_sessions: set[str] | frozenset[str] | None = None, + client_managed_sessions: set[str] | frozenset[str] | None = None, + heartbeat_stale_seconds: int = DEFAULT_HEARTBEAT_STALE_SECONDS, +) -> SessionClassification: + """Classify one session row for retirement decisions (#969). + + Rules (first match wins where noted): + + 1. Non-active / already-terminal status → ``terminal`` (not retireable). + 2. Session holds a live lease → ``protected`` (never retire). + 3. PID missing and no live lease → ``stale`` (retireable: missing owner). + 4. PID alive + process start after session start → ``stale`` (PID reuse). + 5. PID alive + client-managed → ``live`` protected (never retire). + 6. PID alive + fresh heartbeat → ``live``. + 7. PID alive + stale heartbeat → ``disconnected`` (not retireable alone). + 8. PID dead → ``stale`` (retireable). + """ + moment = now or _utc_now() + session_id = str(row.get("session_id") or "").strip() + status = str(row.get("status") or "").strip().lower() or None + raw_pid = row.get("pid") + try: + pid = int(raw_pid) if raw_pid is not None else None + except (TypeError, ValueError): + pid = None + + client_managed = bool( + row.get("client_managed") + or row.get("is_client_managed") + or ( + client_managed_sessions is not None + and session_id in client_managed_sessions + ) + ) + has_live_lease = bool( + live_lease_sessions is not None and session_id in live_lease_sessions + ) + + hb = _parse_ts(row.get("last_heartbeat_at")) + heartbeat_stale = bool( + hb is not None + and (moment - hb).total_seconds() > max(0, int(heartbeat_stale_seconds)) + ) + if hb is None and status == SESSION_STATUS_ACTIVE: + # No heartbeat evidence: treat as stale for liveness bookkeeping only. + heartbeat_stale = True + + started = _parse_ts(row.get("started_at")) + recorded_proc_start = _parse_ts(row.get("owner_process_started_at")) + + if status in TERMINAL_SESSION_STATUSES: + return SessionClassification( + session_id=session_id, + classification=CLASS_TERMINAL, + reason=REASON_ALREADY_TERMINAL, + retireable=False, + status=status, + pid=pid, + pid_alive=None, + pid_reused=False, + heartbeat_stale=heartbeat_stale, + has_live_lease=has_live_lease, + client_managed=client_managed, + ) + + if has_live_lease: + return SessionClassification( + session_id=session_id, + classification=CLASS_PROTECTED, + reason=REASON_LIVE_LEASE, + retireable=False, + status=status, + pid=pid, + pid_alive=pid_checker(pid) if pid is not None else None, + pid_reused=False, + heartbeat_stale=heartbeat_stale, + has_live_lease=True, + client_managed=client_managed, + details={"blocker": "live_workflow_lease"}, + ) + + if pid is None: + return SessionClassification( + session_id=session_id, + classification=CLASS_STALE, + reason=REASON_MISSING_PID, + retireable=True, + status=status, + pid=None, + pid_alive=False, + pid_reused=False, + heartbeat_stale=heartbeat_stale, + has_live_lease=False, + client_managed=client_managed, + ) + + pid_alive = bool(pid_checker(pid)) + pid_reused = False + proc_start: datetime | None = None + + if pid_alive: + proc_start = recorded_proc_start or process_start_probe(pid) + # PID reuse: live process started after the session row itself was + # created. Anchor on started_at only — last_heartbeat alone is not a + # safe bound (synthetic inventories and long-lived processes would + # false-positive against a live ``ps`` probe). + if proc_start is not None and started is not None: + if proc_start > (started + PID_REUSE_SKEW): + pid_reused = True + + if pid_reused: + return SessionClassification( + session_id=session_id, + classification=CLASS_STALE, + reason=REASON_PID_REUSE, + retireable=True, + status=status, + pid=pid, + pid_alive=True, + pid_reused=True, + heartbeat_stale=heartbeat_stale, + has_live_lease=False, + client_managed=client_managed, + details={ + "process_started_at": _ts(proc_start) if proc_start else None, + "session_started_at": row.get("started_at"), + "last_heartbeat_at": row.get("last_heartbeat_at"), + }, + ) + + if pid_alive and client_managed: + return SessionClassification( + session_id=session_id, + classification=CLASS_LIVE, + reason=REASON_CLIENT_MANAGED_LIVE, + retireable=False, + status=status, + pid=pid, + pid_alive=True, + pid_reused=False, + heartbeat_stale=heartbeat_stale, + has_live_lease=False, + client_managed=True, + ) + + if pid_alive and not heartbeat_stale: + return SessionClassification( + session_id=session_id, + classification=CLASS_LIVE, + reason=REASON_LIVE_OWNER, + retireable=False, + status=status, + pid=pid, + pid_alive=True, + pid_reused=False, + heartbeat_stale=False, + has_live_lease=False, + client_managed=client_managed, + ) + + if pid_alive and heartbeat_stale: + # Process still exists but has not heartbeated — disconnected, not + # confirmed stale. Do not retire; operator/reconnect owns next step. + return SessionClassification( + session_id=session_id, + classification=CLASS_DISCONNECTED, + reason=REASON_LIVE_OWNER, + retireable=False, + status=status, + pid=pid, + pid_alive=True, + pid_reused=False, + heartbeat_stale=True, + has_live_lease=False, + client_managed=client_managed, + details={"note": "alive_pid_stale_heartbeat_not_retired"}, + ) + + # PID dead (or checker said not alive). + reason = REASON_DEAD_OWNER + if heartbeat_stale: + reason = REASON_HEARTBEAT_STALE_DEAD + return SessionClassification( + session_id=session_id, + classification=CLASS_STALE, + reason=reason, + retireable=True, + status=status, + pid=pid, + pid_alive=False, + pid_reused=False, + heartbeat_stale=heartbeat_stale, + has_live_lease=False, + client_managed=client_managed, + ) + + +@dataclass(frozen=True) +class SessionFleetReport: + """Fleet-wide classification summary for reconcile + apply.""" + + classifications: tuple[SessionClassification, ...] + live_count: int + disconnected_count: int + stale_count: int + protected_count: int + terminal_count: int + retireable: tuple[SessionClassification, ...] + counts_by_reason: dict[str, int] + + def as_dict(self) -> dict[str, Any]: + return { + "live_count": self.live_count, + "disconnected_count": self.disconnected_count, + "stale_count": self.stale_count, + "protected_count": self.protected_count, + "terminal_count": self.terminal_count, + "retireable_count": len(self.retireable), + "retireable_session_ids": [c.session_id for c in self.retireable], + "counts_by_reason": dict(self.counts_by_reason), + "classifications": [c.as_dict() for c in self.classifications], + "sessions_dimension_resolved": len(self.retireable) == 0, + } + + +def classify_sessions( + sessions: Sequence[Mapping[str, Any]] | None, + *, + leases: Sequence[Mapping[str, Any]] | None = None, + now: datetime | None = None, + pid_checker: Callable[[int | None], bool] = lease_lifecycle.is_process_alive, + process_start_probe: Callable[[int | None], datetime | None] = process_start_time, + client_managed_sessions: set[str] | frozenset[str] | None = None, + heartbeat_stale_seconds: int = DEFAULT_HEARTBEAT_STALE_SECONDS, +) -> SessionFleetReport: + """Classify a fleet of session rows against live leases (#969).""" + moment = now or _utc_now() + live_leases = live_lease_session_ids( + leases, now=moment, pid_checker=pid_checker + ) + results: list[SessionClassification] = [] + for row in sessions or (): + if not isinstance(row, Mapping): + continue + results.append( + classify_session( + row, + now=moment, + pid_checker=pid_checker, + process_start_probe=process_start_probe, + live_lease_sessions=live_leases, + client_managed_sessions=client_managed_sessions, + heartbeat_stale_seconds=heartbeat_stale_seconds, + ) + ) + + live_count = sum(1 for c in results if c.classification == CLASS_LIVE) + disconnected_count = sum( + 1 for c in results if c.classification == CLASS_DISCONNECTED + ) + stale_count = sum(1 for c in results if c.classification == CLASS_STALE) + protected_count = sum(1 for c in results if c.classification == CLASS_PROTECTED) + terminal_count = sum(1 for c in results if c.classification == CLASS_TERMINAL) + retireable = tuple(c for c in results if c.retireable) + by_reason: dict[str, int] = {} + for c in results: + by_reason[c.reason] = by_reason.get(c.reason, 0) + 1 + + return SessionFleetReport( + classifications=tuple(results), + live_count=live_count, + disconnected_count=disconnected_count, + stale_count=stale_count, + protected_count=protected_count, + terminal_count=terminal_count, + retireable=retireable, + counts_by_reason=by_reason, + ) + + +@dataclass(frozen=True) +class RetirementResult: + """Outcome of one session retirement attempt.""" + + session_id: str + outcome: str # retired | already_terminal | skipped | blocked | missing + reason: str + prior_status: str | None = None + new_status: str | None = None + details: dict[str, Any] = field(default_factory=dict) + + def as_dict(self) -> dict[str, Any]: + return { + "session_id": self.session_id, + "outcome": self.outcome, + "reason": self.reason, + "prior_status": self.prior_status, + "new_status": self.new_status, + "details": dict(self.details), + } + + +def apply_session_retirements( + db: cpd.ControlPlaneDB, + report: SessionFleetReport, + *, + dry_run: bool = False, + actor_session_id: str | None = None, + now: datetime | None = None, +) -> dict[str, Any]: + """Terminalize every retireable session in *report* (idempotent). + + Concurrent reconciles are safe: each retirement is a CAS on + ``status='active'`` (or other non-terminal). A second pass that sees the + same session already retired records ``already_terminal`` rather than + duplicating audit noise beyond a single no-op outcome. + """ + moment = now or _utc_now() + results: list[RetirementResult] = [] + retired = 0 + already = 0 + blocked = 0 + missing = 0 + + for classification in report.retireable: + if not classification.retireable: + continue + if dry_run: + results.append( + RetirementResult( + session_id=classification.session_id, + outcome="skipped", + reason=classification.reason, + prior_status=classification.status, + new_status=SESSION_STATUS_RETIRED, + details={"dry_run": True, **classification.details}, + ) + ) + continue + + applied = db.retire_session( + session_id=classification.session_id, + reason=classification.reason, + actor_session_id=actor_session_id, + details={ + "classification": classification.classification, + "pid": classification.pid, + "pid_alive": classification.pid_alive, + "pid_reused": classification.pid_reused, + "client_managed": classification.client_managed, + **classification.details, + }, + now=moment, + ) + outcome = str(applied.get("outcome") or "missing") + results.append( + RetirementResult( + session_id=classification.session_id, + outcome=outcome, + reason=str(applied.get("reason") or classification.reason), + prior_status=applied.get("prior_status"), + new_status=applied.get("new_status"), + details=dict(applied.get("details") or {}), + ) + ) + if outcome == "retired": + retired += 1 + elif outcome == "already_terminal": + already += 1 + elif outcome == "blocked": + blocked += 1 + else: + missing += 1 + + # Non-retireable classifications are recorded for audit completeness when + # dry-run lists the fleet, but apply only mutates retireable rows. + return { + "success": True, + "dry_run": dry_run, + "retired_count": retired, + "already_terminal_count": already, + "blocked_count": blocked, + "missing_count": missing, + "planned_count": len(report.retireable), + "results": [r.as_dict() for r in results], + "audit_action": EVENT_SESSION_RETIRED, + "actor_session_id": actor_session_id, + "recorded_at": _ts(moment), + } + + +def retire_stale_sessions( + db: cpd.ControlPlaneDB, + *, + sessions: Sequence[Mapping[str, Any]] | None = None, + leases: Sequence[Mapping[str, Any]] | None = None, + dry_run: bool = False, + actor_session_id: str | None = None, + now: datetime | None = None, + pid_checker: Callable[[int | None], bool] = lease_lifecycle.is_process_alive, + process_start_probe: Callable[[int | None], datetime | None] = process_start_time, + client_managed_sessions: set[str] | frozenset[str] | None = None, + heartbeat_stale_seconds: int = DEFAULT_HEARTBEAT_STALE_SECONDS, + session_limit: int = 500, +) -> dict[str, Any]: + """End-to-end plan + apply for stale session retirement (#969). + + When *sessions* is omitted the control-plane DB is inventoried (active + rows only). Callers that already gathered inventory should pass it. + """ + moment = now or _utc_now() + if sessions is None: + sessions = db.list_sessions( + statuses=(SESSION_STATUS_ACTIVE,), + limit=max(1, int(session_limit)), + ) + if leases is None: + try: + leases = db.list_leases(statuses=("active",), limit=max(1, int(session_limit))) + except Exception: # noqa: BLE001 + leases = [] + + report = classify_sessions( + sessions, + leases=leases, + now=moment, + pid_checker=pid_checker, + process_start_probe=process_start_probe, + client_managed_sessions=client_managed_sessions, + heartbeat_stale_seconds=heartbeat_stale_seconds, + ) + apply_result = apply_session_retirements( + db, + report, + dry_run=dry_run, + actor_session_id=actor_session_id, + now=moment, + ) + return { + "success": True, + "fleet": report.as_dict(), + "apply": apply_result, + "sessions_dimension_resolved": ( + report.as_dict()["sessions_dimension_resolved"] + if dry_run + else apply_result["planned_count"] + == ( + apply_result["retired_count"] + + apply_result["already_terminal_count"] + ) + and apply_result["blocked_count"] == 0 + ), + } diff --git a/task_capability_map.py b/task_capability_map.py index 393ce98..499840a 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -153,8 +153,9 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.branch.push", "role": "author", }, - # #662: post-restart reconcile is read-only inventory + pure classification. - # Durable follow-up issue creation is a separate apply path (not this task). + # #662: post-restart reconcile is inventory + pure classification. + # #969: optional apply_session_cleanup retires confirmed-stale session rows + # through the same tool; durable follow-up Gitea issues remain separate. "reconcile_after_restart": { "permission": "gitea.read", "role": "author", @@ -163,6 +164,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.read", "role": "author", }, + # #969: explicit plan/apply path for dead-owner session retirement. + "retire_stale_workflow_sessions": { + "permission": "gitea.read", + "role": "author", + }, + "gitea_retire_stale_workflow_sessions": { + "permission": "gitea.read", + "role": "author", + }, # #644: Phase 2 Web Console recovery tasks. "clear_stale_binding": { "permission": "gitea.read", diff --git a/tests/test_control_plane_db.py b/tests/test_control_plane_db.py index 6065f38..b60852b 100644 --- a/tests/test_control_plane_db.py +++ b/tests/test_control_plane_db.py @@ -37,7 +37,7 @@ class ControlPlaneDBTest(unittest.TestCase): rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall()) finally: conn.close() - self.assertEqual(rows["schema_version"], "5") + self.assertEqual(rows["schema_version"], "6") self.assertIn("DB coordinates", rows["architecture"]) self.assertIn("bridge", rows["architecture"].lower()) @@ -868,7 +868,7 @@ class SessionCheckpointTest(unittest.TestCase): conn.close() self.assertIn("session_checkpoints", names) record = self._write() - self.assertEqual(record["checkpoint_schema_version"], 5) + self.assertEqual(record["checkpoint_schema_version"], 6) # AC2 — checkpoints written for multi-role session fixtures. def test_multi_role_fixtures_each_get_a_row(self) -> None: diff --git a/tests/test_issue_969_session_lifecycle.py b/tests/test_issue_969_session_lifecycle.py new file mode 100644 index 0000000..94fc0bd --- /dev/null +++ b/tests/test_issue_969_session_lifecycle.py @@ -0,0 +1,480 @@ +"""Tests for dead-owner / PID-reuse session retirement (#969).""" + +from __future__ import annotations + +import os +import tempfile +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timedelta, timezone + +import control_plane_db as cpd +import post_restart_reconcile as prr +import session_lifecycle as sl + +NOW = datetime(2026, 7, 29, 12, 0, 0, tzinfo=timezone.utc) +EARLIER = NOW - timedelta(hours=2) +LATER = NOW + timedelta(minutes=5) + + +def _session( + session_id: str, + *, + pid: int | None = 4242, + status: str = "active", + started_at: datetime = EARLIER, + last_heartbeat_at: datetime | None = None, + client_managed: bool = False, + owner_process_started_at: datetime | None = None, + role: str = "author", +) -> dict: + hb = last_heartbeat_at if last_heartbeat_at is not None else started_at + row = { + "session_id": session_id, + "role": role, + "profile": "prgs-author", + "pid": pid, + "status": status, + "started_at": cpd._ts(started_at), + "last_heartbeat_at": cpd._ts(hb), + "client_managed": client_managed, + } + if owner_process_started_at is not None: + row["owner_process_started_at"] = cpd._ts(owner_process_started_at) + return row + + +def _alive(pids: set[int]): + def _check(pid): + try: + return int(pid) in pids + except (TypeError, ValueError): + return False + + return _check + + +def _starts(mapping: dict[int, datetime]): + def _probe(pid): + try: + return mapping.get(int(pid)) + except (TypeError, ValueError): + return None + + return _probe + + +class ClassifyDeadOwnerTests(unittest.TestCase): + def test_dead_owner_is_stale_and_retireable(self) -> None: + c = sl.classify_session( + _session("ghost", pid=2_000_000_000), + now=NOW, + pid_checker=_alive(set()), + process_start_probe=_starts({}), + ) + self.assertEqual(c.classification, sl.CLASS_STALE) + self.assertTrue(c.retireable) + self.assertIn(c.reason, {sl.REASON_DEAD_OWNER, sl.REASON_HEARTBEAT_STALE_DEAD}) + + def test_missing_pid_is_stale(self) -> None: + c = sl.classify_session( + _session("no-pid", pid=None), + now=NOW, + pid_checker=_alive(set()), + ) + self.assertEqual(c.classification, sl.CLASS_STALE) + self.assertEqual(c.reason, sl.REASON_MISSING_PID) + self.assertTrue(c.retireable) + + +class PidReuseTests(unittest.TestCase): + def test_pid_reuse_marks_stale_not_live(self) -> None: + # Process with same PID started AFTER the session was recorded. + c = sl.classify_session( + _session("reused", pid=77, started_at=EARLIER, last_heartbeat_at=EARLIER), + now=NOW, + pid_checker=_alive({77}), + process_start_probe=_starts({77: LATER}), + ) + self.assertEqual(c.classification, sl.CLASS_STALE) + self.assertEqual(c.reason, sl.REASON_PID_REUSE) + self.assertTrue(c.pid_reused) + self.assertTrue(c.retireable) + + def test_matching_process_start_is_live(self) -> None: + c = sl.classify_session( + _session( + "same-proc", + pid=88, + started_at=EARLIER, + last_heartbeat_at=NOW - timedelta(seconds=30), + owner_process_started_at=EARLIER - timedelta(seconds=5), + ), + now=NOW, + pid_checker=_alive({88}), + process_start_probe=_starts({88: EARLIER - timedelta(seconds=5)}), + ) + self.assertEqual(c.classification, sl.CLASS_LIVE) + self.assertFalse(c.retireable) + + +class LiveOwnerAndLeaseTests(unittest.TestCase): + def test_live_owner_not_retired(self) -> None: + c = sl.classify_session( + _session( + "live", + pid=os.getpid(), + last_heartbeat_at=NOW - timedelta(seconds=10), + ), + now=NOW, + pid_checker=_alive({os.getpid()}), + process_start_probe=_starts({os.getpid(): EARLIER}), + ) + self.assertEqual(c.classification, sl.CLASS_LIVE) + self.assertFalse(c.retireable) + + def test_live_lease_blocks_retirement_even_if_pid_dead(self) -> None: + c = sl.classify_session( + _session("leased", pid=99999), + now=NOW, + pid_checker=_alive(set()), + live_lease_sessions={"leased"}, + ) + self.assertEqual(c.classification, sl.CLASS_PROTECTED) + self.assertEqual(c.reason, sl.REASON_LIVE_LEASE) + self.assertFalse(c.retireable) + + def test_client_managed_live_never_retired(self) -> None: + c = sl.classify_session( + _session( + "client", + pid=55, + client_managed=True, + last_heartbeat_at=NOW - timedelta(seconds=5), + ), + now=NOW, + pid_checker=_alive({55}), + process_start_probe=_starts({55: EARLIER}), + ) + self.assertEqual(c.classification, sl.CLASS_LIVE) + self.assertEqual(c.reason, sl.REASON_CLIENT_MANAGED_LIVE) + self.assertFalse(c.retireable) + + def test_client_managed_dead_pid_is_retireable(self) -> None: + # Dead client process is not a live client-managed session. + c = sl.classify_session( + _session("client-dead", pid=56, client_managed=True), + now=NOW, + pid_checker=_alive(set()), + ) + self.assertEqual(c.classification, sl.CLASS_STALE) + self.assertTrue(c.retireable) + + +class TerminalAndDisconnectedTests(unittest.TestCase): + def test_already_terminal_not_retireable(self) -> None: + c = sl.classify_session( + _session("done", status="retired", pid=1), + now=NOW, + pid_checker=_alive(set()), + ) + self.assertEqual(c.classification, sl.CLASS_TERMINAL) + self.assertFalse(c.retireable) + + def test_alive_stale_heartbeat_is_disconnected_not_retired(self) -> None: + c = sl.classify_session( + _session( + "quiet", + pid=66, + last_heartbeat_at=NOW - timedelta(hours=5), + ), + now=NOW, + pid_checker=_alive({66}), + process_start_probe=_starts({66: EARLIER}), + ) + self.assertEqual(c.classification, sl.CLASS_DISCONNECTED) + self.assertFalse(c.retireable) + + +class FleetAndApplyTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self._tmp.name, "cp.sqlite3") + self.db = cpd.ControlPlaneDB(self.db_path) + + def tearDown(self) -> None: + self._tmp.cleanup() + + def test_mixed_fleet_and_apply_retires_only_stale(self) -> None: + live_pid = os.getpid() + # Use wall-clock "now" so upsert timestamps align with classification. + moment = datetime.now(timezone.utc) + proc_start = moment - timedelta(hours=1) + self.db.upsert_session( + session_id="s-live", + role="author", + pid=live_pid, + status="active", + owner_process_started_at=cpd._ts(proc_start), + ) + self.db.upsert_session( + session_id="s-dead", role="reviewer", pid=2_000_000_001, status="active" + ) + self.db.upsert_session( + session_id="s-ended", role="merger", pid=3, status="ended" + ) + + sessions = self.db.list_sessions(limit=50) + report = sl.classify_sessions( + sessions, + leases=[], + now=moment, + pid_checker=_alive({live_pid}), + process_start_probe=_starts({live_pid: proc_start}), + ) + self.assertGreaterEqual(report.stale_count, 1) + self.assertTrue( + any(c.session_id == "s-dead" and c.retireable for c in report.classifications) + ) + self.assertTrue( + any( + c.session_id == "s-live" and not c.retireable + for c in report.classifications + ) + ) + + first = sl.apply_session_retirements( + self.db, report, dry_run=False, actor_session_id="actor-1", now=moment + ) + self.assertGreaterEqual(first["retired_count"], 1) + # After retirement, active list should exclude s-dead. + active = { + s["session_id"] + for s in self.db.list_sessions(statuses=("active",), limit=50) + } + self.assertNotIn("s-dead", active) + self.assertIn("s-live", active) + + # Repeated cleanup is idempotent. + report2 = sl.classify_sessions( + self.db.list_sessions(limit=50), + leases=[], + now=moment, + pid_checker=_alive({live_pid}), + process_start_probe=_starts({live_pid: proc_start}), + ) + second = sl.apply_session_retirements( + self.db, report2, dry_run=False, actor_session_id="actor-1", now=moment + ) + # No double-retirement of the same row as a new mutation. + self.assertEqual(second["retired_count"], 0) + + # Durable audit event present. + import sqlite3 + + conn = sqlite3.connect(self.db_path) + try: + events = conn.execute( + "SELECT event_type, message FROM events WHERE event_type = ?", + ("session_retired",), + ).fetchall() + finally: + conn.close() + self.assertTrue(events) + self.assertTrue(any("s-dead" in (m or "") for _, m in events)) + + def test_live_lease_blocks_db_retirement(self) -> None: + self.db.upsert_session( + session_id="s-leased", role="author", pid=2_000_000_002, status="active" + ) + self.db.upsert_work_item( + remote="prgs", + org="org", + repo="repo", + kind="issue", + number=969, + ) + result = self.db.assign_and_lease( + session_id="s-leased", + role="author", + remote="prgs", + org="org", + repo="repo", + kind="issue", + number=969, + ) + self.assertEqual(result.outcome, "assigned") + + # Inventory-style lease with explicit live freshness (authoritative for + # the pure classifier). DB apply also blocks on the active lease row. + leases = [ + { + "lease_id": result.lease_id, + "session_id": "s-leased", + "status": "active", + "freshness": {"freshness": "active"}, + } + ] + report = sl.classify_sessions( + self.db.list_sessions(statuses=("active",), limit=20), + leases=leases, + now=NOW, + pid_checker=_alive(set()), + ) + # Classifier protects via live lease set. + self.assertTrue( + any( + c.session_id == "s-leased" and c.classification == sl.CLASS_PROTECTED + for c in report.classifications + ) + ) + apply = sl.apply_session_retirements( + self.db, report, dry_run=False, actor_session_id="actor", now=NOW + ) + self.assertEqual(apply["retired_count"], 0) + active = { + s["session_id"] + for s in self.db.list_sessions(statuses=("active",), limit=20) + } + self.assertIn("s-leased", active) + + # Direct DB CAS also refuses while an active lease row remains. + blocked = self.db.retire_session( + session_id="s-leased", + reason=sl.REASON_DEAD_OWNER, + actor_session_id="actor", + now=NOW, + ) + self.assertEqual(blocked["outcome"], "blocked") + self.assertEqual(blocked["reason"], "live_lease") + + def test_concurrent_retirement_is_idempotent(self) -> None: + for i in range(20): + self.db.upsert_session( + session_id=f"ghost-{i}", + role="author", + pid=3_000_000 + i, + status="active", + ) + + def _worker() -> dict: + return sl.retire_stale_sessions( + self.db, + dry_run=False, + actor_session_id=f"actor-{threading.get_ident()}", + now=NOW, + pid_checker=_alive(set()), + process_start_probe=_starts({}), + session_limit=100, + ) + + outcomes = [] + with ThreadPoolExecutor(max_workers=4) as pool: + futs = [pool.submit(_worker) for _ in range(4)] + for fut in as_completed(futs): + outcomes.append(fut.result()) + + total_retired = sum(o["apply"]["retired_count"] for o in outcomes) + # Exactly one successful retirement per ghost row across all workers. + self.assertEqual(total_retired, 20) + active = { + s["session_id"] + for s in self.db.list_sessions(statuses=("active",), limit=100) + } + for i in range(20): + self.assertNotIn(f"ghost-{i}", active) + + +class ReconcileIntegrationTests(unittest.TestCase): + def test_unresolved_until_retired_then_resolved(self) -> None: + inv = { + "inventory_complete": True, + "incomplete_reasons": [], + "service_health": {"healthy": True}, + "clients": [{"session_id": "c1", "connected": True}], + "sessions": [ + _session("ghost", pid=2_000_000_099, last_heartbeat_at=EARLIER), + ], + "leases": [], + "checkpoints_available": False, + "worktree_bindings": [], + "pending_mutations": [], + "capabilities": {"stale": False}, + "boot_head_sha": "a" * 40, + "current_head_sha": "a" * 40, + "queue_state": {"safe_to_resume": True}, + } + proof = prr.reconcile_after_restart(inv, now=NOW, mode=prr.MODE_LOG_ONLY) + sess = next(i for i in proof.items if i.dimension == prr.DIM_SESSIONS) + self.assertEqual(sess.status, prr.ITEM_UNRESOLVED) + self.assertIn("ghost", sess.details.get("orphan_session_ids") or []) + + # After retirement inventory (no active orphans) resolves. + inv2 = dict(inv) + inv2["sessions"] = [] + inv2["session_fleet"] = { + "retireable_session_ids": [], + "sessions_dimension_resolved": True, + "live_count": 0, + "stale_count": 0, + } + proof2 = prr.reconcile_after_restart(inv2, now=NOW, mode=prr.MODE_LOG_ONLY) + sess2 = next(i for i in proof2.items if i.dimension == prr.DIM_SESSIONS) + self.assertEqual(sess2.status, prr.ITEM_RESOLVED) + + def test_legacy_orphan_key_still_populated(self) -> None: + inv = { + "inventory_complete": True, + "service_health": {"healthy": True}, + "clients": [], + "sessions": [_session("ghost", pid=2_000_000_100)], + "leases": [], + "checkpoints_available": False, + "worktree_bindings": [], + "pending_mutations": [], + "capabilities": {"stale": False}, + "boot_head_sha": "a" * 40, + "current_head_sha": "a" * 40, + "queue_state": {"safe_to_resume": True}, + } + proof = prr.reconcile_after_restart(inv, now=NOW) + sess = next(i for i in proof.items if i.dimension == prr.DIM_SESSIONS) + self.assertIn("orphan_session_ids", sess.details) + + +class SchemaMigrationTests(unittest.TestCase): + def test_lifecycle_columns_present(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "cp.sqlite3") + db = cpd.ControlPlaneDB(path) + db.upsert_session(session_id="s1", role="author", pid=1) + out = db.retire_session( + session_id="s1", + reason=sl.REASON_DEAD_OWNER, + actor_session_id="tester", + now=NOW, + ) + self.assertEqual(out["outcome"], "retired") + rows = db.list_sessions(limit=5) + # May not appear under active filter + all_rows = db.list_sessions(limit=5) + # Re-open raw to check columns + import sqlite3 + + conn = sqlite3.connect(path) + try: + cols = {r[1] for r in conn.execute("PRAGMA table_info(sessions)")} + version = conn.execute( + "SELECT value FROM schema_meta WHERE key='schema_version'" + ).fetchone()[0] + finally: + conn.close() + self.assertIn("retired_at", cols) + self.assertIn("retire_reason", cols) + self.assertIn("owner_process_started_at", cols) + self.assertEqual(version, "6") + + +if __name__ == "__main__": + unittest.main()