sysadminandClaude Opus 4.8
a7a283f449
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]>
792 lines
26 KiB
Python
792 lines
26 KiB
Python
"""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
|