Compare commits

..
Author SHA1 Message Date
jcwalker3 64e6d7b7df Merge branch 'master' into fix/issue-872-dead-session-two-base-syncs 2026-07-24 15:04:57 -05:00
sysadmin 35714258f0 Merge pull request 'feat(webui): unified session/lease/lock/worktree inventory API (Closes #636)' (#838) from feat/issue-636-inventory-api into master 2026-07-24 13:54:23 -05:00
jcwalker3 4e269f3a7a Merge branch 'master' into feat/issue-636-inventory-api 2026-07-24 13:51:13 -05:00
sysadmin 36fe4785ec Merge pull request 'feat(mcp): post-restart reconciliation and completion proof (Closes #662)' (#879) from fix/issue-662-post-restart-reconcile into master 2026-07-24 09:04:23 -05:00
jcwalker3 1232789b41 Merge branch 'master' into fix/issue-662-post-restart-reconcile 2026-07-24 09:01:48 -05:00
sysadmin 2976c21ee6 Merge pull request 'test(#878): #628 building-block regression coverage (child of #628)' (#795) from feat/issue-628-autonomous-handoffs-orchestration into master 2026-07-24 08:51:56 -05:00
sysadminandClaude Opus 4.8 a7a283f449 feat(mcp): post-restart reconciliation and completion proof (Closes #662)
Add pure post_restart_reconcile.reconcile_after_restart classifier with a
machine-readable completion proof covering service health, sessions, leases,
capabilities, worktrees, interrupted mutations (never auto-resumed),
duplicates, and queue state. Soft-depends on #660 checkpoints (skipped with
reason when the schema module is absent).

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

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

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-24 08:42:20 -04:00
sysadmin e42756b27f fix(author-lock): dead-session recovery cross-session PR base-sync (Closes #872) 2026-07-24 08:40:19 -04:00
jcwalker3 f7ef719bd6 Merge branch 'master' into feat/issue-636-inventory-api 2026-07-24 07:36:54 -05:00
jcwalker3 cad5e44703 Merge branch 'master' into feat/issue-636-inventory-api 2026-07-24 07:08:19 -05:00
jcwalker3 c4d089f931 Merge branch 'master' into feat/issue-636-inventory-api 2026-07-24 06:55:53 -05:00
jcwalker3 baf3a474df Merge branch 'master' into feat/issue-636-inventory-api 2026-07-24 06:16:42 -05:00
sysadminandClaude Opus 4.8 784369cc25 merge(master): sync PR #838 with master after #875
No content conflicts. Master advanced with the #658 restart coordinator;
inventory API surfaces auto-merge cleanly.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-24 07:07:32 -04:00
sysadmin e0b87a0ae5 Merge remote-tracking branch 'prgs/master' into feat/issue-636-inventory-api 2026-07-24 07:01:45 -04:00
sysadmin 34173e079c Merge branch 'master' of https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools into feat/issue-636-inventory-api
# Conflicts:
#	docs/webui-local-dev.md
#	webui/app.py
2026-07-24 07:01:32 -04:00
sysadminandClaude Opus 4.8 b7a63a5579 feat(webui): unified session/lease/lock/worktree inventory API (Closes #636)
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-22 21:13:54 -04:00
11 changed files with 3219 additions and 15 deletions
+56
View File
@@ -0,0 +1,56 @@
# Post-restart MCP reconciliation (#662)
After an MCP process restart, sessions, leases, capabilities, worktrees, and
interrupted mutations must be reconciled before operators claim a clean runtime.
This document describes the #662 completion-proof path.
## Components
| Piece | Where | Responsibility |
|-------|-------|----------------|
| `post_restart_reconcile.reconcile_after_restart` | `post_restart_reconcile.py` | Pure classification: inventory → completion proof DTO. No I/O. |
| `RestartCompletionProof` | `post_restart_reconcile.py` | Machine-readable proof (`.as_dict()` is JSON-serializable). |
| `gitea_reconcile_after_restart` | `gitea_mcp_server.py` | MCP tool: gathers inventory from the #613 control-plane DB + master-parity, classifies, returns the proof. Read-only. |
| Boot hook | `gitea_assess_master_parity` | First post-restart parity probe also runs reconcile once (log-only by default). |
## Dimensions
The assessor classifies:
- **service_health** — process healthy / parity mutation-safe
- **clients** — connected client descriptors (optional inventory)
- **sessions** — active session rows with dead owner pids are unresolved
- **checkpoints** — soft-depends on #660; skipped with reason when schema absent
- **leases** — live control-plane leases after restart
- **capabilities** — master-parity / stale-runtime (#610)
- **worktrees** — lease-bound paths missing on disk
- **interrupted_mutations** — mutating lease phases or explicit pending inventory; **never auto-resumed**
- **duplicates** — multiple live claims on the same work item
- **queue** — allocator resume safety
## Modes
| Mode | Env / arg | Behavior |
|------|-----------|----------|
| `log_only` (default) | unset or `GITEA_POST_RESTART_RECONCILE_MODE=log_only` | Proof only; `mutation_hold=false` |
| `enforce` | `GITEA_POST_RESTART_RECONCILE_MODE=enforce` or `mode=enforce` | Sets `mutation_hold=true` when overall status is degraded/failed or interrupted mutations remain |
## Follow-up issues
Unresolved dimensions produce `proposed_follow_ups` entries suitable for durable
Gitea issues. The MCP tool **does not create** those issues in v1 (rollout is
log-only first). Controllers may file them from the proof payload.
## Links
- Umbrella: #655
- Vision: #652 · Roadmap: #653
- Checkpoint schema: #660 (soft dependency)
- Drain proof: #661 (soft)
- This issue: #662
## Non-goals
- HA multi-instance failover
- Automatic silent mutation replay
- Implementing the #660 checkpoint schema itself
+47
View File
@@ -349,6 +349,53 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
checkout is behind merged safety-gate changes. Restart guidance links to #420;
no tokens or MCP restart actions are exposed.
## Inventory API (#636)
`GET /api/v1/inventory` returns one versioned, read-only snapshot that unifies
what the lease (#433), worktree (#432), and runtime (#430) MVP views each show
separately, so traffic-control and recovery consumers read the same source.
`GET /api/v1/inventory/{section}` returns a single section under the identical
schema (`sessions`, `leases`, `locks`, `worktrees`, `namespaces`); an unknown
section is a `404` with `error: unknown_section`. Both routes are `GET`-only.
Each section carries its own `status` (`ok` / `degraded` / `unavailable`), a
`reason` when not `ok`, and a `scan_ms`. A subsystem that cannot be read
degrades to a reasoned section; it never raises and never emits an empty list
that would read as "nothing is there".
### Field authority
Every section names where its rows came from; authorities are never blended.
| Section | Authority | Source |
|---|---|---|
| `sessions` | `control_plane_db` | #613 control-plane DB (`mode=ro`), authoritative for exclusive ownership (#600/#601) |
| `leases` | `control_plane_db` | #613 control-plane DB; degrades if the `work_items` table is absent |
| `locks` | `filesystem` | durable per-issue lock files (`issue_lock_store`) |
| `worktrees` | `filesystem` | registered git worktrees via the #432 hygiene scanner |
| `namespaces` | `filesystem` | the active profile serving this web process (others are not enumerable) |
The payload restates this map under `field_authority` for machine consumers.
### Ownership safety
`ownership_authority_complete` is true only when every ownership-bearing section
(`sessions`, `leases`, `locks`) read cleanly. While it is false, nothing is
reported as unowned and no collision is asserted from a degraded source —
absence of evidence is reported as absence of evidence, never as free work.
`collisions` surfaces detectable conflicts, each with a `kind` and `severity`:
`lock-without-worktree`, `duplicate-live-lock`, `live-lock-dead-owner` (unexpired
lease, dead pid — a #753 recovery candidate that would read as live to a naive
timestamp check), `stale-lock-dead-owner`, `expired-lock-live-owner` (the
#635/#760 daemon-pid deadlock), `concurrent-active-lease`, `active-lease-past-expiry`,
and `orphan-lease`. Collisions are emitted only from sections that read cleanly.
The control-plane DB is opened through a `mode=ro` URI so a read never creates
or migrates it; paths are collapsed against `$HOME`, URLs lose userinfo and
query strings, and credential-shaped values are redacted at the boundary. Lease
steal/release and worktree deletion are Phase 2+ and have no representation here.
## Workflow-event timeline (#637)
`GET /api/v1/timeline` is a read-only, versioned aggregation of workflow
+255
View File
@@ -2485,6 +2485,7 @@ def _evaluate_issue_lock_recovery(
worktree_path,
prior_head_sha=local_head,
synced_head_sha=remote_head,
remote=remote,
)
# #772: with no remote branch there is no head to measure against, so the
@@ -18022,6 +18023,17 @@ def gitea_assess_master_parity(
}
if parity["restart_required"] and enforced:
out["report"] = master_parity_gate.parity_report(parity)
# #662 AC1: first post-restart master-parity probe also runs the boot
# reconcile once (log-only by default; never raises).
boot_proof = _ensure_boot_post_restart_reconcile()
if boot_proof is not None:
out["post_restart_reconcile"] = {
"reconcile_id": boot_proof.get("reconcile_id"),
"overall_status": boot_proof.get("overall_status"),
"mutation_hold": boot_proof.get("mutation_hold"),
"unresolved_count": boot_proof.get("unresolved_count"),
"mode": boot_proof.get("mode"),
}
return out
@@ -22470,6 +22482,249 @@ def gitea_request_mcp_restart(
return payload
# --- #662 post-restart reconciliation ---------------------------------------
_POST_RESTART_LAST_PROOF: dict | None = None
_POST_RESTART_BOOT_RAN = False
def _post_restart_reconcile_mode() -> str:
"""Return log_only (default) or enforce from process environment."""
raw = (os.environ.get("GITEA_POST_RESTART_RECONCILE_MODE") or "").strip().lower()
if raw in {"enforce", "enforced", "hold"}:
return "enforce"
return "log_only"
def _gather_post_restart_inventory(
*,
remote: str,
org: str,
repo: str,
limit: int = 200,
) -> dict:
"""Gather live control-plane facts for post-restart reconcile (#662).
Best-effort and fail-closed: any inventory section that cannot be read is
recorded on ``incomplete_reasons`` and ``inventory_complete`` is cleared.
Never mutates Gitea or the control-plane DB.
"""
import master_parity_gate
import post_restart_reconcile as prr
inventory_complete = True
incomplete_reasons: list[str] = []
sessions: list[dict] = []
leases: list[dict] = []
worktree_bindings: list[dict] = []
db, db_errs = _control_plane_db_or_error()
if db is None:
inventory_complete = False
incomplete_reasons.extend(
db_errs or ["control-plane DB unavailable; cannot reconcile after restart"]
)
else:
try:
sessions = db.list_sessions(statuses=("active",), limit=max(1, int(limit)))
except Exception as exc: # noqa: BLE001
inventory_complete = False
incomplete_reasons.append(
f"session inventory failed: {_redact(str(exc))}"
)
try:
lease_result = lease_lifecycle.list_active_leases(
db,
remote=remote if remote in REMOTES else remote,
org=org,
repo=repo,
role=None,
include_non_active=True,
limit=max(1, int(limit)),
)
leases = list(lease_result.get("leases") or [])
except Exception as exc: # noqa: BLE001
inventory_complete = False
incomplete_reasons.append(
f"lease inventory failed: {_redact(str(exc))}"
)
leases = []
# Derive worktree bindings from live leases that carry a path.
for row in leases:
wt = (row.get("worktree_path") or "").strip()
if not wt:
continue
exists = bool(lease_lifecycle.worktree_exists(wt))
worktree_bindings.append(
{
"path": wt,
"exists": exists,
"missing": not exists,
"lease_id": row.get("lease_id"),
"session_id": row.get("session_id"),
"work_kind": row.get("work_kind"),
"work_number": row.get("work_number"),
}
)
# Capability / master-parity dimension (code parity after restart).
try:
parity = _current_master_parity()
capabilities = {
"stale": bool(parity.get("stale")),
"startup_head": parity.get("startup_head"),
"current_head": parity.get("current_head"),
"in_parity": parity.get("in_parity"),
"mutation_safe": parity.get("mutation_safe"),
}
service_health = {
"healthy": bool(parity.get("mutation_safe", not parity.get("stale"))),
"parity_summary": master_parity_gate.format_parity(parity),
}
boot_head = parity.get("startup_head") or _process_boot_head_sha
current_head = parity.get("current_head")
except Exception as exc: # noqa: BLE001
inventory_complete = False
incomplete_reasons.append(
f"master-parity inventory failed: {_redact(str(exc))}"
)
capabilities = {"stale": None}
service_health = {"healthy": None, "error": "parity assessment failed"}
boot_head = _process_boot_head_sha
current_head = None
# #660 soft dependency: checkpoint schema not landed → skip dimension.
checkpoints_available = False
try:
import importlib
importlib.import_module("session_checkpoint_schema")
checkpoints_available = True
except Exception: # noqa: BLE001
checkpoints_available = False
return {
"inventory_complete": inventory_complete,
"incomplete_reasons": incomplete_reasons,
"service_health": service_health,
"clients": [], # client transport inventory is host/IDE-owned
"sessions": sessions,
"leases": leases,
"checkpoints_available": checkpoints_available,
"checkpoints": [] if checkpoints_available else None,
"worktree_bindings": worktree_bindings,
"pending_mutations": [],
"capabilities": capabilities,
"boot_head_sha": boot_head,
"current_head_sha": current_head,
"queue_state": {"safe_to_resume": inventory_complete},
"reconcile_version": prr.RECONCILE_VERSION,
}
def _run_post_restart_reconcile(
*,
remote: str = "prgs",
org: str | None = None,
repo: str | None = None,
mode: str | None = None,
limit: int = 200,
) -> dict:
"""Gather + classify post-restart state; cache the latest proof (#662)."""
global _POST_RESTART_LAST_PROOF, _POST_RESTART_BOOT_RAN
import post_restart_reconcile as prr
try:
_h, o, r = _resolve(remote, None, org, repo)
except ValueError as exc:
return {
"success": False,
"read_only": True,
"reasons": [str(exc)],
}
inventory = _gather_post_restart_inventory(
remote=remote, org=o, repo=r, limit=limit
)
proof = prr.reconcile_after_restart(
inventory,
mode=mode or _post_restart_reconcile_mode(),
)
payload = proof.as_dict()
payload["success"] = True
payload["read_only"] = True
payload["remote"] = remote
payload["org"] = o
payload["repo"] = r
payload["follow_up_create_supported"] = False
payload["follow_up_create_note"] = (
"proposed_follow_ups lists durable issues the apply path may create; "
"this tool never creates them (log-only by default, #662 rollout)"
)
_POST_RESTART_LAST_PROOF = payload
_POST_RESTART_BOOT_RAN = True
return payload
def _ensure_boot_post_restart_reconcile() -> dict | None:
"""Run post-restart reconcile once per process (boot hook, #662 AC1)."""
global _POST_RESTART_BOOT_RAN
if _POST_RESTART_BOOT_RAN:
return _POST_RESTART_LAST_PROOF
# Best-effort: never raise from the boot hook.
try:
return _run_post_restart_reconcile()
except Exception: # noqa: BLE001
_POST_RESTART_BOOT_RAN = True
return None
@mcp.tool()
def gitea_reconcile_after_restart(
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
mode: str | None = None,
limit: int = 200,
) -> dict:
"""Run post-restart MCP reconciliation and return a completion proof (#662).
Gathers live control-plane sessions, leases, worktree bindings, and
master-parity evidence, then classifies them with the pure
``post_restart_reconcile.reconcile_after_restart`` assessor. Returns a
machine-readable completion proof listing resolved / unresolved dimensions
and proposed durable follow-up issues.
Read-only by design: never restarts MCP, never auto-resumes write
mutations, and never creates Gitea issues (those are a separate apply
path). Default mode is ``log_only``; set
``GITEA_POST_RESTART_RECONCILE_MODE=enforce`` (or pass ``mode='enforce'``)
to set ``mutation_hold`` when anything remains unresolved.
Soft-depends on #660 for session checkpoints: when the checkpoint schema
module is absent the checkpoints dimension is ``skipped`` with an explicit
reason rather than inventing a schema.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"read_only": True,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
return _run_post_restart_reconcile(
remote=remote,
org=org,
repo=repo,
mode=mode,
limit=limit,
)
@mcp.tool()
def gitea_inspect_workflow_lease(
lease_id: str,
+24 -15
View File
@@ -150,23 +150,14 @@ def read_merge_sync_provenance(
*,
prior_head_sha: str | None,
synced_head_sha: str | None,
remote: str | None = None,
) -> dict:
"""Observe whether ``synced_head_sha`` is a sanctioned merge-based branch sync
that advanced the PR branch past ``prior_head_sha`` (#871).
"""Observe whether ``synced_head_sha`` is a sanctioned merge-sync of a base
into the branch above ``prior_head_sha`` (#871/#872).
``gitea_update_pr_branch_by_merge`` advances a PR branch by merging the base
branch *into* the branch (``POST /pulls/{n}/update?style=merge``). The result
is a merge commit ``M`` on the branch whose **first** parent is the prior
branch head and whose second parent is the base tip. When the owning session
then dies without the durable lock's recorded head being refreshed, the local
worktree still sits at ``prior_head_sha`` while the live PR head is ``M``.
Recovering that drift safely requires proving the remote head is *exactly*
such a merge-sync — not a rewrite, rebase, force-push, or an unrelated
commit. This is that server-side observation. It reports facts only; the
disposition lives in ``issue_lock_recovery``. Every field is read from git in
the declared worktree — nothing is supplied by, or reachable from, an MCP
caller (#871).
Reports server-derived git facts only; the recovery disposition lives in
``issue_lock_recovery``. All comparisons are executed locally in the
declared worktree -- nothing is taken from caller parameters.
Provenance is proven only when ALL hold:
@@ -183,6 +174,7 @@ def read_merge_sync_provenance(
path = (worktree_path or "").strip()
prior = (prior_head_sha or "").strip()
synced = (synced_head_sha or "").strip()
target_remote = (remote or "").strip() or None
result: dict = {
"prior_head_sha": prior or None,
"synced_head_sha": synced or None,
@@ -235,6 +227,23 @@ def read_merge_sync_provenance(
try:
result["prior_present"] = _present(prior)
result["synced_present"] = _present(synced)
if not result["synced_present"] and path and os.path.isdir(path):
if target_remote:
subprocess.run(
["git", "-C", path, "fetch", target_remote, "--quiet"],
capture_output=True,
text=True,
check=False,
)
result["synced_present"] = _present(synced)
if not result["synced_present"]:
subprocess.run(
["git", "-C", path, "fetch", "--quiet"],
capture_output=True,
text=True,
check=False,
)
result["synced_present"] = _present(synced)
except OSError as exc: # git unavailable — fail closed, never assume
result["reasons"].append(f"merge-sync provenance probe could not run: {exc}")
return result
+791
View File
@@ -0,0 +1,791 @@
"""Post-restart MCP reconciliation and completion proof (#662).
After an MCP process restart, sessions, leases, capabilities, worktrees, and
interrupted mutations are not systematically reconciled; operators rebuild
context from chat. This module is the pure classification core of the
post-restart reconcile path.
Design rules (mirrors ``restart_coordinator`` / ``workflow_dashboard``):
* **Pure classification.** :func:`reconcile_after_restart` takes an already
gathered inventory and returns a structured *completion proof*. It never
touches the network, the filesystem, or a live process, so multi-session
fixtures can drive every branch in unit tests.
* **Fail closed.** Incomplete inventory never reports overall ``complete``.
Ambiguous interrupted mutations are ``unresolved`` (never silently resumed).
* **No blind write resume.** The proof never authorizes replaying a mutation;
it only classifies evidence and names follow-up work.
* **#660 soft dependency.** When durable session checkpoints are not present
in the inventory, the checkpoint dimension is ``skipped`` with an explicit
reason rather than inventing a schema (#660 lands separately).
* **Log-only then enforce.** Default mode is ``log_only``. ``enforce`` sets
``mutation_hold`` when anything remains unresolved so callers can block
write ops until reconcile is complete or degraded mode is documented.
The single sanctioned gather+classify entry point is the MCP tool
``gitea_reconcile_after_restart`` (read-only inventory gather + pure classify).
Creating durable follow-up Gitea issues from unresolved items is an explicit
apply step outside this pure module.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Mapping, Sequence
from uuid import uuid4
import lease_lifecycle
RECONCILE_VERSION = "1.0.0-issue-662"
SCHEMA_VERSION = 1
# Overall proof statuses.
STATUS_COMPLETE = "complete"
STATUS_DEGRADED = "degraded"
STATUS_FAILED = "failed"
# Per-dimension item statuses.
ITEM_RESOLVED = "resolved"
ITEM_UNRESOLVED = "unresolved"
ITEM_DEGRADED = "degraded"
ITEM_SKIPPED = "skipped"
# Modes.
MODE_LOG_ONLY = "log_only"
MODE_ENFORCE = "enforce"
# Lease / session phases that imply a write critical section was in flight.
MUTATING_PHASES = frozenset(
{
"implementing",
"publishing",
"merging",
"reviewing",
"committing",
"pushing",
"closing",
"mutating",
"critical_section",
}
)
# Dimensions the acceptance criteria require.
DIM_SERVICE_HEALTH = "service_health"
DIM_CLIENTS = "clients"
DIM_SESSIONS = "sessions"
DIM_CHECKPOINTS = "checkpoints"
DIM_LEASES = "leases"
DIM_CAPABILITIES = "capabilities"
DIM_WORKTREES = "worktrees"
DIM_MUTATIONS = "interrupted_mutations"
DIM_DUPLICATES = "duplicates"
DIM_QUEUE = "queue"
def _utc_now() -> datetime:
return datetime.now(timezone.utc)
def _ts(dt: datetime) -> str:
return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
@dataclass(frozen=True)
class ReconcileItem:
"""One dimension of the post-restart reconcile report."""
dimension: str
status: str
summary: str
details: dict[str, Any] = field(default_factory=dict)
follow_up_required: bool = False
def as_dict(self) -> dict[str, Any]:
return {
"dimension": self.dimension,
"status": self.status,
"summary": self.summary,
"details": dict(self.details),
"follow_up_required": self.follow_up_required,
}
@dataclass(frozen=True)
class FollowUpIssue:
"""A durable follow-up issue the apply path may create for unresolved work."""
title: str
body: str
dimension: str
severity: str = "high"
def as_dict(self) -> dict[str, Any]:
return {
"title": self.title,
"body": self.body,
"dimension": self.dimension,
"severity": self.severity,
}
@dataclass(frozen=True)
class RestartCompletionProof:
"""Machine-readable post-restart completion proof (#662 AC2)."""
schema_version: int
reconcile_version: str
reconcile_id: str
started_at: str
finished_at: str
boot_head_sha: str | None
current_head_sha: str | None
inventory_complete: bool
incomplete_reasons: tuple[str, ...]
mode: str
mutation_hold: bool
overall_status: str
items: tuple[ReconcileItem, ...]
proposed_follow_ups: tuple[FollowUpIssue, ...]
resolved_count: int
unresolved_count: int
skipped_count: int
note: str
def as_dict(self) -> dict[str, Any]:
return {
"schema_version": self.schema_version,
"reconcile_version": self.reconcile_version,
"reconcile_id": self.reconcile_id,
"started_at": self.started_at,
"finished_at": self.finished_at,
"boot_head_sha": self.boot_head_sha,
"current_head_sha": self.current_head_sha,
"inventory_complete": self.inventory_complete,
"incomplete_reasons": list(self.incomplete_reasons),
"mode": self.mode,
"mutation_hold": self.mutation_hold,
"overall_status": self.overall_status,
"items": [i.as_dict() for i in self.items],
"proposed_follow_ups": [f.as_dict() for f in self.proposed_follow_ups],
"resolved_count": self.resolved_count,
"unresolved_count": self.unresolved_count,
"skipped_count": self.skipped_count,
"note": self.note,
"links": {
"umbrella": 655,
"vision": 652,
"roadmap": 653,
"issue": 662,
"checkpoint_schema": 660,
"drain_proof": 661,
},
}
def _item(
dimension: str,
status: str,
summary: str,
*,
details: dict[str, Any] | None = None,
follow_up: bool = False,
) -> ReconcileItem:
return ReconcileItem(
dimension=dimension,
status=status,
summary=summary,
details=dict(details or {}),
follow_up_required=follow_up,
)
def _lease_freshness(lease: Mapping[str, Any]) -> str:
fr = lease.get("freshness")
if isinstance(fr, Mapping):
return str(fr.get("freshness") or fr.get("status") or "unknown")
if isinstance(fr, str):
return fr
# Fall back to pure classifier when raw lease rows are supplied.
try:
return str(lease_lifecycle.classify_lease_freshness(dict(lease)).get("freshness") or "unknown")
except Exception: # noqa: BLE001 - pure path must not raise on bad rows
return "unknown"
def _is_live_freshness(freshness: str) -> bool:
return freshness in {"active", "live", "fresh"}
def _is_mutating_phase(phase: str | None) -> bool:
p = (phase or "").strip().lower()
if not p:
return False
if p in MUTATING_PHASES:
return True
# Soft match for compound phases like "author_implementing".
return any(token in p for token in MUTATING_PHASES)
def _detect_interrupted_mutations(
leases: Sequence[Mapping[str, Any]],
pending_mutations: Sequence[Mapping[str, Any]],
) -> list[dict[str, Any]]:
"""Return interrupted-mutation evidence (never auto-resumes writes)."""
found: list[dict[str, Any]] = []
for raw in pending_mutations or ():
if not isinstance(raw, Mapping):
continue
found.append(
{
"source": "pending_mutation_inventory",
"status": "unresolved",
"phase": raw.get("phase"),
"session_id": raw.get("session_id"),
"work_kind": raw.get("work_kind") or raw.get("kind"),
"work_number": raw.get("work_number") or raw.get("number"),
"reason": raw.get("reason")
or "pending mutation recorded across process restart",
"resume_allowed": False,
}
)
for lease in leases or ():
if not isinstance(lease, Mapping):
continue
phase = lease.get("phase")
freshness = _lease_freshness(lease)
if not _is_mutating_phase(str(phase) if phase is not None else None):
continue
# A mutating phase whose owner is not live is interrupted.
if _is_live_freshness(freshness):
# Still live after restart is itself surprising — flag for review.
found.append(
{
"source": "lease_mutating_phase",
"status": "unresolved",
"phase": phase,
"freshness": freshness,
"lease_id": lease.get("lease_id"),
"session_id": lease.get("session_id"),
"work_kind": lease.get("work_kind"),
"work_number": lease.get("work_number"),
"worktree_path": lease.get("worktree_path"),
"reason": (
"mutating lease phase still classified live after restart; "
"do not auto-resume writes"
),
"resume_allowed": False,
}
)
else:
found.append(
{
"source": "lease_mutating_phase",
"status": "unresolved",
"phase": phase,
"freshness": freshness,
"lease_id": lease.get("lease_id"),
"session_id": lease.get("session_id"),
"work_kind": lease.get("work_kind"),
"work_number": lease.get("work_number"),
"worktree_path": lease.get("worktree_path"),
"reason": (
f"mutating lease phase '{phase}' with non-live freshness "
f"'{freshness}' — interrupted by restart"
),
"resume_allowed": False,
}
)
return found
def _detect_duplicate_work(
leases: Sequence[Mapping[str, Any]],
) -> list[dict[str, Any]]:
"""Surface duplicate live claims on the same work item."""
by_work: dict[tuple[Any, Any], list[Mapping[str, Any]]] = {}
for lease in leases or ():
if not isinstance(lease, Mapping):
continue
if not _is_live_freshness(_lease_freshness(lease)):
continue
key = (lease.get("work_kind"), lease.get("work_number"))
if key[0] is None or key[1] is None:
continue
by_work.setdefault(key, []).append(lease)
dups: list[dict[str, Any]] = []
for (kind, number), rows in sorted(by_work.items(), key=lambda kv: str(kv[0])):
if len(rows) < 2:
continue
dups.append(
{
"work_kind": kind,
"work_number": number,
"claim_count": len(rows),
"session_ids": [r.get("session_id") for r in rows],
"lease_ids": [r.get("lease_id") for r in rows],
}
)
return dups
def _follow_up_for_item(item: ReconcileItem) -> FollowUpIssue | None:
if not item.follow_up_required:
return None
title = f"[post-restart] unresolved {item.dimension} after MCP restart"
body = (
f"## Post-restart reconcile follow-up (#662)\n\n"
f"**Dimension:** `{item.dimension}`\n"
f"**Status:** `{item.status}`\n"
f"**Summary:** {item.summary}\n\n"
f"```json\n{item.details!r}\n```\n\n"
f"Parent umbrella: #655 · Vision: #652 · Roadmap: #653 · Reconcile: #662\n"
f"Do **not** auto-resume write mutations; reconcile evidence first.\n"
)
return FollowUpIssue(
title=title,
body=body,
dimension=item.dimension,
severity="high" if item.dimension == DIM_MUTATIONS else "medium",
)
def reconcile_after_restart(
inventory: Mapping[str, Any],
*,
now: datetime | None = None,
mode: str = MODE_LOG_ONLY,
reconcile_id: str | None = None,
) -> RestartCompletionProof:
"""Classify a post-restart inventory into a completion proof (#662).
Parameters
----------
inventory:
Gathered facts. Expected keys (all optional except completeness):
* ``inventory_complete`` (bool) — fail closed when false
* ``incomplete_reasons`` (list[str])
* ``service_health`` (dict with ``healthy`` bool)
* ``clients`` (list) — connected client descriptors
* ``sessions`` (list)
* ``leases`` (list, optionally with ``freshness``)
* ``checkpoints`` (list | None) — durable session checkpoints (#660)
* ``checkpoints_available`` (bool) — False when #660 schema absent
* ``worktree_bindings`` (list)
* ``pending_mutations`` (list) — explicit interrupted-mutation evidence
* ``capabilities`` (dict with optional ``stale`` / heads)
* ``boot_head_sha`` / ``current_head_sha``
* ``queue_state`` (dict)
mode:
``log_only`` (default) or ``enforce`` (sets mutation_hold on unresolved).
"""
started = now or _utc_now()
mode_norm = (mode or MODE_LOG_ONLY).strip().lower()
if mode_norm not in {MODE_LOG_ONLY, MODE_ENFORCE}:
mode_norm = MODE_LOG_ONLY
inventory_complete = bool(inventory.get("inventory_complete", False))
incomplete_reasons = tuple(
str(r) for r in (inventory.get("incomplete_reasons") or []) if str(r).strip()
)
items: list[ReconcileItem] = []
# --- service health -------------------------------------------------
health = inventory.get("service_health") or {}
if not isinstance(health, Mapping):
health = {}
if not inventory_complete and "service_health" not in inventory:
items.append(
_item(
DIM_SERVICE_HEALTH,
ITEM_UNRESOLVED,
"service health unknown because inventory is incomplete",
details={"inventory_complete": False},
follow_up=True,
)
)
elif health.get("healthy") is True:
items.append(
_item(
DIM_SERVICE_HEALTH,
ITEM_RESOLVED,
"service health verified",
details=dict(health),
)
)
elif health.get("healthy") is False:
items.append(
_item(
DIM_SERVICE_HEALTH,
ITEM_UNRESOLVED,
"service health check failed",
details=dict(health),
follow_up=True,
)
)
else:
items.append(
_item(
DIM_SERVICE_HEALTH,
ITEM_DEGRADED,
"service health not reported; treating as degraded",
details=dict(health),
follow_up=True,
)
)
# --- clients --------------------------------------------------------
clients = list(inventory.get("clients") or [])
disconnected = [
c
for c in clients
if isinstance(c, Mapping) and c.get("connected") is False
]
if "clients" not in inventory:
items.append(
_item(
DIM_CLIENTS,
ITEM_SKIPPED,
"client inventory not supplied",
details={},
)
)
elif disconnected:
items.append(
_item(
DIM_CLIENTS,
ITEM_UNRESOLVED,
f"{len(disconnected)} disconnected client(s) need reconnect",
details={"disconnected": disconnected, "total": len(clients)},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_CLIENTS,
ITEM_RESOLVED,
f"{len(clients)} client(s) accounted for",
details={"total": len(clients)},
)
)
# --- sessions -------------------------------------------------------
sessions = [s for s in (inventory.get("sessions") or []) if isinstance(s, Mapping)]
orphan_sessions = [
s
for s in sessions
if str(s.get("status") or "").lower() == "active"
and s.get("pid") is not None
and not lease_lifecycle.is_process_alive(s.get("pid"))
]
if orphan_sessions:
items.append(
_item(
DIM_SESSIONS,
ITEM_UNRESOLVED,
f"{len(orphan_sessions)} active session row(s) with dead owner pid",
details={
"orphan_session_ids": [s.get("session_id") for s in orphan_sessions],
"total_sessions": len(sessions),
},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_SESSIONS,
ITEM_RESOLVED,
f"{len(sessions)} session row(s) reconciled (no dead-pid orphans)",
details={"total_sessions": len(sessions)},
)
)
# --- checkpoints (#660 soft) ----------------------------------------
checkpoints_available = inventory.get("checkpoints_available")
checkpoints = inventory.get("checkpoints")
if checkpoints_available is False or (
checkpoints is None and "checkpoints" not in inventory
):
items.append(
_item(
DIM_CHECKPOINTS,
ITEM_SKIPPED,
"durable session checkpoint schema not available yet (#660)",
details={"depends_on": 660},
)
)
else:
cp_list = [c for c in (checkpoints or []) if isinstance(c, Mapping)]
stale_cp = [c for c in cp_list if c.get("stale") or c.get("invalid")]
if stale_cp:
items.append(
_item(
DIM_CHECKPOINTS,
ITEM_UNRESOLVED,
f"{len(stale_cp)} checkpoint(s) invalid or stale vs live state",
details={"stale_count": len(stale_cp), "total": len(cp_list)},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_CHECKPOINTS,
ITEM_RESOLVED,
f"{len(cp_list)} checkpoint(s) consistent with live state",
details={"total": len(cp_list)},
)
)
# --- leases / locks -------------------------------------------------
leases = [L for L in (inventory.get("leases") or []) if isinstance(L, Mapping)]
live_leases = [L for L in leases if _is_live_freshness(_lease_freshness(L))]
items.append(
_item(
DIM_LEASES,
ITEM_RESOLVED if inventory_complete else ITEM_DEGRADED,
f"{len(live_leases)} live lease(s) of {len(leases)} inventoried",
details={
"live_count": len(live_leases),
"total": len(leases),
"live_lease_ids": [L.get("lease_id") for L in live_leases],
},
follow_up=not inventory_complete,
)
)
# --- capabilities / stale runtime -----------------------------------
caps = inventory.get("capabilities") or {}
if not isinstance(caps, Mapping):
caps = {}
if caps.get("stale") is True:
items.append(
_item(
DIM_CAPABILITIES,
ITEM_UNRESOLVED,
"runtime code is stale vs on-disk master; restart did not reach parity",
details=dict(caps),
follow_up=True,
)
)
else:
items.append(
_item(
DIM_CAPABILITIES,
ITEM_RESOLVED,
"capability/runtime parity acceptable",
details=dict(caps) if caps else {"stale": False},
)
)
# --- worktrees ------------------------------------------------------
bindings = [
b for b in (inventory.get("worktree_bindings") or []) if isinstance(b, Mapping)
]
missing_wt = [
b
for b in bindings
if b.get("missing") is True or b.get("exists") is False
]
if "worktree_bindings" not in inventory:
items.append(
_item(
DIM_WORKTREES,
ITEM_SKIPPED,
"worktree binding inventory not supplied",
)
)
elif missing_wt:
items.append(
_item(
DIM_WORKTREES,
ITEM_UNRESOLVED,
f"{len(missing_wt)} worktree binding(s) missing on disk",
details={"missing": missing_wt, "total": len(bindings)},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_WORKTREES,
ITEM_RESOLVED,
f"{len(bindings)} worktree binding(s) present",
details={"total": len(bindings)},
)
)
# --- interrupted mutations (AC4) ------------------------------------
pending = [
m
for m in (inventory.get("pending_mutations") or [])
if isinstance(m, Mapping)
]
interrupted = _detect_interrupted_mutations(leases, pending)
if interrupted:
items.append(
_item(
DIM_MUTATIONS,
ITEM_UNRESOLVED,
f"{len(interrupted)} interrupted mutation(s); write resume forbidden",
details={"interrupted": interrupted},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_MUTATIONS,
ITEM_RESOLVED,
"no interrupted mutations detected",
details={"interrupted": []},
)
)
# --- duplicates -----------------------------------------------------
dups = _detect_duplicate_work(leases)
if dups:
items.append(
_item(
DIM_DUPLICATES,
ITEM_UNRESOLVED,
f"{len(dups)} work item(s) have multiple live claims",
details={"duplicates": dups},
follow_up=True,
)
)
else:
items.append(
_item(
DIM_DUPLICATES,
ITEM_RESOLVED,
"no duplicate live claims detected",
details={"duplicates": []},
)
)
# --- queue ----------------------------------------------------------
queue = inventory.get("queue_state")
if queue is None:
items.append(
_item(
DIM_QUEUE,
ITEM_SKIPPED,
"allocator queue state not supplied",
)
)
elif isinstance(queue, Mapping) and queue.get("safe_to_resume") is False:
items.append(
_item(
DIM_QUEUE,
ITEM_UNRESOLVED,
"allocator queue not safe to resume",
details=dict(queue),
follow_up=True,
)
)
else:
items.append(
_item(
DIM_QUEUE,
ITEM_RESOLVED,
"allocator queue state acceptable",
details=dict(queue) if isinstance(queue, Mapping) else {},
)
)
# Incomplete inventory always degrades the whole proof.
if not inventory_complete:
# Ensure at least one follow-up names the incomplete inventory.
items.append(
_item(
"inventory",
ITEM_UNRESOLVED,
"control-plane inventory incomplete; reconcile cannot claim success",
details={"reasons": list(incomplete_reasons)},
follow_up=True,
)
)
resolved = sum(1 for i in items if i.status == ITEM_RESOLVED)
unresolved = sum(1 for i in items if i.status in {ITEM_UNRESOLVED, ITEM_DEGRADED})
skipped = sum(1 for i in items if i.status == ITEM_SKIPPED)
if not inventory_complete or any(i.status == ITEM_UNRESOLVED for i in items):
if any(i.status == ITEM_UNRESOLVED for i in items) and inventory_complete:
overall = STATUS_DEGRADED
elif not inventory_complete:
overall = STATUS_FAILED
else:
overall = STATUS_DEGRADED
elif any(i.status == ITEM_DEGRADED for i in items):
overall = STATUS_DEGRADED
else:
overall = STATUS_COMPLETE
# Enforce mode holds mutations whenever anything is unresolved/failed.
mutation_hold = False
if mode_norm == MODE_ENFORCE and overall in {STATUS_DEGRADED, STATUS_FAILED}:
mutation_hold = True
if mode_norm == MODE_ENFORCE and any(
i.dimension == DIM_MUTATIONS and i.status == ITEM_UNRESOLVED for i in items
):
mutation_hold = True
follow_ups = tuple(
fu for i in items if (fu := _follow_up_for_item(i)) is not None
)
finished = _utc_now() if now is None else now
note = (
"Read-only completion proof. Never auto-resumes write mutations. "
"Unresolved items require durable follow-up before claiming clean restart. "
f"Mode={mode_norm}."
)
return RestartCompletionProof(
schema_version=SCHEMA_VERSION,
reconcile_version=RECONCILE_VERSION,
reconcile_id=(reconcile_id or f"reconcile-{uuid4().hex[:12]}"),
started_at=_ts(started),
finished_at=_ts(finished),
boot_head_sha=(
str(inventory.get("boot_head_sha")).strip()
if inventory.get("boot_head_sha")
else None
),
current_head_sha=(
str(inventory.get("current_head_sha")).strip()
if inventory.get("current_head_sha")
else None
),
inventory_complete=inventory_complete,
incomplete_reasons=incomplete_reasons,
mode=mode_norm,
mutation_hold=mutation_hold,
overall_status=overall,
items=tuple(items),
proposed_follow_ups=follow_ups,
resolved_count=resolved,
unresolved_count=unresolved,
skipped_count=skipped,
note=note,
)
def mutations_allowed(proof: RestartCompletionProof | Mapping[str, Any] | None) -> bool:
"""Return whether write mutations may proceed under the given proof."""
if proof is None:
return True # no proof yet → caller decides; enforce path sets hold
if isinstance(proof, RestartCompletionProof):
return not proof.mutation_hold
if isinstance(proof, Mapping):
return not bool(proof.get("mutation_hold"))
return True
+10
View File
@@ -132,6 +132,16 @@ 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).
"reconcile_after_restart": {
"permission": "gitea.read",
"role": "author",
},
"gitea_reconcile_after_restart": {
"permission": "gitea.read",
"role": "author",
},
# PR synchronization lifecycle: assess is read-only (any role with gitea.read);
# update-by-merge is author-only and mutates the PR head via Gitea API.
"assess_pr_sync_status": {
@@ -0,0 +1,249 @@
"""Tests for post-restart MCP reconciliation and completion proof (#662)."""
from __future__ import annotations
import json
import os
import unittest
from datetime import datetime, timezone
import post_restart_reconcile as prr
NOW = datetime(2026, 7, 24, 12, 0, 0, tzinfo=timezone.utc)
def _base_inventory(**overrides):
inv = {
"inventory_complete": True,
"incomplete_reasons": [],
"service_health": {"healthy": True},
"clients": [{"session_id": "c1", "connected": True}],
"sessions": [
{
"session_id": "s-live",
"status": "active",
"pid": os.getpid(),
"role": "author",
}
],
"leases": [],
"checkpoints_available": False,
"worktree_bindings": [{"path": "/tmp/wt", "exists": True}],
"pending_mutations": [],
"capabilities": {"stale": False},
"boot_head_sha": "a" * 40,
"current_head_sha": "a" * 40,
"queue_state": {"safe_to_resume": True},
}
inv.update(overrides)
return inv
class IncompleteInventoryTest(unittest.TestCase):
def test_incomplete_inventory_fails_closed(self) -> None:
proof = prr.reconcile_after_restart(
{
"inventory_complete": False,
"incomplete_reasons": ["control-plane DB unavailable"],
},
now=NOW,
mode=prr.MODE_ENFORCE,
reconcile_id="test-incomplete",
)
self.assertEqual(proof.overall_status, prr.STATUS_FAILED)
self.assertTrue(proof.mutation_hold)
self.assertFalse(proof.inventory_complete)
self.assertTrue(proof.proposed_follow_ups)
self.assertIn("control-plane DB unavailable", proof.incomplete_reasons)
class HappyPathTest(unittest.TestCase):
def test_clean_restart_is_complete_without_mutation_hold(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(),
now=NOW,
mode=prr.MODE_ENFORCE,
reconcile_id="test-clean",
)
self.assertEqual(proof.overall_status, prr.STATUS_COMPLETE)
self.assertFalse(proof.mutation_hold)
self.assertEqual(proof.unresolved_count, 0)
cp = next(i for i in proof.items if i.dimension == prr.DIM_CHECKPOINTS)
self.assertEqual(cp.status, prr.ITEM_SKIPPED)
links = proof.as_dict()["links"]
self.assertEqual(links["umbrella"], 655)
self.assertEqual(links["issue"], 662)
self.assertEqual(links["vision"], 652)
self.assertEqual(links["roadmap"], 653)
class InterruptedMutationTest(unittest.TestCase):
def test_mutating_lease_with_dead_owner_is_unresolved(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
leases=[
{
"lease_id": "lease-mut",
"session_id": "s-dead",
"phase": "implementing",
"work_kind": "issue",
"work_number": 662,
"worktree_path": "/tmp/wt-662",
"freshness": {"freshness": "stale_dead_process"},
}
]
),
now=NOW,
mode=prr.MODE_ENFORCE,
)
mut = next(i for i in proof.items if i.dimension == prr.DIM_MUTATIONS)
self.assertEqual(mut.status, prr.ITEM_UNRESOLVED)
self.assertTrue(mut.follow_up_required)
interrupted = mut.details["interrupted"]
self.assertEqual(len(interrupted), 1)
self.assertFalse(interrupted[0]["resume_allowed"])
self.assertTrue(proof.mutation_hold)
self.assertTrue(
any(f.dimension == prr.DIM_MUTATIONS for f in proof.proposed_follow_ups)
)
def test_explicit_pending_mutation_inventory(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
pending_mutations=[
{
"session_id": "s1",
"phase": "publishing",
"work_kind": "pr",
"work_number": 856,
"reason": "push interrupted mid-flight",
}
]
),
now=NOW,
mode=prr.MODE_LOG_ONLY,
)
mut = next(i for i in proof.items if i.dimension == prr.DIM_MUTATIONS)
self.assertEqual(mut.status, prr.ITEM_UNRESOLVED)
# log_only never holds mutations even when unresolved
self.assertFalse(proof.mutation_hold)
self.assertEqual(proof.overall_status, prr.STATUS_DEGRADED)
class DuplicateClaimsTest(unittest.TestCase):
def test_duplicate_live_claims_flagged(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
leases=[
{
"lease_id": "l1",
"session_id": "s1",
"phase": "allocated",
"work_kind": "issue",
"work_number": 100,
"freshness": {"freshness": "active"},
},
{
"lease_id": "l2",
"session_id": "s2",
"phase": "allocated",
"work_kind": "issue",
"work_number": 100,
"freshness": {"freshness": "active"},
},
]
),
now=NOW,
mode=prr.MODE_ENFORCE,
)
dups = next(i for i in proof.items if i.dimension == prr.DIM_DUPLICATES)
self.assertEqual(dups.status, prr.ITEM_UNRESOLVED)
self.assertEqual(dups.details["duplicates"][0]["claim_count"], 2)
self.assertTrue(proof.mutation_hold)
class OrphanSessionTest(unittest.TestCase):
def test_active_session_dead_pid_is_unresolved(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
sessions=[
{
"session_id": "ghost",
"status": "active",
"pid": 2_000_000_000,
"role": "author",
}
]
),
now=NOW,
mode=prr.MODE_ENFORCE,
)
sess = next(i for i in proof.items if i.dimension == prr.DIM_SESSIONS)
self.assertEqual(sess.status, prr.ITEM_UNRESOLVED)
self.assertIn("ghost", sess.details["orphan_session_ids"])
class CapabilityStaleTest(unittest.TestCase):
def test_stale_runtime_unresolved(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(capabilities={"stale": True, "startup_head": "aaa"}),
now=NOW,
mode=prr.MODE_ENFORCE,
)
caps = next(i for i in proof.items if i.dimension == prr.DIM_CAPABILITIES)
self.assertEqual(caps.status, prr.ITEM_UNRESOLVED)
self.assertTrue(proof.mutation_hold)
class MutationsAllowedHelperTest(unittest.TestCase):
def test_mutations_allowed_respects_hold(self) -> None:
held = prr.reconcile_after_restart(
_base_inventory(
pending_mutations=[{"phase": "merging", "session_id": "x"}]
),
now=NOW,
mode=prr.MODE_ENFORCE,
)
self.assertFalse(prr.mutations_allowed(held))
self.assertFalse(prr.mutations_allowed(held.as_dict()))
clean = prr.reconcile_after_restart(
_base_inventory(), now=NOW, mode=prr.MODE_ENFORCE
)
self.assertTrue(prr.mutations_allowed(clean))
class CheckpointSoftDependencyTest(unittest.TestCase):
def test_checkpoints_when_schema_present(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
checkpoints_available=True,
checkpoints=[{"session_id": "s1", "stale": False}],
),
now=NOW,
)
cp = next(i for i in proof.items if i.dimension == prr.DIM_CHECKPOINTS)
self.assertEqual(cp.status, prr.ITEM_RESOLVED)
def test_stale_checkpoints_unresolved(self) -> None:
proof = prr.reconcile_after_restart(
_base_inventory(
checkpoints_available=True,
checkpoints=[{"session_id": "s1", "stale": True}],
),
now=NOW,
mode=prr.MODE_ENFORCE,
)
cp = next(i for i in proof.items if i.dimension == prr.DIM_CHECKPOINTS)
self.assertEqual(cp.status, prr.ITEM_UNRESOLVED)
class ProofSerializationTest(unittest.TestCase):
def test_as_dict_is_json_friendly(self) -> None:
proof = prr.reconcile_after_restart(_base_inventory(), now=NOW)
blob = json.dumps(proof.as_dict())
self.assertIn("reconcile_id", blob)
self.assertIn("proposed_follow_ups", blob)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,332 @@
"""Dead-session recovery for cross-session PR base-syncs (#872).
Validates that when a sanctioned server-side base-sync (``gitea_update_pr_branch_by_merge``)
advances a PR's remote head from A to B (a merge commit whose first parent is A),
a subsequent author session whose local worktree is at A can recover the dead-session
lock via HEAD_RELATION_REMOTE_MERGE_SYNCED and execute a second base-sync (B to C)
without deadlock or RuntimeError.
"""
from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_lock_recovery # noqa: E402
import issue_lock_store # noqa: E402
import issue_lock_worktree # noqa: E402
ISSUE = 8720
PR_NUMBER = 8721
BRANCH = f"fix/issue-{ISSUE}-dead-session-two-base-syncs"
IDENTITY = "example-author"
PROFILE = "prgs-author"
REMOTE = "prgs"
ORG = "ExampleOrg"
REPO = "ExampleRepo"
def dead_pid() -> int:
proc = subprocess.Popen([sys.executable, "-c", "pass"])
proc.wait()
return proc.pid
def future_ts(hours: int = 4) -> str:
return (
(datetime.now(timezone.utc) + timedelta(hours=hours))
.isoformat()
.replace("+00:00", "Z")
)
def _git(cwd, *args):
return subprocess.run(
["git", "-C", cwd, *args],
capture_output=True,
text=True,
check=True,
)
def _rev(cwd, ref="HEAD") -> str:
return _git(cwd, "rev-parse", ref).stdout.strip()
def build_two_base_sync_repo(tmp: str) -> dict:
"""Build a repo simulating two sequential base-sync merges."""
_git(tmp, "init", "-q", "-b", "master")
_git(tmp, "config", "user.email", "[email protected]")
_git(tmp, "config", "user.name", "Author")
Path(tmp, "base.txt").write_text("base 1\n")
_git(tmp, "add", "-A")
_git(tmp, "commit", "-q", "-m", "initial master")
# Feature branch cut from initial master -> Head A (prior/recorded head)
_git(tmp, "checkout", "-q", "-b", BRANCH)
Path(tmp, "feature.txt").write_text("feature work\n")
_git(tmp, "add", "-A")
_git(tmp, "commit", "-q", "-m", "feature commit A")
head_a = _rev(tmp)
# Master advances -> Master 1
_git(tmp, "checkout", "-q", "master")
Path(tmp, "base.txt").write_text("base 1\nbase 2\n")
_git(tmp, "add", "-A")
_git(tmp, "commit", "-q", "-m", "master advance 1")
master_1 = _rev(tmp)
# First sync: merge master into feature -> Head B (merge commit, first parent = A)
_git(tmp, "checkout", "-q", BRANCH)
_git(tmp, "merge", "-q", "--no-ff", "-m", "First base-sync (merge master)", "master")
head_b = _rev(tmp)
# Master advances again -> Master 2
_git(tmp, "checkout", "-q", "master")
Path(tmp, "base.txt").write_text("base 1\nbase 2\nbase 3\n")
_git(tmp, "add", "-A")
_git(tmp, "commit", "-q", "-m", "master advance 2")
master_2 = _rev(tmp)
# Second sync: merge master into feature -> Head C (merge commit, first parent = B)
_git(tmp, "checkout", "-q", BRANCH)
_git(tmp, "merge", "-q", "--no-ff", "-m", "Second base-sync (merge master)", "master")
head_c = _rev(tmp)
# Non-merge rebase/force-pushed branch shape
_git(tmp, "checkout", "-q", "-b", "rebased-branch", head_a)
Path(tmp, "rebase.txt").write_text("rebased\n")
_git(tmp, "add", "-A")
_git(tmp, "commit", "-q", "-m", "rebased commit")
rebased_head = _rev(tmp)
# Reset worktree back to head A, as a dead session leaving local worktree at A
_git(tmp, "checkout", "-q", BRANCH)
_git(tmp, "reset", "-q", "--hard", head_a)
return {
"head_a": head_a,
"master_1": master_1,
"head_b": head_b,
"master_2": master_2,
"head_c": head_c,
"rebased_head": rebased_head,
}
def make_dead_lock(worktree, **over):
pid = dead_pid()
lock = {
"issue_number": ISSUE,
"branch_name": BRANCH,
"worktree_path": worktree,
"remote": REMOTE,
"org": ORG,
"repo": REPO,
"session_pid": pid,
"pid": pid,
"claimant": {"username": IDENTITY, "profile": PROFILE},
"work_lease": {
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
"issue_number": ISSUE,
"branch": BRANCH,
"worktree_path": worktree,
"claimant": {"username": IDENTITY, "profile": PROFILE},
"expires_at": future_ts(),
},
}
lock.update(over)
return lock
class TestDeadSessionTwoBaseSyncs(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp()
self.shas = build_two_base_sync_repo(self.tmp)
self.head_a = self.shas["head_a"]
self.head_b = self.shas["head_b"]
self.head_c = self.shas["head_c"]
def test_first_base_sync_dead_session_recovery(self):
"""AC1: dead session after first base sync (remote=B, local=A) recovers via merge-sync."""
lock = make_dead_lock(self.tmp, synced_pr_head=self.head_b)
obs = issue_lock_worktree.read_merge_sync_provenance(
self.tmp,
prior_head_sha=self.head_a,
synced_head_sha=self.head_b,
remote=REMOTE,
)
self.assertTrue(obs["is_merge_sync"], obs["reasons"])
self.assertTrue(obs["prior_is_ancestor"])
self.assertTrue(obs["synced_is_merge"])
self.assertTrue(obs["first_parent_reaches_prior"])
res = issue_lock_recovery.assess_dead_session_lock_recovery(
lock,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.tmp,
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
current_branch=BRANCH,
porcelain_status="",
head_sha=self.head_a,
remote_head_sha=self.head_b,
pr_head_sha=self.head_b,
pr_number=PR_NUMBER,
competing_live_locks=[],
candidate_branches=[BRANCH],
current_pid=os.getpid(),
remote_branch_exists=True,
sync_provenance=obs,
)
self.assertEqual(res["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED, res["reasons"])
self.assertEqual(
res["evidence"]["head_relation"],
issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED,
)
self.assertEqual(res["evidence"]["accepted_head"], self.head_b)
def test_second_base_sync_head_refresh_after_recovery(self):
"""AC2: after recovery, second base sync (remote B -> C) updates durable lock head."""
lock_dir = tempfile.mkdtemp()
lock_data = make_dead_lock(self.tmp, synced_pr_head=self.head_b)
# Rebind to current PID as gitea_lock_issue does upon sanctioned recovery
lock_data["session_pid"] = os.getpid()
lock_data["pid"] = os.getpid()
issue_lock_store.save_lock_file(
issue_lock_store.lock_file_path(
remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, lock_dir=lock_dir
),
lock_data,
)
refresh_res = issue_lock_store.apply_durable_lock_head_refresh(
remote=REMOTE,
org=ORG,
repo=REPO,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.tmp,
pr_number=PR_NUMBER,
identity=IDENTITY,
profile=PROFILE,
current_pid=os.getpid(),
expected_old_head=self.head_b,
new_head=self.head_c,
synced_at=future_ts(0),
base_head=self.shas["master_2"],
lock_dir=lock_dir,
)
self.assertTrue(refresh_res["refreshed"], refresh_res["reasons"])
self.assertTrue(refresh_res["read_after_write_ok"])
loaded = issue_lock_store.load_issue_lock(
remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, lock_dir=lock_dir
)
self.assertEqual(loaded.get("synced_pr_head"), self.head_c)
def test_dirty_worktree_blocks_recovery(self):
"""AC3: tracked dirty edits block dead-session recovery."""
lock = make_dead_lock(self.tmp)
obs = issue_lock_worktree.read_merge_sync_provenance(
self.tmp, prior_head_sha=self.head_a, synced_head_sha=self.head_b
)
res = issue_lock_recovery.assess_dead_session_lock_recovery(
lock,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.tmp,
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
current_branch=BRANCH,
porcelain_status=" M feature.txt\n",
head_sha=self.head_a,
remote_head_sha=self.head_b,
pr_head_sha=self.head_b,
pr_number=PR_NUMBER,
competing_live_locks=[],
candidate_branches=[BRANCH],
current_pid=os.getpid(),
remote_branch_exists=True,
sync_provenance=obs,
)
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
self.assertTrue(any("dirty" in r or "tracked" in r for r in res["reasons"]))
def test_foreign_reclaimer_blocks_recovery(self):
"""AC4: a foreign claimant cannot recover a dead-session lock."""
lock = make_dead_lock(self.tmp)
obs = issue_lock_worktree.read_merge_sync_provenance(
self.tmp, prior_head_sha=self.head_a, synced_head_sha=self.head_b
)
res = issue_lock_recovery.assess_dead_session_lock_recovery(
lock,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.tmp,
remote=REMOTE,
org=ORG,
repo=REPO,
identity="intruder-user",
profile=PROFILE,
current_branch=BRANCH,
porcelain_status="",
head_sha=self.head_a,
remote_head_sha=self.head_b,
pr_head_sha=self.head_b,
pr_number=PR_NUMBER,
competing_live_locks=[],
candidate_branches=[BRANCH],
current_pid=os.getpid(),
remote_branch_exists=True,
sync_provenance=obs,
)
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
def test_non_merge_rebased_remote_head_blocks_recovery(self):
"""AC5: a rebased/force-pushed remote head (not a merge commit) is refused."""
lock = make_dead_lock(self.tmp)
obs = issue_lock_worktree.read_merge_sync_provenance(
self.tmp, prior_head_sha=self.head_a, synced_head_sha=self.shas["rebased_head"]
)
self.assertFalse(obs["is_merge_sync"])
res = issue_lock_recovery.assess_dead_session_lock_recovery(
lock,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.tmp,
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
current_branch=BRANCH,
porcelain_status="",
head_sha=self.head_a,
remote_head_sha=self.shas["rebased_head"],
pr_head_sha=self.shas["rebased_head"],
pr_number=PR_NUMBER,
competing_live_locks=[],
candidate_branches=[BRANCH],
current_pid=os.getpid(),
remote_branch_exists=True,
sync_provenance=obs,
)
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
if __name__ == "__main__":
unittest.main()
+468
View File
@@ -0,0 +1,468 @@
"""Tests for the unified web-console inventory API (#636).
Covers the four cases the issue names — empty, populated, partial failure, and
the no-false-unowned invariant — plus redaction, collision detection, the
resource-split routes, and read-only guarantees against a real control-plane
database and real durable lock files.
"""
import json
import os
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
import control_plane_db
from webui.app import create_app
from webui import inventory
def _iso(dt: datetime) -> str:
return dt.astimezone(timezone.utc).isoformat()
def _write_lock(lock_dir: str, name: str, payload: dict) -> str:
path = os.path.join(lock_dir, name)
with open(path, "w", encoding="utf-8") as handle:
json.dump(payload, handle)
return path
def _live_lock_payload(
*,
issue_number: int,
branch: str,
worktree_path: str,
pid: int,
username: str = "jcwalker3",
profile: str = "prgs-author",
) -> dict:
now = datetime.now(timezone.utc)
future = now + timedelta(hours=2)
return {
"branch_name": branch,
"issue_number": issue_number,
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"remote": "prgs",
"pid": pid,
"session_pid": pid,
"lock_generation": 1,
"worktree_path": worktree_path,
"claimant": {"username": username, "profile": profile},
"work_lease": {
"branch": branch,
"issue_number": issue_number,
"operation_type": "author_issue_work",
"created_at": _iso(now),
"expires_at": _iso(future),
"last_heartbeat_at": _iso(now),
"claimant": {"username": username, "profile": profile},
},
}
class _FixtureMixin(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.tmp = self._tmp.name
self.lock_dir = os.path.join(self.tmp, "locks")
os.makedirs(self.lock_dir, mode=0o700)
self.db_path = os.path.join(self.tmp, "control_plane.db")
self.addCleanup(self._tmp.cleanup)
def _seed_db(self) -> control_plane_db.ControlPlaneDB:
db = control_plane_db.ControlPlaneDB(self.db_path)
db.upsert_session(
session_id="prgs-author-1",
role="author",
profile="prgs-author",
namespace="gitea-author",
pid=os.getpid(),
)
db.upsert_work_item(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
kind="issue",
number=636,
)
db.assign_and_lease(
session_id="prgs-author-1",
role="author",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
kind="issue",
number=636,
)
return db
class TestRedaction(unittest.TestCase):
def test_redact_path_collapses_home(self):
home = os.path.expanduser("~")
self.assertEqual(
inventory.redact_path(f"{home}/Development/Gitea-Tools"),
"~/Development/Gitea-Tools",
)
def test_redact_url_strips_userinfo_and_query(self):
self.assertEqual(
inventory.redact_url("https://user:[email protected]/api?token=abc"),
"https://gitea.prgs.cc/api",
)
def test_scrub_drops_credential_keys(self):
scrubbed = inventory.scrub(
{"token": "abc123", "api_key": "k", "profile": "prgs-author"}
)
self.assertEqual(scrubbed["token"], "[redacted]")
self.assertEqual(scrubbed["api_key"], "[redacted]")
self.assertEqual(scrubbed["profile"], "prgs-author")
def test_scrub_is_recursive_and_never_raises(self):
class Weird:
def __repr__(self) -> str:
return "weird-obj"
out = inventory.scrub({"nested": [{"password": "p", "obj": Weird()}]})
self.assertEqual(out["nested"][0]["password"], "[redacted]")
self.assertEqual(out["nested"][0]["obj"], "weird-obj")
class TestEmptyInventory(_FixtureMixin):
def test_empty_db_and_locks_degrade_without_raising(self):
# No DB file, no locks: sessions/leases unavailable, locks ok+empty.
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
sessions = snap.section("sessions")
leases = snap.section("leases")
locks = snap.section("locks")
self.assertEqual(sessions.status, inventory.STATUS_UNAVAILABLE)
self.assertEqual(leases.status, inventory.STATUS_UNAVAILABLE)
self.assertEqual(locks.status, inventory.STATUS_OK)
self.assertEqual(len(locks.items), 0)
# Ownership authority is incomplete because the DB is missing.
self.assertFalse(snap.ownership_authority_complete)
self.assertEqual(snap.collisions, ())
def test_empty_db_present_but_unpopulated(self):
control_plane_db.ControlPlaneDB(self.db_path) # creates schema, no rows
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
self.assertEqual(len(snap.section("sessions").items), 0)
self.assertEqual(snap.section("leases").status, inventory.STATUS_OK)
self.assertTrue(snap.ownership_authority_complete)
class TestPopulatedInventory(_FixtureMixin):
def test_sections_populated_and_correlated(self):
self._seed_db()
wt = f"{self.tmp}/branches/issue-636-inventory-api"
_write_lock(
self.lock_dir,
"prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
_live_lock_payload(
issue_number=636,
branch="feat/issue-636-inventory-api",
worktree_path=wt,
pid=os.getpid(),
),
)
hygiene = _StubHygiene(
entries=(
_StubEntry(
rel_path="branches/issue-636-inventory-api",
branch="feat/issue-636-inventory-api",
classification="active-issue",
),
)
)
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: hygiene,
)
self.assertTrue(snap.ownership_authority_complete)
self.assertEqual(len(snap.section("sessions").items), 1)
self.assertEqual(len(snap.section("leases").items), 1)
self.assertEqual(len(snap.section("locks").items), 1)
self.assertEqual(len(snap.section("worktrees").items), 1)
# The lease, lock, and worktree for #636 correlate onto one row.
row = next(r for r in snap.correlations if r["issue_number"] == 636)
self.assertEqual(row["branch"], "feat/issue-636-inventory-api")
self.assertTrue(row["lock_live"])
self.assertEqual(row["worktree_classification"], "active-issue")
self.assertEqual(len(row["lease_ids"]), 1)
# No collision: live lock, live pid, matching worktree.
self.assertEqual(snap.collisions, ())
def test_serialized_payload_declares_field_authority(self):
self._seed_db()
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
payload = inventory.snapshot_to_dict(snap)
self.assertEqual(payload["api_version"], "v1")
self.assertEqual(payload["schema_version"], 1)
self.assertEqual(payload["field_authority"]["sessions"], "control_plane_db")
self.assertEqual(payload["field_authority"]["locks"], "filesystem")
self.assertIn("sessions", payload["sections"])
class TestPartialFailure(_FixtureMixin):
def test_worktree_scan_failure_degrades_only_that_section(self):
self._seed_db()
def _boom():
raise RuntimeError("git worktree list exploded")
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=_boom,
)
self.assertEqual(
snap.section("worktrees").status, inventory.STATUS_UNAVAILABLE
)
self.assertIn("exploded", snap.section("worktrees").reason)
# DB-backed sections still healthy.
self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
self.assertIn("worktrees", snap.degraded_sections)
def test_degraded_ownership_suppresses_unowned_claim(self):
# DB absent → sessions/leases unavailable → ownership incomplete even
# though a lock exists and could look "unclaimed" by the DB alone.
_write_lock(
self.lock_dir,
"prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
_live_lock_payload(
issue_number=636,
branch="feat/issue-636-inventory-api",
worktree_path=f"{self.tmp}/wt",
pid=os.getpid(),
),
)
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
self.assertFalse(snap.ownership_authority_complete)
payload = inventory.snapshot_to_dict(snap)
self.assertIn("may be treated as unowned", payload["ownership_note"])
class TestCollisionDetection(_FixtureMixin):
def test_live_lock_dead_owner_flagged(self):
_write_lock(
self.lock_dir,
"prgs-Scaled-Tech-Consulting-Gitea-Tools-700.json",
_live_lock_payload(
issue_number=700,
branch="feat/issue-700-x",
worktree_path=f"{self.tmp}/wt700",
pid=999_999_999, # not a running pid
),
)
control_plane_db.ControlPlaneDB(self.db_path) # empty but present
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
kinds = {c.kind for c in snap.collisions}
self.assertIn("live-lock-dead-owner", kinds)
# Also lock-without-worktree, since no worktree carries the branch.
self.assertIn("lock-without-worktree", kinds)
def test_duplicate_live_lock_on_same_branch(self):
for issue in (800, 801):
_write_lock(
self.lock_dir,
f"prgs-Scaled-Tech-Consulting-Gitea-Tools-{issue}.json",
_live_lock_payload(
issue_number=issue,
branch="feat/issue-800-shared",
worktree_path=f"{self.tmp}/wt{issue}",
pid=os.getpid(),
),
)
control_plane_db.ControlPlaneDB(self.db_path)
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
self.assertIn(
"duplicate-live-lock", {c.kind for c in snap.collisions}
)
def test_no_collision_when_sections_degraded(self):
# locks ok but worktrees unavailable → lock-without-worktree must NOT
# be asserted (a missing scan is not a missing worktree).
_write_lock(
self.lock_dir,
"prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
_live_lock_payload(
issue_number=636,
branch="feat/issue-636-inventory-api",
worktree_path=f"{self.tmp}/wt",
pid=os.getpid(),
),
)
control_plane_db.ControlPlaneDB(self.db_path)
def _boom():
raise RuntimeError("scan down")
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=_boom,
)
self.assertNotIn(
"lock-without-worktree", {c.kind for c in snap.collisions}
)
class TestSectionInclude(_FixtureMixin):
def test_include_restricts_scanned_sections(self):
self._seed_db()
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
include=("locks",),
)
self.assertIsNotNone(snap.section("locks"))
self.assertIsNone(snap.section("sessions"))
self.assertIsNone(snap.section("worktrees"))
class TestRoutes(_FixtureMixin):
def setUp(self) -> None:
super().setUp()
# Point the loaders at the fixture DB and lock dir via env, and stub
# the worktree scan so the route does not shell out to git.
self._prev_env = {
"GITEA_CONTROL_PLANE_DB": os.environ.get("GITEA_CONTROL_PLANE_DB"),
"GITEA_ISSUE_LOCK_DIR": os.environ.get("GITEA_ISSUE_LOCK_DIR"),
"WEBUI_TEST_OFFLINE": os.environ.get("WEBUI_TEST_OFFLINE"),
}
os.environ["GITEA_CONTROL_PLANE_DB"] = self.db_path
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
os.environ["WEBUI_TEST_OFFLINE"] = "1"
self._seed_db()
self.client = TestClient(create_app())
def tearDown(self) -> None:
for key, value in self._prev_env.items():
if value is None:
os.environ.pop(key, None)
else:
os.environ[key] = value
def test_inventory_route_returns_versioned_payload(self):
resp = self.client.get("/api/v1/inventory")
self.assertEqual(resp.status_code, 200)
body = resp.json()
self.assertEqual(body["api_version"], "v1")
self.assertIn("sessions", body["sections"])
self.assertIn("field_authority", body)
def test_section_route_restricts_and_labels(self):
resp = self.client.get("/api/v1/inventory/locks")
self.assertEqual(resp.status_code, 200)
body = resp.json()
self.assertEqual(body["requested_section"], "locks")
self.assertIn("locks", body["sections"])
self.assertNotIn("sessions", body["sections"])
def test_unknown_section_is_404(self):
resp = self.client.get("/api/v1/inventory/bogus")
self.assertEqual(resp.status_code, 404)
self.assertEqual(resp.json()["error"], "unknown_section")
def test_inventory_route_rejects_post(self):
resp = self.client.post("/api/v1/inventory")
self.assertEqual(resp.status_code, 405)
class TestReadOnly(_FixtureMixin):
def test_snapshot_does_not_create_db_file(self):
missing = os.path.join(self.tmp, "does-not-exist.db")
inventory.load_inventory_snapshot(
db_path=missing,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
self.assertFalse(os.path.exists(missing))
def test_readonly_connection_refuses_write(self):
self._seed_db()
conn = inventory._open_readonly(self.db_path)
try:
with self.assertRaises(Exception):
conn.execute(
"INSERT INTO sessions(session_id, role, started_at, "
"last_heartbeat_at, status) VALUES ('x','author',"
"'t','t','active')"
)
conn.commit()
finally:
conn.close()
# ── lightweight stand-ins for the #432 hygiene snapshot ──────────────────────
class _StubEntry:
def __init__(
self,
*,
rel_path: str,
branch: str | None = None,
classification: str = "stale-clean",
head_sha: str | None = "abc123",
dirty_tracked: int = 0,
dirty_untracked: bool = False,
detached: bool = False,
registered_worktree: bool = True,
notes: str = "",
) -> None:
self.rel_path = rel_path
self.folder_name = rel_path.split("/", 1)[-1]
self.branch = branch
self.classification = classification
self.head_sha = head_sha
self.dirty_tracked = dirty_tracked
self.dirty_untracked = dirty_untracked
self.detached = detached
self.registered_worktree = registered_worktree
self.notes = notes
class _StubHygiene:
def __init__(self, *, entries=(), scan_error=None) -> None:
self.entries = tuple(entries)
self.scan_error = scan_error
if __name__ == "__main__":
unittest.main()
+35
View File
@@ -46,6 +46,11 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
from webui.worktree_views import render_worktrees_page
from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
from webui.runtime_views import render_runtime_page
from webui.inventory import (
SECTION_NAMES as _INVENTORY_SECTIONS,
load_inventory_snapshot,
snapshot_to_dict as inventory_snapshot_to_dict,
)
from webui.timeline import load_timeline, snapshot_to_dict as timeline_snapshot_to_dict
from webui.analytics_loader import (
load_analytics,
@@ -466,6 +471,30 @@ async def api_action_attempt(request: Request) -> JSONResponse:
return JSONResponse(result, status_code=status)
async def api_inventory(_request: Request) -> JSONResponse:
"""Unified read-only session/lease/lock/worktree inventory (#636)."""
snapshot = load_inventory_snapshot()
return JSONResponse(inventory_snapshot_to_dict(snapshot))
async def api_inventory_section(request: Request) -> JSONResponse:
"""Resource-split view: one inventory section under the shared schema."""
section = request.path_params["section"]
if section not in _INVENTORY_SECTIONS:
return JSONResponse(
{
"error": "unknown_section",
"detail": f"no inventory section named {section!r}",
"available": sorted(_INVENTORY_SECTIONS),
},
status_code=404,
)
snapshot = load_inventory_snapshot(include=(section,))
payload = inventory_snapshot_to_dict(snapshot)
payload["requested_section"] = section
return JSONResponse(payload)
async def api_console_security_model(_request: Request) -> JSONResponse:
"""Read-only publication of the #633 authorization/redaction/audit model."""
return JSONResponse({
@@ -744,6 +773,12 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
methods=["POST"],
),
Route("/api/leases", api_leases, methods=["GET"]),
Route("/api/v1/inventory", api_inventory, methods=["GET"]),
Route(
"/api/v1/inventory/{section}",
api_inventory_section,
methods=["GET"],
),
Route(
"/api/console/security-model",
api_console_security_model,
+952
View File
@@ -0,0 +1,952 @@
"""Unified session/lease/lock/worktree inventory for the web console (#636).
Leases (#433), worktrees (#432), and runtime (#430) each ship their own MVP
view, each with its own shape and its own idea of what "owned" means. A
traffic-control or recovery operator has to read all three and correlate them
by hand, which is exactly the step that goes wrong under collision pressure.
This module aggregates them into one versioned, read-only snapshot so the
console, and any worker asking "what is safe to do next", read the same
inventory from the same authority.
Field authority is explicit and never blended. Every section declares where its
rows came from:
* ``control_plane_db`` — the #613 substrate: sessions, leases, assignments.
Authoritative for *exclusive ownership* (#600/#601).
* ``filesystem`` — durable per-issue lock files (:mod:`issue_lock_store`) and
registered git worktrees. Authoritative for *what exists on this machine*.
* ``gitea`` — remote issue/PR state, reached only through existing loaders.
Safety invariants:
* **Read-only.** The control-plane database is opened through a ``mode=ro``
URI. :class:`control_plane_db.ControlPlaneDB` creates directories and runs
migrations in its constructor, which an inventory read must never do, so this
module talks to sqlite directly rather than through that class.
* **Fail-soft, never fail-silent.** A subsystem that cannot be read degrades to
a section carrying ``status`` and ``reason``. It never raises, and it never
produces an empty list that reads like "nothing is there".
* **Never invent active ownership.** This is the invariant that matters most.
A degraded ownership source sets ``ownership_authority_complete`` false, and
while that flag is false no work item is reported unowned and no collision is
asserted. Absence of evidence is reported as absence of evidence.
* **Redaction at the boundary.** Absolute paths are collapsed against the home
directory, URLs lose userinfo and query strings, and no credential-shaped
value is emitted. No session token exists in these sources and none is read.
Phase 1 is read-only. Lease steal/release and worktree deletion are Phase 2+
and deliberately have no representation here, not even a disabled one.
"""
from __future__ import annotations
import os
import re
import sqlite3
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable
from urllib.parse import urlparse
import control_plane_db
import issue_lock_store
SCHEMA_VERSION = 1
API_VERSION = "v1"
#: Sections whose absence would make an ownership claim unprovable. If any of
#: these is not ``ok``, the snapshot refuses to describe anything as unowned.
OWNERSHIP_SECTIONS = ("sessions", "leases", "locks")
SECTION_NAMES = ("sessions", "leases", "locks", "worktrees", "namespaces")
STATUS_OK = "ok"
STATUS_DEGRADED = "degraded"
STATUS_UNAVAILABLE = "unavailable"
AUTHORITY_CONTROL_PLANE_DB = "control_plane_db"
AUTHORITY_FILESYSTEM = "filesystem"
AUTHORITY_GITEA = "gitea"
_CREDENTIAL_KEY_RE = re.compile(
r"(token|secret|password|passwd|api[_-]?key|authorization|bearer|credential)",
re.IGNORECASE,
)
_REDACTED = "[redacted]"
@dataclass(frozen=True)
class InventorySection:
"""One subsystem's contribution, with its authority and health."""
name: str
authority: str
status: str
items: tuple[dict[str, Any], ...] = ()
reason: str | None = None
scan_ms: float | None = None
@property
def ok(self) -> bool:
return self.status == STATUS_OK
def to_dict(self) -> dict[str, Any]:
return {
"name": self.name,
"authority": self.authority,
"status": self.status,
"count": len(self.items),
"reason": self.reason,
"scan_ms": self.scan_ms,
"items": [dict(item) for item in self.items],
}
@dataclass(frozen=True)
class CollisionSignal:
"""A detected conflict between two ownership records."""
kind: str
message: str
severity: str = "warning"
issue_number: int | None = None
branch: str | None = None
worktree_path: str | None = None
session_ids: tuple[str, ...] = ()
def to_dict(self) -> dict[str, Any]:
return {
"kind": self.kind,
"severity": self.severity,
"message": self.message,
"issue_number": self.issue_number,
"branch": self.branch,
"worktree_path": self.worktree_path,
"session_ids": list(self.session_ids),
}
@dataclass(frozen=True)
class InventorySnapshot:
"""Versioned aggregate of every inventory section."""
generated_at: str
sections: tuple[InventorySection, ...]
collisions: tuple[CollisionSignal, ...] = ()
correlations: tuple[dict[str, Any], ...] = ()
schema_version: int = SCHEMA_VERSION
api_version: str = API_VERSION
scan_ms: float | None = None
_section_index: dict[str, InventorySection] = field(
default_factory=dict, repr=False, compare=False
)
def section(self, name: str) -> InventorySection | None:
return self._section_index.get(name)
@property
def degraded_sections(self) -> tuple[str, ...]:
return tuple(s.name for s in self.sections if not s.ok)
@property
def ownership_authority_complete(self) -> bool:
"""True only when every ownership-bearing section read cleanly.
While this is false the snapshot must not describe any work item as
unowned: a lease the reader could not load is not an absent lease.
"""
for name in OWNERSHIP_SECTIONS:
section = self._section_index.get(name)
if section is None or not section.ok:
return False
return True
@property
def status(self) -> str:
if all(s.ok for s in self.sections):
return STATUS_OK
return STATUS_DEGRADED
# ── redaction ────────────────────────────────────────────────────────────────
def redact_path(path: str | None) -> str | None:
"""Collapse an absolute path against ``$HOME`` for browser display."""
if not path:
return path
text = str(path)
home = os.path.expanduser("~")
if home and home != "/" and text.startswith(home):
return "~" + text[len(home) :]
return text
def redact_url(value: str | None) -> str | None:
"""Strip userinfo and query string from a URL."""
if not value:
return value
text = str(value)
try:
parsed = urlparse(text)
except ValueError:
return _REDACTED
if not parsed.scheme or not parsed.netloc:
return text
netloc = parsed.hostname or ""
if parsed.port:
netloc = f"{netloc}:{parsed.port}"
rebuilt = f"{parsed.scheme}://{netloc}{parsed.path}"
return rebuilt.rstrip("/") or rebuilt
def scrub(value: Any, *, key: str | None = None) -> Any:
"""Recursively drop credential-shaped values and redact paths/URLs.
Never raises: an unexpected object degrades to its ``repr`` rather than
propagating out of a read-only view.
"""
if key and _CREDENTIAL_KEY_RE.search(key):
return _REDACTED
if isinstance(value, dict):
return {str(k): scrub(v, key=str(k)) for k, v in value.items()}
if isinstance(value, (list, tuple)):
return [scrub(v, key=key) for v in value]
if isinstance(value, str):
if value.startswith(("http://", "https://")):
return redact_url(value)
if value.startswith("/") or value.startswith("~"):
return redact_path(value)
return value
if isinstance(value, (int, float, bool)) or value is None:
return value
return repr(value)
# ── control-plane database (read-only) ───────────────────────────────────────
def _open_readonly(db_path: str) -> sqlite3.Connection:
"""Open the control-plane DB without creating or migrating anything."""
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=5)
conn.row_factory = sqlite3.Row
return conn
def _table_names(conn: sqlite3.Connection) -> set[str]:
rows = conn.execute(
"SELECT name FROM sqlite_master WHERE type = 'table'"
).fetchall()
return {str(row[0]) for row in rows}
def _load_cp_db_sections(
*,
db_path: str | None = None,
limit: int = 200,
) -> tuple[InventorySection, InventorySection]:
"""Return the ``sessions`` and ``leases`` sections from the #613 DB."""
path = (db_path or control_plane_db.default_db_path()).strip()
def _both_unavailable(reason: str) -> tuple[InventorySection, InventorySection]:
return (
InventorySection(
name="sessions",
authority=AUTHORITY_CONTROL_PLANE_DB,
status=STATUS_UNAVAILABLE,
reason=reason,
),
InventorySection(
name="leases",
authority=AUTHORITY_CONTROL_PLANE_DB,
status=STATUS_UNAVAILABLE,
reason=reason,
),
)
if not path:
return _both_unavailable("control-plane database path is not configured")
if not os.path.exists(path):
return _both_unavailable(
f"control-plane database not present at {redact_path(path)}; "
"no session or lease authority available"
)
started = time.perf_counter()
try:
conn = _open_readonly(path)
except sqlite3.Error as exc:
return _both_unavailable(f"control-plane database could not be opened: {exc}")
try:
tables = _table_names(conn)
if "sessions" not in tables or "leases" not in tables:
missing = sorted({"sessions", "leases"} - tables)
return _both_unavailable(
"control-plane database is missing required tables: "
+ ", ".join(missing)
)
session_rows = [
dict(row)
for row in conn.execute(
"SELECT session_id, role, profile, namespace, pid, started_at,"
" last_heartbeat_at, status FROM sessions"
" ORDER BY last_heartbeat_at DESC LIMIT ?",
(max(1, int(limit)),),
).fetchall()
]
has_work_items = "work_items" in tables
if has_work_items:
lease_sql = (
"SELECT l.lease_id, l.session_id, l.role, l.phase, l.status,"
" l.expires_at, w.remote, w.org, w.repo, w.kind AS work_kind,"
" w.number AS work_number, w.state AS work_state,"
" s.pid AS session_pid, s.profile AS session_profile,"
" s.namespace AS session_namespace, s.status AS session_status"
" FROM leases l"
" JOIN work_items w ON w.work_item_id = l.work_item_id"
" LEFT JOIN sessions s ON s.session_id = l.session_id"
" ORDER BY l.expires_at DESC LIMIT ?"
)
else:
lease_sql = (
"SELECT l.lease_id, l.session_id, l.role, l.phase, l.status,"
" l.expires_at FROM leases l"
" ORDER BY l.expires_at DESC LIMIT ?"
)
lease_rows = [
dict(row)
for row in conn.execute(lease_sql, (max(1, int(limit)),)).fetchall()
]
except sqlite3.Error as exc:
return _both_unavailable(f"control-plane database read failed: {exc}")
finally:
conn.close()
elapsed = (time.perf_counter() - started) * 1000.0
now = datetime.now(timezone.utc)
sessions = tuple(
scrub(
{
"session_id": row.get("session_id"),
"role": row.get("role"),
"profile": row.get("profile"),
"namespace": row.get("namespace"),
"pid": row.get("pid"),
"pid_alive": issue_lock_store.is_process_alive(row.get("pid")),
"started_at": row.get("started_at"),
"last_heartbeat_at": row.get("last_heartbeat_at"),
"status": row.get("status"),
}
)
for row in session_rows
)
leases = tuple(
scrub(
{
"lease_id": row.get("lease_id"),
"session_id": row.get("session_id"),
"role": row.get("role"),
"phase": row.get("phase"),
"status": row.get("status"),
"expires_at": row.get("expires_at"),
"expired": _is_expired(row.get("expires_at"), now=now),
"remote": row.get("remote"),
"org": row.get("org"),
"repo": row.get("repo"),
"work_kind": row.get("work_kind"),
"work_number": row.get("work_number"),
"work_state": row.get("work_state"),
"session_pid": row.get("session_pid"),
"session_profile": row.get("session_profile"),
"session_namespace": row.get("session_namespace"),
"session_status": row.get("session_status"),
}
)
for row in lease_rows
)
degraded_reason = (
None
if has_work_items
else "work_items table absent; lease rows carry no work linkage"
)
lease_status = STATUS_OK if has_work_items else STATUS_DEGRADED
return (
InventorySection(
name="sessions",
authority=AUTHORITY_CONTROL_PLANE_DB,
status=STATUS_OK,
items=sessions,
scan_ms=round(elapsed, 3),
),
InventorySection(
name="leases",
authority=AUTHORITY_CONTROL_PLANE_DB,
status=lease_status,
items=leases,
reason=degraded_reason,
scan_ms=round(elapsed, 3),
),
)
def _is_expired(expires_at: str | None, *, now: datetime) -> bool | None:
if not expires_at:
return None
text = str(expires_at).strip().replace("Z", "+00:00")
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed <= now
# ── durable issue locks (filesystem) ─────────────────────────────────────────
def _load_locks_section(*, lock_dir: str | None = None) -> InventorySection:
started = time.perf_counter()
try:
paths = issue_lock_store.iter_lock_files(lock_dir)
except OSError as exc:
return InventorySection(
name="locks",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_UNAVAILABLE,
reason=f"issue lock directory could not be listed: {exc}",
)
items: list[dict[str, Any]] = []
unreadable = 0
for path in paths:
try:
record = issue_lock_store.read_lock_file(path)
except (OSError, ValueError):
unreadable += 1
continue
if not record:
unreadable += 1
continue
try:
freshness = issue_lock_store.assess_lock_freshness(record)
except Exception: # noqa: BLE001 — a read-only view never raises
freshness = {"status": "unknown", "live": False, "stale": False}
claimant = record.get("claimant") or (
(record.get("work_lease") or {}).get("claimant") or {}
)
items.append(
scrub(
{
"issue_number": record.get("issue_number"),
"branch_name": record.get("branch_name"),
"remote": record.get("remote"),
"org": record.get("org"),
"repo": record.get("repo"),
"worktree_path": record.get("worktree_path"),
"pid": record.get("session_pid") or record.get("pid"),
"pid_alive": issue_lock_store.is_process_alive(
record.get("session_pid") or record.get("pid")
),
"claimant_username": (claimant or {}).get("username"),
"claimant_profile": (claimant or {}).get("profile"),
"lock_generation": record.get("lock_generation"),
"freshness_status": freshness.get("status"),
"live": bool(freshness.get("live")),
"stale": bool(freshness.get("stale")),
"freshness_reason": freshness.get("reason"),
"lock_path": record.get("lock_file_path") or path,
}
)
)
elapsed = (time.perf_counter() - started) * 1000.0
reason = (
f"{unreadable} lock file(s) were unreadable and are not represented"
if unreadable
else None
)
return InventorySection(
name="locks",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_DEGRADED if unreadable else STATUS_OK,
items=tuple(items),
reason=reason,
scan_ms=round(elapsed, 3),
)
# ── worktrees (filesystem, via the #432 scanner) ─────────────────────────────
def _load_worktrees_section(
*, load_hygiene: Callable[[], Any] | None = None
) -> InventorySection:
started = time.perf_counter()
try:
loader = load_hygiene
if loader is None:
from webui.worktree_scanner import load_hygiene_snapshot
loader = load_hygiene_snapshot
snapshot = loader()
except Exception as exc: # noqa: BLE001 — fail soft, never fail the request
return InventorySection(
name="worktrees",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_UNAVAILABLE,
reason=f"worktree scan failed: {exc}",
)
items = tuple(
scrub(
{
"rel_path": entry.rel_path,
"folder_name": entry.folder_name,
"classification": entry.classification,
"branch": entry.branch,
"head_sha": entry.head_sha,
"dirty_tracked": entry.dirty_tracked,
"dirty_untracked": entry.dirty_untracked,
"detached": entry.detached,
"registered_worktree": entry.registered_worktree,
"notes": entry.notes,
}
)
for entry in snapshot.entries
)
scan_error = getattr(snapshot, "scan_error", None)
elapsed = (time.perf_counter() - started) * 1000.0
return InventorySection(
name="worktrees",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_DEGRADED if scan_error else STATUS_OK,
items=items,
reason=scan_error,
scan_ms=round(elapsed, 3),
)
# ── namespaces / capability summary ──────────────────────────────────────────
def _load_namespaces_section() -> InventorySection:
started = time.perf_counter()
try:
from gitea_auth import get_profile
profile = get_profile() or {}
except Exception as exc: # noqa: BLE001
return InventorySection(
name="namespaces",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_UNAVAILABLE,
reason=f"active profile could not be resolved: {exc}",
)
allowed = list(profile.get("allowed_operations") or [])
forbidden = list(profile.get("forbidden_operations") or [])
profile_name = str(profile.get("profile_name") or "")
namespace = None
try:
import role_namespace_gate
namespace = role_namespace_gate.infer_mcp_namespace(profile_name)
except Exception: # noqa: BLE001 — namespace inference is advisory
namespace = None
item = scrub(
{
"profile_name": profile_name,
"role": profile.get("role"),
"mcp_namespace": namespace,
"allowed_operations": sorted(allowed),
"forbidden_operations": sorted(forbidden),
"capability_summary": {
"can_author": "gitea.pr.create" in allowed,
"can_review": "gitea.pr.approve" in allowed,
"can_merge": "gitea.pr.merge" in allowed,
"can_close_pr": "gitea.pr.close" in allowed,
},
"active": True,
}
)
elapsed = (time.perf_counter() - started) * 1000.0
return InventorySection(
name="namespaces",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_OK,
items=(item,),
reason=(
"only the profile serving this web process is observable; other "
"namespaces are not enumerable from here"
),
scan_ms=round(elapsed, 3),
)
# ── correlation and collision detection ──────────────────────────────────────
def _issue_from_branch(branch: str | None) -> int | None:
match = re.search(r"issue-(\d+)", str(branch or ""), re.IGNORECASE)
return int(match.group(1)) if match else None
def correlate(
*,
leases: InventorySection,
locks: InventorySection,
worktrees: InventorySection,
sessions: InventorySection,
) -> tuple[tuple[dict[str, Any], ...], tuple[CollisionSignal, ...]]:
"""Join lease owner ↔ lock ↔ worktree ↔ namespace where evidence allows.
Correlation rows are emitted from whatever sections did load. Collision
signals are only emitted from sections that are ``ok``: a conflict inferred
from a partially-read source would be a false accusation.
"""
correlations: list[dict[str, Any]] = []
collisions: list[CollisionSignal] = []
worktree_by_branch: dict[str, dict[str, Any]] = {}
for entry in worktrees.items:
branch = (entry.get("branch") or "").strip()
if branch:
worktree_by_branch.setdefault(branch, entry)
session_by_id = {
str(s.get("session_id")): s for s in sessions.items if s.get("session_id")
}
# Lock-centred rows: a durable lock names an issue, a branch, and a worktree.
for lock in locks.items:
branch = (lock.get("branch_name") or "").strip()
worktree = worktree_by_branch.get(branch)
matching_leases = [
lease
for lease in leases.items
if lease.get("work_kind") == "issue"
and lease.get("work_number") == lock.get("issue_number")
]
correlations.append(
{
"issue_number": lock.get("issue_number"),
"branch": branch or None,
"lock_live": bool(lock.get("live")),
"lock_claimant": lock.get("claimant_profile"),
"lock_pid": lock.get("pid"),
"lock_pid_alive": lock.get("pid_alive"),
"worktree_rel_path": (worktree or {}).get("rel_path"),
"worktree_classification": (worktree or {}).get("classification"),
"worktree_registered": (worktree or {}).get("registered_worktree"),
"lease_ids": [
lease.get("lease_id")
for lease in matching_leases
if lease.get("lease_id")
],
"lease_sessions": [
lease.get("session_id")
for lease in matching_leases
if lease.get("session_id")
],
}
)
if locks.ok and worktrees.ok:
# A claim whose lease window is still open but has no registered
# worktree is an anomaly regardless of whether its pid is alive; a
# fully time-expired lease is on its way out and is not flagged.
if (
lock.get("freshness_status") != "expired"
and branch
and worktree is None
):
collisions.append(
CollisionSignal(
kind="lock-without-worktree",
severity="warning",
issue_number=lock.get("issue_number"),
branch=branch,
worktree_path=lock.get("worktree_path"),
message=(
f"Live lock on issue #{lock.get('issue_number')} names "
f"branch {branch!r} but no registered worktree carries "
"that branch (#404)"
),
)
)
if locks.ok:
# A lock whose recorded pid is gone is held by nobody: a clean #753
# dead-session recovery candidate. Subclassify by the lease window,
# because the two cases need different operator urgency. When the
# window is still open the lock would read as live to a naive
# timestamp check even though the owner is dead — the more dangerous
# case — so it is flagged distinctly from a fully time-expired lease.
if lock.get("pid_alive") is False and lock.get("stale"):
if lock.get("freshness_status") == "expired":
collisions.append(
CollisionSignal(
kind="stale-lock-dead-owner",
severity="warning",
issue_number=lock.get("issue_number"),
branch=branch or None,
message=(
f"Lock on issue #{lock.get('issue_number')} is stale "
f"and its recorded pid {lock.get('pid')} is not running "
"(#753 dead-session recovery candidate)"
),
)
)
else:
collisions.append(
CollisionSignal(
kind="live-lock-dead-owner",
severity="warning",
issue_number=lock.get("issue_number"),
branch=branch or None,
message=(
f"Lock on issue #{lock.get('issue_number')} has an "
"unexpired lease but its recorded pid "
f"{lock.get('pid')} is not running; it would read as "
"live to a timestamp check (#753 dead-session recovery "
"candidate)"
),
)
)
# The #635 trap: the lease has expired but the recorded pid is a
# still-running daemon, so neither dead-pid reclaim nor exact-owner
# renewal applies. This is the collision an operator must see.
elif (
lock.get("freshness_status") == "expired"
and lock.get("pid_alive") is True
):
collisions.append(
CollisionSignal(
kind="expired-lock-live-owner",
severity="error",
issue_number=lock.get("issue_number"),
branch=branch or None,
message=(
f"Lock on issue #{lock.get('issue_number')} has an expired "
f"lease but its recorded pid {lock.get('pid')} is still "
"running (daemon-pid deadlock; needs an operator decision, "
"#635/#760)"
),
)
)
# Two live locks on one branch, or two active leases on one work item.
if locks.ok:
by_branch: dict[str, list[dict[str, Any]]] = {}
for lock in locks.items:
if not lock.get("live"):
continue
branch = (lock.get("branch_name") or "").strip()
if branch:
by_branch.setdefault(branch, []).append(lock)
for branch, entries in sorted(by_branch.items()):
if len(entries) > 1:
collisions.append(
CollisionSignal(
kind="duplicate-live-lock",
severity="error",
branch=branch,
message=(
f"{len(entries)} live locks name branch {branch!r}: "
"issues "
+ ", ".join(
f"#{e.get('issue_number')}" for e in entries
)
),
)
)
if leases.ok:
by_work: dict[tuple[str, int], list[dict[str, Any]]] = {}
for lease in leases.items:
if str(lease.get("status") or "").lower() != "active":
continue
kind = str(lease.get("work_kind") or "").strip().lower()
number = lease.get("work_number")
if not kind or number is None:
continue
by_work.setdefault((kind, int(number)), []).append(lease)
for (kind, number), entries in sorted(by_work.items()):
sessions_held = {
str(e.get("session_id")) for e in entries if e.get("session_id")
}
if len(sessions_held) > 1:
collisions.append(
CollisionSignal(
kind="concurrent-active-lease",
severity="error",
issue_number=number if kind == "issue" else None,
session_ids=tuple(sorted(sessions_held)),
message=(
f"{len(sessions_held)} sessions hold an active lease on "
f"{kind} #{number}"
),
)
)
for entry in entries:
if entry.get("expired") is True:
collisions.append(
CollisionSignal(
kind="active-lease-past-expiry",
severity="warning",
issue_number=number if kind == "issue" else None,
session_ids=(
(str(entry.get("session_id")),)
if entry.get("session_id")
else ()
),
message=(
f"Lease {entry.get('lease_id')} on {kind} #{number} "
"is still marked active past its expiry"
),
)
)
# A lease whose owning session is gone is an orphan, not free work.
if leases.ok and sessions.ok:
for lease in leases.items:
if str(lease.get("status") or "").lower() != "active":
continue
session_id = str(lease.get("session_id") or "")
if session_id and session_id not in session_by_id:
collisions.append(
CollisionSignal(
kind="orphan-lease",
severity="error",
session_ids=(session_id,),
message=(
f"Active lease {lease.get('lease_id')} names session "
f"{session_id}, which has no session record"
),
)
)
return tuple(correlations), tuple(collisions)
# ── snapshot assembly ────────────────────────────────────────────────────────
def load_inventory_snapshot(
*,
db_path: str | None = None,
lock_dir: str | None = None,
load_hygiene: Callable[[], Any] | None = None,
include: tuple[str, ...] | None = None,
) -> InventorySnapshot:
"""Build the unified inventory snapshot.
Every section is loaded independently and fails soft. *include* restricts
the sections that are scanned; omitted sections are simply absent rather
than reported as empty, so a resource-split request cannot be mistaken for
a whole-inventory answer.
"""
started = time.perf_counter()
wanted = tuple(include) if include else SECTION_NAMES
sections: list[InventorySection] = []
sessions_section: InventorySection | None = None
leases_section: InventorySection | None = None
if "sessions" in wanted or "leases" in wanted:
sessions_section, leases_section = _load_cp_db_sections(db_path=db_path)
if "sessions" in wanted:
sections.append(sessions_section)
if "leases" in wanted:
sections.append(leases_section)
locks_section = (
_load_locks_section(lock_dir=lock_dir)
if "locks" in wanted
else _empty_section("locks", AUTHORITY_FILESYSTEM)
)
if "locks" in wanted:
sections.append(locks_section)
worktrees_section = (
_load_worktrees_section(load_hygiene=load_hygiene)
if "worktrees" in wanted
else _empty_section("worktrees", AUTHORITY_FILESYSTEM)
)
if "worktrees" in wanted:
sections.append(worktrees_section)
if "namespaces" in wanted:
sections.append(_load_namespaces_section())
correlations, collisions = correlate(
leases=leases_section or _empty_section("leases", AUTHORITY_CONTROL_PLANE_DB),
locks=locks_section,
worktrees=worktrees_section,
sessions=sessions_section
or _empty_section("sessions", AUTHORITY_CONTROL_PLANE_DB),
)
elapsed = (time.perf_counter() - started) * 1000.0
index = {section.name: section for section in sections}
return InventorySnapshot(
generated_at=datetime.now(timezone.utc).isoformat(),
sections=tuple(sections),
collisions=collisions,
correlations=correlations,
scan_ms=round(elapsed, 3),
_section_index=index,
)
def _empty_section(name: str, authority: str) -> InventorySection:
"""A section that was not requested — never a claim that it is empty."""
return InventorySection(
name=name,
authority=authority,
status=STATUS_UNAVAILABLE,
reason="section not requested in this scan",
)
def snapshot_to_dict(snapshot: InventorySnapshot) -> dict[str, Any]:
"""Serialize the snapshot for the versioned API."""
return {
"api_version": snapshot.api_version,
"schema_version": snapshot.schema_version,
"generated_at": snapshot.generated_at,
"status": snapshot.status,
"scan_ms": snapshot.scan_ms,
"ownership_authority_complete": snapshot.ownership_authority_complete,
"ownership_note": (
"Every ownership source read cleanly; an item absent from leases "
"and locks is genuinely unclaimed."
if snapshot.ownership_authority_complete
else "One or more ownership sources are degraded; nothing in this "
"snapshot may be treated as unowned. Collisions are reported only "
"from sections that read cleanly."
),
"degraded_sections": list(snapshot.degraded_sections),
"field_authority": {
"sessions": AUTHORITY_CONTROL_PLANE_DB,
"leases": AUTHORITY_CONTROL_PLANE_DB,
"locks": AUTHORITY_FILESYSTEM,
"worktrees": AUTHORITY_FILESYSTEM,
"namespaces": AUTHORITY_FILESYSTEM,
},
"sections": {section.name: section.to_dict() for section in snapshot.sections},
"correlations": [dict(row) for row in snapshot.correlations],
"collisions": [signal.to_dict() for signal in snapshot.collisions],
}