Merge branch 'master' into fix/issue-872-dead-session-two-base-syncs
This commit is contained in:
+400
-4
@@ -2065,6 +2065,7 @@ import allocator_dependencies # noqa: E402
|
||||
import dependency_graph # noqa: E402 # #784 durable dependency edges
|
||||
import control_plane_db # noqa: E402
|
||||
import lease_lifecycle # noqa: E402
|
||||
import lease_policy # noqa: E402
|
||||
import workflow_dashboard # noqa: E402 # #605 live queue/lease dashboard
|
||||
import restart_coordinator # noqa: E402 # #658 MCP restart coordinator/impact
|
||||
import incident_bridge # noqa: E402
|
||||
@@ -2310,7 +2311,6 @@ import canonical_comment_validator as ccv # noqa: E402
|
||||
# GITEA_ISSUE_LOCK_DIR, bound to the current MCP session via a per-PID pointer.
|
||||
# Legacy global path retained only for test/doc references — do not seed manually.
|
||||
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
|
||||
WORK_LEASE_TTL_HOURS = 4
|
||||
AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
|
||||
VALID_WORK_LEASE_OPERATIONS = frozenset({
|
||||
AUTHOR_ISSUE_WORK_LEASE,
|
||||
@@ -2626,7 +2626,12 @@ def _build_author_issue_work_lease(
|
||||
host: str | None,
|
||||
) -> dict:
|
||||
created = _work_lease_now()
|
||||
expires = created + timedelta(hours=WORK_LEASE_TTL_HOURS)
|
||||
# #790 Slice A: the window comes from the central policy, not a literal here.
|
||||
# It is also now a *sliding* window — the lease lives ``initial_ttl_minutes``
|
||||
# past its last valid heartbeat rather than a fixed four hours past its
|
||||
# creation, so an abandoned task stops holding the claim within one TTL.
|
||||
policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
|
||||
expires = created + timedelta(minutes=policy.initial_ttl_minutes)
|
||||
return {
|
||||
"operation_type": AUTHOR_ISSUE_WORK_LEASE,
|
||||
"issue_number": issue_number,
|
||||
@@ -2637,6 +2642,15 @@ def _build_author_issue_work_lease(
|
||||
"created_at": _work_lease_timestamp(created),
|
||||
"expires_at": _work_lease_timestamp(expires),
|
||||
"last_heartbeat_at": _work_lease_timestamp(created),
|
||||
# #790 AC-N1: the ownership key for this task. Distinct from the recorded
|
||||
# PID, which is the shared daemon and identifies no individual task.
|
||||
"task_session_id": issue_lock_store.mint_task_session_id(
|
||||
AUTHOR_ISSUE_WORK_LEASE
|
||||
),
|
||||
# #790 AC-N8: the explicit lifecycle marker. Its absence — never a
|
||||
# timestamp comparison — is what makes a lock legacy.
|
||||
"lifecycle_version": lease_policy.LIFECYCLE_HEARTBEAT_V1,
|
||||
"heartbeat_count": 1,
|
||||
}
|
||||
|
||||
|
||||
@@ -4407,6 +4421,135 @@ def gitea_lock_issue(
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_heartbeat_issue_lock(
|
||||
issue_number: int,
|
||||
branch_name: str,
|
||||
task_session_id: str | None = None,
|
||||
remote: str = "dadeschools",
|
||||
host: str | None = None,
|
||||
org: str | None = None,
|
||||
repo: str | None = None,
|
||||
worktree_path: str | None = None,
|
||||
expected_generation: int | None = None,
|
||||
) -> dict:
|
||||
"""Prove an owned author issue lease is still active (#790 Slice A).
|
||||
|
||||
The task-liveness signal the lifecycle was missing. Before this, an author
|
||||
lease carried a fixed four-hour expiry that nothing could shorten, and the
|
||||
only liveness evidence was the recorded PID — the long-lived MCP daemon,
|
||||
which stays alive across every task it serves and so proved nothing about
|
||||
whether the authoring task still held the work.
|
||||
|
||||
Each successful call slides the lease ``initial_ttl_minutes`` past *now*
|
||||
from the central policy, so an actively heartbeating session is never
|
||||
evicted while an abandoned one releases its claim within one TTL.
|
||||
|
||||
What this tool cannot do, by construction:
|
||||
|
||||
* **Acquire.** It refuses when no durable lock exists.
|
||||
* **Take over.** Exact issue, branch, realpath-normalized worktree,
|
||||
claimant username, claimant profile, and recorded task-session identifier
|
||||
must all match; a superseded session holding an older identifier is
|
||||
refused.
|
||||
* **Revive.** A lease already past its grace is not heartbeatable — that
|
||||
would let a session restore ownership it had stopped proving. It must use
|
||||
the sanctioned reclaim path, which mints a new generation.
|
||||
|
||||
A lock predating the heartbeat lifecycle is rebound rather than heartbeated:
|
||||
its exact owner is re-verified and a genuine task-session identifier and
|
||||
first heartbeat are minted (#790 AC-N8). The rebind is decided server-side
|
||||
from the durable lifecycle marker; there is no caller-facing switch.
|
||||
|
||||
Args:
|
||||
issue_number: The locked issue number.
|
||||
branch_name: The branch recorded on the lock.
|
||||
task_session_id: The identifier this session received when it acquired
|
||||
or rebound the lock. It is a fencing token, not an ownership
|
||||
assertion: it is compared against durable state and can only ever
|
||||
cause a refusal, never grant anything. Omitted only when rebinding a
|
||||
legacy lock, which has no identifier yet and mints one.
|
||||
remote: Known instance — 'dadeschools' or 'prgs'.
|
||||
host: Override the Gitea host.
|
||||
org: Override the owner/organization.
|
||||
repo: Override the repository name.
|
||||
worktree_path: Author worktree recorded on the lock.
|
||||
expected_generation: Optional fencing value. The per-issue flock already
|
||||
serializes the read and the write, so this is for a caller that
|
||||
wants to pin the generation it last observed across calls; a moved
|
||||
generation fails closed.
|
||||
|
||||
Returns:
|
||||
dict with 'success', 'performed', the sliding 'expires_at',
|
||||
'last_heartbeat_at', 'lock_generation', 'task_session_id', the applied
|
||||
'policy', and post-write 'freshness'; on refusal 'success'/'performed'
|
||||
False with 'reasons' naming exactly what did not match.
|
||||
"""
|
||||
blocked = _profile_permission_block(
|
||||
task_capability_map.required_permission("heartbeat_issue_lock"),
|
||||
issue_number=issue_number,
|
||||
remote=remote,
|
||||
host=host,
|
||||
org=org,
|
||||
repo=repo,
|
||||
org_explicit=org is not None,
|
||||
repo_explicit=repo is not None,
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
|
||||
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
|
||||
worktree_path, _canonical_local_git_root()
|
||||
)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
claimant = _work_lease_claimant(h)
|
||||
identity = claimant.get("username")
|
||||
profile = claimant.get("profile")
|
||||
|
||||
existing = _load_existing_issue_lock(
|
||||
remote=remote, org=o, repo=r, issue_number=issue_number
|
||||
)
|
||||
if not existing:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"issue_number": issue_number,
|
||||
"reasons": [
|
||||
f"no durable lock for issue #{issue_number}; heartbeat cannot "
|
||||
"acquire a claim (fail closed)"
|
||||
],
|
||||
}
|
||||
|
||||
if issue_lock_store.is_legacy_lease(existing):
|
||||
outcome = issue_lock_store.rebind_legacy_lock(
|
||||
remote=remote,
|
||||
org=o,
|
||||
repo=r,
|
||||
issue_number=issue_number,
|
||||
branch_name=branch_name,
|
||||
worktree_path=resolved_worktree,
|
||||
identity=identity,
|
||||
profile=profile,
|
||||
expected_generation=expected_generation,
|
||||
)
|
||||
outcome["operation"] = "legacy_rebind"
|
||||
return outcome
|
||||
|
||||
outcome = issue_lock_store.heartbeat_session_lock(
|
||||
remote=remote,
|
||||
org=o,
|
||||
repo=r,
|
||||
issue_number=issue_number,
|
||||
branch_name=branch_name,
|
||||
worktree_path=resolved_worktree,
|
||||
identity=identity,
|
||||
profile=profile,
|
||||
task_session_id=str(task_session_id or ""),
|
||||
expected_generation=expected_generation,
|
||||
)
|
||||
outcome["operation"] = "heartbeat"
|
||||
return outcome
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_recover_dirty_orphaned_issue_worktree(
|
||||
issue_number: int,
|
||||
@@ -4680,6 +4823,7 @@ def gitea_recover_dirty_orphaned_issue_worktree(
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_rebind_dirty_same_claimant_author_session(
|
||||
issue_number: int,
|
||||
@@ -4936,8 +5080,6 @@ def gitea_rebind_dirty_same_claimant_author_session(
|
||||
}
|
||||
return result
|
||||
|
||||
return result
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_assess_work_issue_duplicate(
|
||||
@@ -17881,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
|
||||
|
||||
|
||||
@@ -22329,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,
|
||||
|
||||
Reference in New Issue
Block a user