feat: first-class control-plane lease lifecycle (Closes #601)

Make active leases queryable workflow state with canonical adopt, release,
expire/reclaim, and abandon operations on the #613 control-plane DB substrate.

- lease_lifecycle.py: policy, proof gates, safe_next_action vocabulary
- control_plane_db: schema v3 provenance columns + list/inspect/adopt/abandon
- MCP tools: gitea_*_workflow_lease(s) for list/inspect/adopt/release/expire/abandon/reclaim
- issue_lock_store: sanctioned reclaim of expired author locks (dead pid / missing wt)
- Tests cover lifecycle, foreign steal refusal, allocator compatibility, merger handoff
This commit is contained in:
2026-07-10 11:30:53 -04:00
parent a654cda058
commit 780ef0b1be
9 changed files with 2202 additions and 33 deletions
+598 -25
View File
@@ -11,9 +11,9 @@ Implements the durable coordination store described by
shared Postgres or single allocator daemon can replace the connection shared Postgres or single allocator daemon can replace the connection
layer later without changing call sites. layer later without changing call sites.
This module is the **substrate** for #600 (allocator policy/tool) and #612 This module is the **substrate** for #600 (allocator policy/tool), #601
(incident bridge). It does **not** implement ``gitea_allocate_next_work`` (first-class lease lifecycle), and #612 (incident bridge). It does **not**
routing policy or provider adapters. implement ``gitea_allocate_next_work`` routing policy or provider adapters.
""" """
from __future__ import annotations from __future__ import annotations
@@ -29,7 +29,7 @@ from dataclasses import dataclass
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from typing import Any, Iterator, Sequence from typing import Any, Iterator, Sequence
SCHEMA_VERSION = 2 SCHEMA_VERSION = 3
# Assignable work kinds only — raw monitoring incidents are never work items. # Assignable work kinds only — raw monitoring incidents are never work items.
WORK_KINDS = frozenset({"issue", "pr"}) WORK_KINDS = frozenset({"issue", "pr"})
@@ -301,6 +301,7 @@ class ControlPlaneDB:
try: try:
conn.executescript(_SCHEMA_SQL) conn.executescript(_SCHEMA_SQL)
self._migrate_incident_links_null_scope(conn) self._migrate_incident_links_null_scope(conn)
self._migrate_lease_lifecycle_columns(conn)
conn.execute( conn.execute(
"INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)", "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)",
("schema_version", str(SCHEMA_VERSION)), ("schema_version", str(SCHEMA_VERSION)),
@@ -604,6 +605,8 @@ class ControlPlaneDB:
lease_ttl_seconds: int = DEFAULT_LEASE_TTL_SECONDS, lease_ttl_seconds: int = DEFAULT_LEASE_TTL_SECONDS,
phase: str = "claimed", phase: str = "claimed",
now: datetime | None = None, now: datetime | None = None,
worktree_path: str | None = None,
owner_pid: int | None = None,
) -> AssignmentResult: ) -> AssignmentResult:
"""Atomically create assignment + lease for one Gitea work item. """Atomically create assignment + lease for one Gitea work item.
@@ -746,6 +749,32 @@ class ControlPlaneDB:
""", """,
(lease_id, work_item_id, session_id, role, phase, expires, now_s), (lease_id, work_item_id, session_id, role, phase, expires, now_s),
) )
# #601 optional lifecycle columns (present after schema v3 migration)
try:
lcols = {
r[1]
for r in conn.execute("PRAGMA table_info(leases)").fetchall()
}
if "worktree_path" in lcols and worktree_path:
conn.execute(
"UPDATE leases SET worktree_path = ? WHERE lease_id = ?",
(worktree_path, lease_id),
)
if "owner_pid" in lcols:
conn.execute(
"UPDATE leases SET owner_pid = ? WHERE lease_id = ?",
(
owner_pid if owner_pid is not None else os.getpid(),
lease_id,
),
)
if "expected_head_sha" in lcols and expected_head_sha:
conn.execute(
"UPDATE leases SET expected_head_sha = ? WHERE lease_id = ?",
(expected_head_sha, lease_id),
)
except sqlite3.Error:
pass
conn.execute( conn.execute(
""" """
INSERT INTO assignments( INSERT INTO assignments(
@@ -876,32 +905,15 @@ class ControlPlaneDB:
} }
def release_lease(self, lease_id: str, *, session_id: str) -> None: def release_lease(self, lease_id: str, *, session_id: str) -> None:
with self._tx() as conn: """Release a lease; unknown ids are no-ops for backward compatibility."""
with self._tx(immediate=False) as conn:
row = conn.execute( row = conn.execute(
"SELECT * FROM leases WHERE lease_id = ?", "SELECT lease_id FROM leases WHERE lease_id = ?",
(lease_id,), (lease_id,),
).fetchone() ).fetchone()
if not row: if not row:
return return
if row["session_id"] != session_id: self.release_lease_recorded(lease_id, session_id=session_id)
raise ForeignLeaseError(
f"cannot release lease {lease_id} owned by {row['session_id']}"
)
conn.execute(
"UPDATE leases SET status = 'released' WHERE lease_id = ?",
(lease_id,),
)
conn.execute(
"UPDATE assignments SET status = 'released' WHERE lease_id = ?",
(lease_id,),
)
conn.execute(
"""
INSERT INTO events(work_item_id, event_type, message, created_at)
VALUES (?, 'lease_released', ?, ?)
""",
(row["work_item_id"], f"session {session_id} released {lease_id}", _ts()),
)
def require_valid_assignment( def require_valid_assignment(
self, self,
@@ -1177,3 +1189,564 @@ class ControlPlaneDB:
(provider, base_url, p_org, p_project, str(provider_issue_id)), (provider, base_url, p_org, p_project, str(provider_issue_id)),
).fetchone() ).fetchone()
return dict(row) if row else None return dict(row) if row else None
# ── lease lifecycle (#601) ────────────────────────────────────────────
_LEASE_LIFECYCLE_COLUMNS: tuple[tuple[str, str], ...] = (
("worktree_path", "TEXT"),
("owner_pid", "INTEGER"),
("expected_head_sha", "TEXT"),
("adopted_from_session_id", "TEXT"),
("adopted_by_session_id", "TEXT"),
("provenance_json", "TEXT"),
("abandon_proof_json", "TEXT"),
)
def _migrate_lease_lifecycle_columns(self, conn: sqlite3.Connection) -> None:
"""Add provenance/worktree columns to leases for first-class lifecycle (#601)."""
cols = {
row[1]
for row in conn.execute("PRAGMA table_info(leases)").fetchall()
}
if not cols:
return
for name, decl in self._LEASE_LIFECYCLE_COLUMNS:
if name not in cols:
conn.execute(f"ALTER TABLE leases ADD COLUMN {name} {decl}")
def _lease_columns(self, conn: sqlite3.Connection) -> set[str]:
return {
row[1]
for row in conn.execute("PRAGMA table_info(leases)").fetchall()
}
def list_leases(
self,
*,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
role: str | None = None,
statuses: Sequence[str] | None = None,
limit: int = 100,
) -> list[dict[str, Any]]:
"""List leases joined with work_items as first-class workflow state."""
clauses: list[str] = []
params: list[Any] = []
if remote:
clauses.append("w.remote = ?")
params.append(remote)
if org:
clauses.append("w.org = ?")
params.append(org)
if repo:
clauses.append("w.repo = ?")
params.append(repo)
if role:
clauses.append("l.role = ?")
params.append(role)
if statuses:
placeholders = ", ".join("?" for _ in statuses)
clauses.append(f"l.status IN ({placeholders})")
params.extend(statuses)
where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
sql = f"""
SELECT l.*, w.remote, w.org, w.repo, w.kind AS work_kind,
w.number AS work_number, w.state AS work_state,
w.current_head_sha AS work_head_sha,
s.pid AS session_pid, s.profile AS session_profile,
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
{where}
ORDER BY l.expires_at DESC
LIMIT ?
"""
params.append(max(1, int(limit)))
with self._tx(immediate=False) as conn:
rows = conn.execute(sql, params).fetchall()
return [dict(r) for r in rows]
def get_lease_workflow_state(self, lease_id: str) -> dict[str, Any] | None:
"""Return lease + assignment + work_item + session for one lease id."""
with self._tx(immediate=False) as conn:
lease = conn.execute(
"SELECT * FROM leases WHERE lease_id = ?",
(lease_id,),
).fetchone()
if not lease:
return None
work = conn.execute(
"SELECT * FROM work_items WHERE work_item_id = ?",
(lease["work_item_id"],),
).fetchone()
asn = conn.execute(
"""
SELECT * FROM assignments
WHERE lease_id = ?
ORDER BY created_at DESC LIMIT 1
""",
(lease_id,),
).fetchone()
sess = conn.execute(
"SELECT * FROM sessions WHERE session_id = ?",
(lease["session_id"],),
).fetchone()
provenance = None
if "provenance_json" in lease.keys() and lease["provenance_json"]:
try:
provenance = json.loads(lease["provenance_json"])
except (TypeError, json.JSONDecodeError):
provenance = {"raw": lease["provenance_json"]}
return {
"lease": dict(lease),
"work_item": dict(work) if work else None,
"assignment": dict(asn) if asn else None,
"session": dict(sess) if sess else None,
"provenance": provenance,
}
def attach_lease_provenance(
self, lease_id: str, provenance: dict[str, Any]
) -> None:
payload = json.dumps(provenance)
with self._tx() as conn:
cols = self._lease_columns(conn)
if "provenance_json" not in cols:
return
conn.execute(
"UPDATE leases SET provenance_json = ? WHERE lease_id = ?",
(payload, lease_id),
)
if provenance.get("adopted_from_session_id") and "adopted_from_session_id" in cols:
conn.execute(
"""
UPDATE leases
SET adopted_from_session_id = ?, adopted_by_session_id = ?
WHERE lease_id = ?
""",
(
provenance.get("adopted_from_session_id"),
provenance.get("adopted_by_session_id"),
lease_id,
),
)
if provenance.get("worktree_path") and "worktree_path" in cols:
conn.execute(
"UPDATE leases SET worktree_path = ? WHERE lease_id = ?",
(provenance.get("worktree_path"), lease_id),
)
if provenance.get("expected_head_sha") and "expected_head_sha" in cols:
conn.execute(
"UPDATE leases SET expected_head_sha = ? WHERE lease_id = ?",
(provenance.get("expected_head_sha"), lease_id),
)
def release_lease_recorded(
self, lease_id: str, *, session_id: str
) -> dict[str, Any]:
"""Explicit release with audit event + provenance proof (#601)."""
now_s = _ts()
with self._tx() as conn:
row = conn.execute(
"SELECT * FROM leases WHERE lease_id = ?",
(lease_id,),
).fetchone()
if not row:
raise ControlPlaneError(f"unknown lease_id {lease_id}")
if row["session_id"] != session_id:
raise ForeignLeaseError(
f"cannot release lease {lease_id} owned by {row['session_id']}"
)
if row["status"] not in ("active", "expired"):
# idempotent-ish for already released
if row["status"] == "released":
return {
"lease_id": lease_id,
"status": "released",
"session_id": session_id,
"released_at": now_s,
"idempotent": True,
}
conn.execute(
"UPDATE leases SET status = 'released' WHERE lease_id = ?",
(lease_id,),
)
conn.execute(
"UPDATE assignments SET status = 'released' WHERE lease_id = ?",
(lease_id,),
)
proof = {
"lease_id": lease_id,
"status": "released",
"session_id": session_id,
"released_at": now_s,
"prior_status": row["status"],
"work_item_id": row["work_item_id"],
}
cols = self._lease_columns(conn)
if "provenance_json" in cols:
prior = {}
if row["provenance_json"]:
try:
prior = json.loads(row["provenance_json"])
except (TypeError, json.JSONDecodeError):
prior = {}
prior["last_release"] = proof
conn.execute(
"UPDATE leases SET provenance_json = ? WHERE lease_id = ?",
(json.dumps(prior), lease_id),
)
conn.execute(
"""
INSERT INTO events(work_item_id, event_type, message, created_at)
VALUES (?, 'lease_released', ?, ?)
""",
(
row["work_item_id"],
f"session {session_id} released {lease_id}",
now_s,
),
)
return proof
def force_expire_lease(self, lease_id: str, *, reason: str = "") -> None:
now_s = _ts()
with self._tx() as conn:
row = conn.execute(
"SELECT * FROM leases WHERE lease_id = ?",
(lease_id,),
).fetchone()
if not row:
raise ControlPlaneError(f"unknown lease_id {lease_id}")
conn.execute(
"UPDATE leases SET status = 'expired' WHERE lease_id = ?",
(lease_id,),
)
conn.execute(
"UPDATE assignments SET status = 'expired' WHERE lease_id = ?",
(lease_id,),
)
conn.execute(
"""
INSERT INTO events(work_item_id, event_type, message, created_at)
VALUES (?, 'lease_expired', ?, ?)
""",
(
row["work_item_id"],
f"lease {lease_id} force-expired: {reason or 'unspecified'}",
now_s,
),
)
def abandon_lease(
self,
*,
lease_id: str,
requester_session_id: str,
proof: dict[str, Any],
) -> dict[str, Any]:
now_s = _ts()
with self._tx() as conn:
row = conn.execute(
"SELECT * FROM leases WHERE lease_id = ?",
(lease_id,),
).fetchone()
if not row:
raise ControlPlaneError(f"unknown lease_id {lease_id}")
conn.execute(
"UPDATE leases SET status = 'abandoned' WHERE lease_id = ?",
(lease_id,),
)
conn.execute(
"UPDATE assignments SET status = 'abandoned' WHERE lease_id = ?",
(lease_id,),
)
cols = self._lease_columns(conn)
if "abandon_proof_json" in cols:
conn.execute(
"UPDATE leases SET abandon_proof_json = ? WHERE lease_id = ?",
(json.dumps(proof), lease_id),
)
msg = (
f"session {requester_session_id} abandoned lease {lease_id} "
f"(prior owner {row['session_id']})"
)
conn.execute(
"""
INSERT INTO events(work_item_id, event_type, message, created_at)
VALUES (?, 'lease_abandoned', ?, ?)
""",
(row["work_item_id"], msg, now_s),
)
return {
"lease_id": lease_id,
"status": "abandoned",
"prior_owner_session_id": row["session_id"],
"requester_session_id": requester_session_id,
"abandoned_at": now_s,
"proof": proof,
}
def adopt_lease(
self,
*,
lease_id: str,
adopter_session_id: str,
role: str,
worktree_path: str | None = None,
expected_head_sha: str | None = None,
owner_pid: int | None = None,
provenance: dict[str, Any] | None = None,
lease_ttl_seconds: int = DEFAULT_LEASE_TTL_SECONDS,
) -> dict[str, Any]:
"""Transfer or refresh a lease with provenance (#601).
* Same owner + active → refresh (owner-resume).
* Expired/abandoned/released → create new assignment+lease with provenance.
* Active foreign → raise ForeignLeaseError (never silent steal).
"""
now = _utc_now()
now_s = _ts(now)
expires = _ts(now + timedelta(seconds=int(lease_ttl_seconds)))
provenance = provenance or {}
with self._tx(immediate=True) as conn:
lease = conn.execute(
"SELECT * FROM leases WHERE lease_id = ?",
(lease_id,),
).fetchone()
if not lease:
raise ControlPlaneError(f"unknown lease_id {lease_id}")
work = conn.execute(
"SELECT * FROM work_items WHERE work_item_id = ?",
(lease["work_item_id"],),
).fetchone()
if not work:
raise ControlPlaneError("lease has no work_item")
# Expire by time if needed
exp = _parse_ts(lease["expires_at"])
status = lease["status"]
if status == "active" and exp and exp <= now:
conn.execute(
"UPDATE leases SET status = 'expired' WHERE lease_id = ?",
(lease_id,),
)
conn.execute(
"UPDATE assignments SET status = 'expired' WHERE lease_id = ?",
(lease_id,),
)
status = "expired"
owner = lease["session_id"]
if status == "active" and owner != adopter_session_id:
raise ForeignLeaseError(
f"cannot adopt active foreign lease {lease_id} owned by {owner}"
)
# Ensure adopter session exists
sess = conn.execute(
"SELECT session_id FROM sessions WHERE session_id = ?",
(adopter_session_id,),
).fetchone()
if not sess:
conn.execute(
"""
INSERT INTO sessions(
session_id, role, profile, namespace, pid,
started_at, last_heartbeat_at, status
) VALUES (?, ?, NULL, NULL, ?, ?, ?, 'active')
""",
(adopter_session_id, role, owner_pid or os.getpid(), now_s, now_s),
)
cols = self._lease_columns(conn)
if status == "active" and owner == adopter_session_id:
# Owner-resume refresh
conn.execute(
"""
UPDATE leases
SET heartbeat_at = ?, expires_at = ?, phase = ?
WHERE lease_id = ?
""",
(now_s, expires, "adopted", lease_id),
)
if "worktree_path" in cols and worktree_path:
conn.execute(
"UPDATE leases SET worktree_path = ? WHERE lease_id = ?",
(worktree_path, lease_id),
)
if "owner_pid" in cols and owner_pid is not None:
conn.execute(
"UPDATE leases SET owner_pid = ? WHERE lease_id = ?",
(owner_pid, lease_id),
)
if "provenance_json" in cols:
prior = {}
if lease["provenance_json"]:
try:
prior = json.loads(lease["provenance_json"])
except (TypeError, json.JSONDecodeError):
prior = {}
prior["last_adopt"] = provenance
conn.execute(
"UPDATE leases SET provenance_json = ? WHERE lease_id = ?",
(json.dumps(prior), lease_id),
)
asn = conn.execute(
"""
SELECT * FROM assignments
WHERE lease_id = ? AND status = 'active'
ORDER BY created_at DESC LIMIT 1
""",
(lease_id,),
).fetchone()
lease2 = conn.execute(
"SELECT * FROM leases WHERE lease_id = ?", (lease_id,)
).fetchone()
conn.execute(
"""
INSERT INTO events(work_item_id, event_type, message, created_at)
VALUES (?, 'lease_adopted', ?, ?)
""",
(
lease["work_item_id"],
f"owner-resume adopt lease {lease_id} by {adopter_session_id}",
now_s,
),
)
return {
"outcome": "adopted_owner_resume",
"lease": dict(lease2) if lease2 else dict(lease),
"assignment": dict(asn) if asn else None,
"reasons": ["owner-resume: refreshed lease with provenance"],
}
# Non-active: create new lease + assignment (transfer)
new_lease_id = f"lease-{uuid.uuid4().hex[:16]}"
new_asn_id = f"asn-{uuid.uuid4().hex[:16]}"
# Mark prior non-active if still active somehow
if status == "active":
conn.execute(
"UPDATE leases SET status = 'released' WHERE lease_id = ?",
(lease_id,),
)
conn.execute(
"UPDATE assignments SET status = 'released' WHERE lease_id = ?",
(lease_id,),
)
# Prefer prior assignment allowed/forbidden
prior_asn = conn.execute(
"""
SELECT * FROM assignments WHERE lease_id = ?
ORDER BY created_at DESC LIMIT 1
""",
(lease_id,),
).fetchone()
allowed = (
prior_asn["allowed_actions"]
if prior_asn
else json.dumps(["implement", "comment", "push", "create_pr"])
)
forbidden = (
prior_asn["forbidden_actions"]
if prior_asn
else json.dumps(
["approve", "merge", "request_changes", "self_select_without_assignment"]
)
)
head = expected_head_sha or (
prior_asn["expected_head_sha"] if prior_asn else work["current_head_sha"]
)
# Dynamic insert for optional columns
base_cols = [
"lease_id",
"work_item_id",
"session_id",
"role",
"phase",
"expires_at",
"heartbeat_at",
"status",
]
base_vals: list[Any] = [
new_lease_id,
lease["work_item_id"],
adopter_session_id,
role,
"adopted",
expires,
now_s,
"active",
]
optional = {
"worktree_path": worktree_path,
"owner_pid": owner_pid,
"expected_head_sha": head,
"adopted_from_session_id": owner,
"adopted_by_session_id": adopter_session_id,
"provenance_json": json.dumps(provenance),
}
for col, val in optional.items():
if col in cols and val is not None:
base_cols.append(col)
base_vals.append(val)
placeholders = ", ".join("?" for _ in base_cols)
conn.execute(
f"INSERT INTO leases({', '.join(base_cols)}) VALUES ({placeholders})",
base_vals,
)
conn.execute(
"""
INSERT INTO assignments(
assignment_id, work_item_id, session_id, lease_id,
allowed_actions, forbidden_actions, expected_head_sha,
role, status, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'active', ?)
""",
(
new_asn_id,
lease["work_item_id"],
adopter_session_id,
new_lease_id,
allowed if isinstance(allowed, str) else json.dumps(list(allowed)),
forbidden if isinstance(forbidden, str) else json.dumps(list(forbidden)),
head,
role,
now_s,
),
)
conn.execute(
"""
INSERT INTO events(work_item_id, event_type, message, created_at)
VALUES (?, 'lease_adopted', ?, ?)
""",
(
lease["work_item_id"],
f"session {adopter_session_id} adopted from {owner} "
f"prior={lease_id} new={new_lease_id}",
now_s,
),
)
lease2 = conn.execute(
"SELECT * FROM leases WHERE lease_id = ?", (new_lease_id,)
).fetchone()
asn2 = conn.execute(
"SELECT * FROM assignments WHERE assignment_id = ?",
(new_asn_id,),
).fetchone()
return {
"outcome": "adopted_transfer",
"lease": dict(lease2) if lease2 else None,
"assignment": dict(asn2) if asn2 else None,
"reasons": [
f"transferred lease ownership from {owner} to {adopter_session_id}"
],
}
@@ -51,6 +51,35 @@ Module: `allocator_service.py` · MCP tool: `gitea_allocate_next_work`
- Routing follows ADR §5.3 (REQUEST_CHANGES → author, terminal path first, foreign lease → wait). - Routing follows ADR §5.3 (REQUEST_CHANGES → author, terminal path first, foreign lease → wait).
- Raw monitoring incidents are never candidates. - Raw monitoring incidents are never candidates.
## Lease lifecycle API (#601)
Module: `lease_lifecycle.py` · MCP tools: `gitea_*_workflow_lease(s)`
Active control-plane leases are **first-class workflow state**:
| Tool | Purpose |
|------|---------|
| `gitea_list_workflow_leases` | List active (or all) leases for a repo |
| `gitea_inspect_workflow_lease` | Freshness + `safe_next_action` for one lease id |
| `gitea_adopt_workflow_lease` | Owner-resume or sanctioned reclaim with provenance |
| `gitea_release_workflow_lease` | Explicit owner release (audited) |
| `gitea_expire_workflow_leases` | Deterministic expire of past-`expires_at` leases |
| `gitea_abandon_workflow_lease` | Abandon with required proof |
| `gitea_reclaim_expired_workflow_lease` | Expire + re-assign with provenance |
### Hard rules
1. Control-plane DB is the coordination authority — file locks / comment-only leases are not authoritative alone.
2. Active foreign leases cannot be stolen; inspect returns `wait_foreign_active`.
3. Abandon requires `(dead_process OR missing_worktree)` and `no_live_mutation_risk`; foreign abandon also needs `operator_authorized` or full dead+missing+no_open_pr proof.
4. Adopt provenance always records `adopted_from_session_id`, `adopted_by_session_id`, work identity, optional head SHA, and worktree path.
5. Reviewer→merger comment handoff (`gitea_adopt_merger_pr_lease`) is unchanged and remains the Gitea-thread durability path.
```bash
python3 -m pytest tests/test_lease_lifecycle.py tests/test_control_plane_db.py tests/test_allocator_service.py tests/test_merger_lease_adoption.py -q
```
## Incident bridge (#612) ## Incident bridge (#612)
Module: `incident_bridge.py` · MCP tools: `gitea_observability_*` Module: `incident_bridge.py` · MCP tools: `gitea_observability_*`
+310
View File
@@ -837,6 +837,7 @@ import mcp_session_state # noqa: E402
import stale_review_decision_lock # noqa: E402 import stale_review_decision_lock # noqa: E402
import allocator_service # noqa: E402 import allocator_service # noqa: E402
import control_plane_db # noqa: E402 import control_plane_db # noqa: E402
import lease_lifecycle # noqa: E402
import incident_bridge # noqa: E402 import incident_bridge # noqa: E402
import agent_temp_artifacts import agent_temp_artifacts
import issue_lock_worktree # noqa: E402 import issue_lock_worktree # noqa: E402
@@ -11622,6 +11623,315 @@ def gitea_allocate_next_work(
return result return result
# ── Control-plane lease lifecycle (#601) ──────────────────────────────────────
@mcp.tool()
def gitea_list_workflow_leases(
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
role: str | None = None,
include_non_active: bool = False,
limit: int = 100,
) -> dict:
"""List control-plane leases as first-class workflow state (#601).
Read-only. Returns active (default) or all leases from the #613 DB substrate.
File locks and comment-only leases are not listed as authoritative here.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"reasons": read_block,
"leases": [],
"permission_report": _permission_block_report("gitea.read"),
}
try:
h, o, r = _resolve(remote, host, org, repo)
except ValueError as exc:
return {"success": False, "reasons": [str(exc)], "leases": []}
db, errs = _control_plane_db_or_error()
if db is None:
return {"success": False, "reasons": errs, "leases": []}
result = lease_lifecycle.list_active_leases(
db,
remote=remote if remote in REMOTES else remote,
org=o,
repo=r,
role=role,
include_non_active=bool(include_non_active),
limit=int(limit),
)
result["remote"] = remote
result["org"] = o
result["repo"] = r
return result
@mcp.tool()
def gitea_inspect_workflow_lease(
lease_id: str,
session_id: str | None = None,
worktree_path: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
) -> dict:
"""Inspect one control-plane lease with safe_next_action (#601).
Distinguishes owner-resume, wait-foreign, reclaim-expired, abandon-allowed,
and stale prompt ids. Control-plane DB is authoritative.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
db, errs = _control_plane_db_or_error()
if db is None:
return {"success": False, "reasons": errs}
profile = get_profile()
profile_name = (profile.get("profile_name") or "").strip() or "session"
sid = (session_id or "").strip() or f"{profile_name}-{os.getpid()}"
return lease_lifecycle.inspect_lease(
db,
lease_id,
caller_session_id=sid,
caller_worktree=worktree_path,
)
@mcp.tool()
def gitea_adopt_workflow_lease(
lease_id: str,
session_id: str | None = None,
role: str | None = None,
worktree_path: str | None = None,
expected_head_sha: str | None = None,
operator_authorized: bool = False,
remote: str = "dadeschools",
host: str | None = None,
) -> dict:
"""Adopt a control-plane lease through the sanctioned path (#601).
Same-owner resume refreshes provenance. Foreign active leases are refused.
Expired leases may be reclaimed; provenance records adopted_from/by.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
db, errs = _control_plane_db_or_error()
if db is None:
return {"success": False, "reasons": errs}
profile = get_profile()
profile_name = (profile.get("profile_name") or "").strip() or "session"
active_role = _profile_role_kind(profile) or "author"
sid = (session_id or "").strip() or (
f"{profile_name}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
)
try:
return lease_lifecycle.adopt_lease(
db,
lease_id=lease_id,
adopter_session_id=sid,
role=(role or active_role).strip() or "author",
worktree_path=worktree_path,
expected_head_sha=expected_head_sha,
owner_pid=os.getpid(),
operator_authorized=bool(operator_authorized),
)
except (lease_lifecycle.LeaseLifecycleError, control_plane_db.ControlPlaneError) as exc:
return {
"success": False,
"outcome": "blocked",
"reasons": [_redact(str(exc))],
"lease_id": lease_id,
"authoritative_source": "control_plane_db",
"file_lock_only": False,
"comment_lease_only": False,
}
@mcp.tool()
def gitea_release_workflow_lease(
lease_id: str,
session_id: str,
remote: str = "dadeschools",
host: str | None = None,
) -> dict:
"""Explicitly release a control-plane lease owned by *session_id* (#601).
Recorded in the DB event log with release provenance. Foreign release fails closed.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
db, errs = _control_plane_db_or_error()
if db is None:
return {"success": False, "reasons": errs}
try:
return lease_lifecycle.release_lease(
db, lease_id=lease_id, session_id=session_id
)
except (lease_lifecycle.LeaseLifecycleError, control_plane_db.ControlPlaneError) as exc:
return {
"success": False,
"outcome": "blocked",
"reasons": [_redact(str(exc))],
"lease_id": lease_id,
"authoritative_source": "control_plane_db",
}
@mcp.tool()
def gitea_expire_workflow_leases(
remote: str = "dadeschools",
host: str | None = None,
) -> dict:
"""Expire stale control-plane leases whose expires_at has passed (#601).
Deterministic reclaim prerequisite. Does not steal active non-expired leases.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
db, errs = _control_plane_db_or_error()
if db is None:
return {"success": False, "reasons": errs}
return lease_lifecycle.expire_leases(db)
@mcp.tool()
def gitea_abandon_workflow_lease(
lease_id: str,
session_id: str | None = None,
dead_process: bool = False,
missing_worktree: bool = False,
no_open_pr: bool = False,
no_live_mutation_risk: bool = False,
operator_authorized: bool = False,
worktree_path: str | None = None,
owner_pid: int | None = None,
notes: str = "",
remote: str = "dadeschools",
host: str | None = None,
) -> dict:
"""Abandon a control-plane lease with required proof (#601).
Requires (dead_process or missing_worktree) and no_live_mutation_risk.
Foreign abandon also needs operator_authorized or
(dead_process and missing_worktree and no_open_pr).
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
db, errs = _control_plane_db_or_error()
if db is None:
return {"success": False, "reasons": errs}
profile = get_profile()
profile_name = (profile.get("profile_name") or "").strip() or "session"
sid = (session_id or "").strip() or f"{profile_name}-{os.getpid()}"
proof = lease_lifecycle.AbandonProof(
dead_process=bool(dead_process),
missing_worktree=bool(missing_worktree),
no_open_pr=bool(no_open_pr),
no_live_mutation_risk=bool(no_live_mutation_risk),
operator_authorized=bool(operator_authorized),
worktree_path=worktree_path,
owner_pid=owner_pid,
notes=notes or "",
)
try:
return lease_lifecycle.abandon_lease(
db,
lease_id=lease_id,
requester_session_id=sid,
proof=proof,
)
except (lease_lifecycle.LeaseLifecycleError, control_plane_db.ControlPlaneError) as exc:
return {
"success": False,
"outcome": "blocked",
"reasons": [_redact(str(exc))],
"lease_id": lease_id,
"authoritative_source": "control_plane_db",
"file_lock_only": False,
"comment_lease_only": False,
}
@mcp.tool()
def gitea_reclaim_expired_workflow_lease(
lease_id: str,
session_id: str | None = None,
role: str | None = None,
worktree_path: str | None = None,
expected_head_sha: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
) -> dict:
"""Reclaim an expired control-plane lease for a new/owner session (#601).
Deterministic: expire markers applied, then atomic assign+lease with provenance.
Active foreign leases are refused.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
db, errs = _control_plane_db_or_error()
if db is None:
return {"success": False, "reasons": errs}
profile = get_profile()
profile_name = (profile.get("profile_name") or "").strip() or "session"
active_role = _profile_role_kind(profile) or "author"
sid = (session_id or "").strip() or (
f"{profile_name}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
)
try:
return lease_lifecycle.reclaim_expired_lease(
db,
lease_id=lease_id,
session_id=sid,
role=(role or active_role).strip() or "author",
worktree_path=worktree_path,
expected_head_sha=expected_head_sha,
)
except (lease_lifecycle.LeaseLifecycleError, control_plane_db.ControlPlaneError) as exc:
return {
"success": False,
"outcome": "blocked",
"reasons": [_redact(str(exc))],
"lease_id": lease_id,
"authoritative_source": "control_plane_db",
}
# ── Entry point ─────────────────────────────────────────────────────────────── # ── Entry point ───────────────────────────────────────────────────────────────
if __name__ == "__main__": if __name__ == "__main__":
+63
View File
@@ -375,6 +375,64 @@ def _same_realpath(left: str | None, right: str | None) -> bool:
return left == right return left == right
def assess_expired_lock_reclaim(
existing_lock: dict[str, Any] | None,
*,
now: datetime | None = None,
) -> dict[str, Any]:
"""Decide whether an expired/stale author issue lock may be reclaimed (#601).
Required proof (any reclaim of non-live lock):
* lease not live (expired or dead pid)
* owner process dead OR worktree missing
* no force-delete of live foreign ownership
"""
if not existing_lock:
return {
"reclaim_allowed": True,
"reasons": ["no existing lock"],
"freshness": assess_lock_freshness(None, now=now),
}
freshness = assess_lock_freshness(existing_lock, now=now)
if freshness.get("live"):
return {
"reclaim_allowed": False,
"reasons": ["lock is still live; cannot reclaim (fail closed)"],
"freshness": freshness,
}
pid = existing_lock.get("session_pid")
if pid is None:
pid = existing_lock.get("pid")
dead = not is_process_alive(pid) if pid is not None else True
wt = str(existing_lock.get("worktree_path") or "")
missing_wt = (not wt) or (not os.path.isdir(os.path.realpath(wt)))
if not (dead or missing_wt):
return {
"reclaim_allowed": False,
"reasons": [
"expired/stale lock still has live owner pid and present worktree; "
"recovery review required (fail closed)"
],
"freshness": freshness,
"owner_pid_dead": dead,
"worktree_missing": missing_wt,
}
return {
"reclaim_allowed": True,
"reasons": [
"non-live lock with dead process and/or missing worktree; "
"sanctioned reclaim allowed"
],
"freshness": freshness,
"owner_pid_dead": dead,
"worktree_missing": missing_wt,
"prior_branch": existing_lock.get("branch_name"),
"prior_worktree": existing_lock.get("worktree_path"),
"prior_pid": pid,
}
def assess_same_issue_lease_conflict( def assess_same_issue_lease_conflict(
existing_lock: dict[str, Any] | None, existing_lock: dict[str, Any] | None,
*, *,
@@ -405,6 +463,11 @@ def assess_same_issue_lease_conflict(
and _same_realpath(str(existing_worktree or ""), worktree_path) and _same_realpath(str(existing_worktree or ""), worktree_path)
) )
if is_lease_expired(existing_lock, now=now): if is_lease_expired(existing_lock, now=now):
reclaim = assess_expired_lock_reclaim(existing_lock, now=now)
if reclaim.get("reclaim_allowed"):
# #601: expired + dead pid / missing worktree may be reclaimed
# through the normal lock path (sanctioned overwrite).
return None
return ( return (
f"Issue #{issue_number} has an expired {operation_type} lease on " f"Issue #{issue_number} has an expired {operation_type} lease on "
f"branch '{existing_branch}' from worktree '{existing_worktree}'. " f"branch '{existing_branch}' from worktree '{existing_worktree}'. "
+723
View File
@@ -0,0 +1,723 @@
"""First-class control-plane lease lifecycle (#601).
Active leases are queryable workflow state. Canonical operations:
* list / inspect
* adopt (with provenance)
* release (explicit, recorded)
* expire / reclaim
* abandon (requires proof: dead process and/or missing worktree, etc.)
Authority model:
* Control-plane DB is the coordination source for assignment/lease.
* File locks and comment-only leases are **not** authoritative alone.
* Reviewer/merger comment leases (``reviewer_pr_lease`` / merger adoption)
remain for Gitea-thread durability; they must not bypass DB assignment when
the control-plane path is in use.
Fail closed on ambiguous ownership, active foreign leases, mismatched
worktree, stale head, missing capability, and unsafe cleanup.
"""
from __future__ import annotations
import json
import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable, Mapping
import control_plane_db as cpd
# Outcomes / safe next actions (stable vocabulary for tools + tests).
SAFE_OWNER_RESUME = "owner_resume"
SAFE_WAIT_FOREIGN = "wait_foreign_active"
SAFE_RECLAIM_EXPIRED = "reclaim_expired"
SAFE_ABANDON_ALLOWED = "abandon_allowed"
SAFE_RELEASE_OWNED = "release_owned"
SAFE_STALE_PROMPT = "stale_prompt_lease"
SAFE_UNKNOWN = "inspect_only"
SAFE_NO_AUTHORITY = "file_or_comment_not_authoritative"
LEASE_STATUS_ACTIVE = "active"
LEASE_STATUS_RELEASED = "released"
LEASE_STATUS_EXPIRED = "expired"
LEASE_STATUS_ABANDONED = "abandoned"
AUTHORITATIVE_SOURCE = "control_plane_db"
class LeaseLifecycleError(RuntimeError):
"""Fail-closed lifecycle policy error."""
@dataclass(frozen=True)
class AbandonProof:
"""Required evidence to abandon a non-owned or expired lease safely."""
dead_process: bool = False
missing_worktree: bool = False
same_owner: bool = False
no_open_pr: bool = False
no_live_mutation_risk: bool = False
operator_authorized: bool = False
worktree_path: str | None = None
owner_pid: int | None = None
notes: str = ""
def as_dict(self) -> dict[str, Any]:
return {
"dead_process": self.dead_process,
"missing_worktree": self.missing_worktree,
"same_owner": self.same_owner,
"no_open_pr": self.no_open_pr,
"no_live_mutation_risk": self.no_live_mutation_risk,
"operator_authorized": self.operator_authorized,
"worktree_path": self.worktree_path,
"owner_pid": self.owner_pid,
"notes": self.notes,
}
def is_sufficient(self) -> bool:
"""Abandon requires process death or missing worktree + no mutation risk.
Foreign active leases additionally need operator_authorized unless the
owner process is dead and the worktree is missing.
"""
if not self.no_live_mutation_risk:
return False
if not (self.dead_process or self.missing_worktree):
return False
if self.same_owner:
return True
# Foreign abandon: operator flag OR (dead + missing worktree + no risk)
if self.operator_authorized:
return True
return bool(self.dead_process and self.missing_worktree and self.no_open_pr)
def is_process_alive(pid: int | None) -> bool:
if pid is None:
return False
try:
pid_i = int(pid)
except (TypeError, ValueError):
return False
if pid_i <= 0:
return False
try:
os.kill(pid_i, 0)
return True
except ProcessLookupError:
return False
except PermissionError:
# Exists but not owned by us — treat as alive (fail closed).
return True
except OSError:
return False
def worktree_exists(path: str | None) -> bool:
if not path:
return False
try:
return os.path.isdir(os.path.realpath(path))
except OSError:
return False
def _utc_now() -> datetime:
return datetime.now(timezone.utc)
def _parse_ts(value: str | None) -> datetime | None:
return cpd._parse_ts(value)
def classify_lease_freshness(
lease: Mapping[str, Any],
*,
now: datetime | None = None,
pid_checker: Callable[[int | None], bool] = is_process_alive,
worktree_checker: Callable[[str | None], bool] = worktree_exists,
) -> dict[str, Any]:
"""Classify a control-plane lease row for workflow tooling."""
moment = now or _utc_now()
status = (lease.get("status") or "").strip().lower()
expires = _parse_ts(lease.get("expires_at"))
owner_pid = lease.get("owner_pid")
if owner_pid is None:
owner_pid = lease.get("session_pid")
wt = lease.get("worktree_path")
pid_alive = pid_checker(owner_pid) if owner_pid is not None else None
wt_present = worktree_checker(wt) if wt else None
expired_by_time = bool(expires and expires <= moment)
if status == LEASE_STATUS_ABANDONED:
freshness = "abandoned"
elif status == LEASE_STATUS_RELEASED:
freshness = "released"
elif status == LEASE_STATUS_EXPIRED or expired_by_time:
freshness = "expired"
elif status != LEASE_STATUS_ACTIVE:
freshness = status or "unknown"
elif pid_alive is False:
freshness = "stale_dead_process"
elif wt is not None and wt_present is False:
freshness = "stale_missing_worktree"
else:
freshness = "active"
return {
"freshness": freshness,
"status": status,
"expired_by_time": expired_by_time,
"owner_pid": owner_pid,
"owner_pid_alive": pid_alive,
"worktree_path": wt,
"worktree_present": wt_present,
"expires_at": lease.get("expires_at"),
"authoritative_source": AUTHORITATIVE_SOURCE,
"file_lock_only": False,
"comment_lease_only": False,
}
def decide_safe_next_action(
*,
lease: Mapping[str, Any] | None,
caller_session_id: str,
freshness: Mapping[str, Any] | None = None,
caller_worktree: str | None = None,
) -> dict[str, Any]:
"""Return safe_next_action + reasons for a caller inspecting a lease."""
if not lease:
return {
"safe_next_action": SAFE_STALE_PROMPT,
"reasons": ["lease not found in control-plane DB (stale prompt id?)"],
"block": True,
}
fr = freshness or classify_lease_freshness(lease)
owner = str(lease.get("session_id") or "")
same_owner = owner == str(caller_session_id)
status = fr.get("freshness")
reasons: list[str] = []
# Worktree mismatch for owner resume
lease_wt = lease.get("worktree_path")
if (
same_owner
and lease_wt
and caller_worktree
and os.path.realpath(str(lease_wt)) != os.path.realpath(str(caller_worktree))
):
return {
"safe_next_action": SAFE_UNKNOWN,
"reasons": [
f"worktree mismatch: lease has {lease_wt!r}, caller has "
f"{caller_worktree!r} (fail closed)"
],
"block": True,
"same_owner": True,
}
if status in ("abandoned", "released"):
return {
"safe_next_action": SAFE_STALE_PROMPT,
"reasons": [f"lease status is {status}; do not adopt blindly"],
"block": True,
"same_owner": same_owner,
}
if status == "expired" or fr.get("expired_by_time"):
return {
"safe_next_action": SAFE_RECLAIM_EXPIRED,
"reasons": ["lease expired; reclaim via expire+assign or adopt reclaim path"],
"block": False,
"same_owner": same_owner,
}
if status in ("stale_dead_process", "stale_missing_worktree"):
if same_owner:
return {
"safe_next_action": SAFE_OWNER_RESUME,
"reasons": [
f"caller owns lease with freshness={status}; rebind via "
"adopt/owner-resume or abandon with proof"
],
"block": False,
"same_owner": True,
"also_allowed": [SAFE_ABANDON_ALLOWED, SAFE_RELEASE_OWNED],
}
return {
"safe_next_action": SAFE_ABANDON_ALLOWED,
"reasons": [
f"lease freshness={status}; abandon with required proof then reassign"
],
"block": False,
"same_owner": False,
}
if same_owner and status == "active":
return {
"safe_next_action": SAFE_OWNER_RESUME,
"reasons": [
"caller owns active lease; resume via heartbeat/assign owner-resume "
"or release explicitly"
],
"block": False,
"same_owner": True,
"also_allowed": [SAFE_RELEASE_OWNED],
}
if not same_owner and status == "active":
return {
"safe_next_action": SAFE_WAIT_FOREIGN,
"reasons": [
f"foreign active lease held by session {owner}; "
"do not steal (fail closed)"
],
"block": True,
"same_owner": False,
"owner_session_id": owner,
}
return {
"safe_next_action": SAFE_UNKNOWN,
"reasons": [f"unclassified freshness={status}"],
"block": True,
"same_owner": same_owner,
}
def non_db_lease_authority_report(
*,
file_lock_present: bool = False,
comment_lease_present: bool = False,
db_lease_present: bool = False,
) -> dict[str, Any]:
"""File/comment leases alone are not coordination authority (#601 / #600)."""
authoritative = bool(db_lease_present)
reasons: list[str] = []
if file_lock_present and not db_lease_present:
reasons.append(
"file lock present but control-plane DB lease absent — "
"file lock is not authoritative alone"
)
if comment_lease_present and not db_lease_present:
reasons.append(
"comment-only lease present but control-plane DB lease absent — "
"comment lease is not authoritative alone"
)
if authoritative:
reasons.append("control-plane DB lease is the coordination authority")
return {
"authoritative_source": AUTHORITATIVE_SOURCE if authoritative else None,
"db_lease_present": db_lease_present,
"file_lock_present": file_lock_present,
"comment_lease_present": comment_lease_present,
"safe_next_action": (
SAFE_UNKNOWN if authoritative else SAFE_NO_AUTHORITY
),
"reasons": reasons,
"file_lock_only": bool(file_lock_present and not db_lease_present),
"comment_lease_only": bool(comment_lease_present and not db_lease_present),
}
def build_adopt_provenance(
*,
adopted_from_session_id: str,
adopted_by_session_id: str,
work_kind: str,
work_number: int,
remote: str,
org: str,
repo: str,
worktree_path: str | None,
expected_head_sha: str | None,
prior_lease_id: str | None,
reason: str,
) -> dict[str, Any]:
return {
"adopted_from_session_id": adopted_from_session_id,
"adopted_by_session_id": adopted_by_session_id,
"work_kind": work_kind,
"work_number": int(work_number),
"remote": remote,
"org": org,
"repo": repo,
"worktree_path": worktree_path,
"expected_head_sha": expected_head_sha,
"prior_lease_id": prior_lease_id,
"reason": reason,
"recorded_at": cpd._ts(),
"source": "lease_lifecycle.adopt",
}
def inspect_lease(
db: cpd.ControlPlaneDB,
lease_id: str,
*,
caller_session_id: str,
caller_worktree: str | None = None,
) -> dict[str, Any]:
"""Full first-class lease workflow state for one lease id."""
state = db.get_lease_workflow_state(lease_id)
if not state:
decision = decide_safe_next_action(
lease=None, caller_session_id=caller_session_id
)
return {
"success": True,
"found": False,
"lease_id": lease_id,
"lease": None,
"freshness": None,
**decision,
"authoritative_source": AUTHORITATIVE_SOURCE,
"file_lock_only": False,
"comment_lease_only": False,
}
lease = state["lease"]
freshness = classify_lease_freshness(lease)
decision = decide_safe_next_action(
lease=lease,
caller_session_id=caller_session_id,
freshness=freshness,
caller_worktree=caller_worktree,
)
return {
"success": True,
"found": True,
"lease_id": lease_id,
"lease": lease,
"assignment": state.get("assignment"),
"work_item": state.get("work_item"),
"session": state.get("session"),
"freshness": freshness,
"provenance": state.get("provenance"),
**decision,
"authoritative_source": AUTHORITATIVE_SOURCE,
"file_lock_only": False,
"comment_lease_only": False,
}
def list_active_leases(
db: cpd.ControlPlaneDB,
*,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
role: str | None = None,
include_non_active: bool = False,
limit: int = 100,
) -> dict[str, Any]:
statuses = None if include_non_active else (LEASE_STATUS_ACTIVE,)
rows = db.list_leases(
remote=remote,
org=org,
repo=repo,
role=role,
statuses=statuses,
limit=limit,
)
enriched = []
for row in rows:
fr = classify_lease_freshness(row)
enriched.append({**row, "freshness": fr})
return {
"success": True,
"count": len(enriched),
"leases": enriched,
"authoritative_source": AUTHORITATIVE_SOURCE,
"file_lock_only": False,
"comment_lease_only": False,
"include_non_active": include_non_active,
}
def adopt_lease(
db: cpd.ControlPlaneDB,
*,
lease_id: str,
adopter_session_id: str,
role: str,
worktree_path: str | None = None,
expected_head_sha: str | None = None,
owner_pid: int | None = None,
operator_authorized: bool = False,
) -> dict[str, Any]:
"""Sanctioned adopt path with provenance; never silent foreign steal."""
state = db.get_lease_workflow_state(lease_id)
if not state:
raise LeaseLifecycleError(
f"unknown lease_id {lease_id}; stale prompt id (fail closed)"
)
lease = state["lease"]
work = state["work_item"]
freshness = classify_lease_freshness(lease)
owner = str(lease.get("session_id") or "")
same_owner = owner == str(adopter_session_id)
if freshness["freshness"] == "active" and not same_owner:
raise LeaseLifecycleError(
f"refusing to steal active foreign lease {lease_id} owned by "
f"{owner} (fail closed)"
)
if freshness["freshness"] in ("abandoned", "released"):
raise LeaseLifecycleError(
f"lease {lease_id} is {freshness['freshness']}; cannot adopt "
"(fail closed)"
)
# Expired or stale: require abandon-style safety before ownership transfer
# when not same owner; same owner may reclaim.
if not same_owner and freshness["freshness"] in (
"expired",
"stale_dead_process",
"stale_missing_worktree",
):
if not operator_authorized and freshness["freshness"] == "expired":
# Deterministic reclaim of expired foreign lease is allowed
# without operator flag (sanctioned expire reclaim).
pass
elif freshness["freshness"] != "expired" and not operator_authorized:
# stale active-looking requires abandon proof path
raise LeaseLifecycleError(
f"lease {lease_id} freshness={freshness['freshness']}; "
"use abandon with proof before foreign adopt (fail closed)"
)
provenance = build_adopt_provenance(
adopted_from_session_id=owner,
adopted_by_session_id=adopter_session_id,
work_kind=str(work.get("kind")),
work_number=int(work.get("number")),
remote=str(work.get("remote")),
org=str(work.get("org")),
repo=str(work.get("repo")),
worktree_path=worktree_path,
expected_head_sha=expected_head_sha or lease.get("expected_head_sha"),
prior_lease_id=lease_id,
reason=(
"owner-resume-adopt" if same_owner else "sanctioned-reclaim-adopt"
),
)
result = db.adopt_lease(
lease_id=lease_id,
adopter_session_id=adopter_session_id,
role=role,
worktree_path=worktree_path,
expected_head_sha=expected_head_sha,
owner_pid=owner_pid if owner_pid is not None else os.getpid(),
provenance=provenance,
)
return {
"success": True,
"outcome": result.get("outcome"),
"same_owner": same_owner,
"prior_lease_id": lease_id,
"lease": result.get("lease"),
"assignment": result.get("assignment"),
"provenance": provenance,
"authoritative_source": AUTHORITATIVE_SOURCE,
"file_lock_only": False,
"comment_lease_only": False,
"reasons": result.get("reasons") or [],
}
def release_lease(
db: cpd.ControlPlaneDB,
*,
lease_id: str,
session_id: str,
) -> dict[str, Any]:
proof = db.release_lease_recorded(lease_id=lease_id, session_id=session_id)
return {
"success": True,
"outcome": "released",
"lease_id": lease_id,
"session_id": session_id,
"release_proof": proof,
"authoritative_source": AUTHORITATIVE_SOURCE,
"file_lock_only": False,
"comment_lease_only": False,
}
def expire_leases(db: cpd.ControlPlaneDB) -> dict[str, Any]:
n = db.expire_stale_leases()
return {
"success": True,
"outcome": "expired",
"expired_count": int(n),
"authoritative_source": AUTHORITATIVE_SOURCE,
"file_lock_only": False,
"comment_lease_only": False,
}
def reclaim_expired_lease(
db: cpd.ControlPlaneDB,
*,
lease_id: str,
session_id: str,
role: str,
worktree_path: str | None = None,
expected_head_sha: str | None = None,
) -> dict[str, Any]:
"""Expire-if-needed then assign to *session_id* for the same work item."""
state = db.get_lease_workflow_state(lease_id)
if not state:
raise LeaseLifecycleError(f"unknown lease_id {lease_id}")
lease = state["lease"]
work = state["work_item"]
freshness = classify_lease_freshness(lease)
if freshness["freshness"] == "active":
if lease.get("session_id") != session_id:
raise LeaseLifecycleError(
f"cannot reclaim active foreign lease {lease_id} (fail closed)"
)
# owner resume
return adopt_lease(
db,
lease_id=lease_id,
adopter_session_id=session_id,
role=role,
worktree_path=worktree_path,
expected_head_sha=expected_head_sha,
)
if freshness["freshness"] not in (
"expired",
"stale_dead_process",
"stale_missing_worktree",
"abandoned",
"released",
):
raise LeaseLifecycleError(
f"lease {lease_id} freshness={freshness['freshness']} not reclaimable"
)
# Ensure expired marker is applied
db.expire_stale_leases()
# If still active due to clock skew / non-time stale, force expire this lease
refreshed = db.get_lease_workflow_state(lease_id)
if refreshed and refreshed["lease"].get("status") == "active":
db.force_expire_lease(lease_id, reason="reclaim_expired_lease")
head = expected_head_sha or work.get("current_head_sha")
assigned = db.assign_and_lease(
session_id=session_id,
role=role,
remote=str(work["remote"]),
org=str(work["org"]),
repo=str(work["repo"]),
kind=str(work["kind"]),
number=int(work["number"]),
expected_head_sha=head,
phase="reclaimed",
worktree_path=worktree_path,
owner_pid=os.getpid(),
)
if assigned.outcome != "assigned":
raise LeaseLifecycleError(
f"reclaim assign failed: {assigned.outcome}{assigned.reason}"
)
provenance = build_adopt_provenance(
adopted_from_session_id=str(lease.get("session_id")),
adopted_by_session_id=session_id,
work_kind=str(work["kind"]),
work_number=int(work["number"]),
remote=str(work["remote"]),
org=str(work["org"]),
repo=str(work["repo"]),
worktree_path=worktree_path,
expected_head_sha=head,
prior_lease_id=lease_id,
reason="reclaim_expired",
)
db.attach_lease_provenance(assigned.lease_id, provenance)
return {
"success": True,
"outcome": "reclaimed",
"prior_lease_id": lease_id,
"assignment": assigned.as_dict(),
"provenance": provenance,
"authoritative_source": AUTHORITATIVE_SOURCE,
"file_lock_only": False,
"comment_lease_only": False,
}
def abandon_lease(
db: cpd.ControlPlaneDB,
*,
lease_id: str,
requester_session_id: str,
proof: AbandonProof,
) -> dict[str, Any]:
state = db.get_lease_workflow_state(lease_id)
if not state:
raise LeaseLifecycleError(f"unknown lease_id {lease_id}")
lease = state["lease"]
owner = str(lease.get("session_id") or "")
same_owner = owner == str(requester_session_id)
# Enrich proof with live checks when not provided
owner_pid = proof.owner_pid
if owner_pid is None:
owner_pid = lease.get("owner_pid")
wt = proof.worktree_path if proof.worktree_path is not None else lease.get(
"worktree_path"
)
dead = proof.dead_process or (
owner_pid is not None and not is_process_alive(owner_pid)
)
missing_wt = proof.missing_worktree or (
bool(wt) and not worktree_exists(str(wt))
)
enriched = AbandonProof(
dead_process=bool(dead),
missing_worktree=bool(missing_wt),
same_owner=same_owner or proof.same_owner,
no_open_pr=proof.no_open_pr,
no_live_mutation_risk=proof.no_live_mutation_risk,
operator_authorized=proof.operator_authorized,
worktree_path=str(wt) if wt else None,
owner_pid=int(owner_pid) if owner_pid is not None else None,
notes=proof.notes,
)
if not enriched.is_sufficient():
raise LeaseLifecycleError(
"abandon proof insufficient: need (dead_process or missing_worktree) "
"and no_live_mutation_risk; foreign abandon also needs "
"operator_authorized or (dead+missing_worktree+no_open_pr). "
f"proof={enriched.as_dict()}"
)
# Active foreign with live process + present worktree: never abandon without
# operator (already covered by is_sufficient).
result = db.abandon_lease(
lease_id=lease_id,
requester_session_id=requester_session_id,
proof=enriched.as_dict(),
)
return {
"success": True,
"outcome": "abandoned",
"lease_id": lease_id,
"requester_session_id": requester_session_id,
"prior_owner_session_id": owner,
"abandon_proof": enriched.as_dict(),
"audit": result,
"authoritative_source": AUTHORITATIVE_SOURCE,
"file_lock_only": False,
"comment_lease_only": False,
}
+59
View File
@@ -149,6 +149,65 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.read", "permission": "gitea.read",
"role": "author", "role": "author",
}, },
# #601 first-class lease lifecycle — inspect/list need read; mutations gate on
# ownership in the control-plane DB (not a separate Gitea write permission).
"list_workflow_leases": {
"permission": "gitea.read",
"role": "author",
},
"gitea_list_workflow_leases": {
"permission": "gitea.read",
"role": "author",
},
"inspect_workflow_lease": {
"permission": "gitea.read",
"role": "author",
},
"gitea_inspect_workflow_lease": {
"permission": "gitea.read",
"role": "author",
},
"adopt_workflow_lease": {
"permission": "gitea.read",
"role": "author",
},
"gitea_adopt_workflow_lease": {
"permission": "gitea.read",
"role": "author",
},
"release_workflow_lease": {
"permission": "gitea.read",
"role": "author",
},
"gitea_release_workflow_lease": {
"permission": "gitea.read",
"role": "author",
},
"expire_workflow_leases": {
"permission": "gitea.read",
"role": "author",
},
"gitea_expire_workflow_leases": {
"permission": "gitea.read",
"role": "author",
},
"abandon_workflow_lease": {
"permission": "gitea.read",
"role": "author",
},
"gitea_abandon_workflow_lease": {
"permission": "gitea.read",
"role": "author",
},
"reclaim_expired_workflow_lease": {
"permission": "gitea.read",
"role": "author",
},
"gitea_reclaim_expired_workflow_lease": {
"permission": "gitea.read",
"role": "author",
},
# #612 incident bridge — reconcile uses create_issue for apply; # #612 incident bridge — reconcile uses create_issue for apply;
# dry-run needs read only. Tools gate apply paths themselves. # dry-run needs read only. Tools gate apply paths themselves.
"observability_reconcile_incident": { "observability_reconcile_incident": {
+1 -1
View File
@@ -36,7 +36,7 @@ class ControlPlaneDBTest(unittest.TestCase):
rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall()) rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall())
finally: finally:
conn.close() conn.close()
self.assertEqual(rows["schema_version"], "2") self.assertEqual(rows["schema_version"], "3")
self.assertIn("DB coordinates", rows["architecture"]) self.assertIn("DB coordinates", rows["architecture"])
self.assertIn("bridge", rows["architecture"].lower()) self.assertIn("bridge", rows["architecture"].lower())
+22 -2
View File
@@ -103,20 +103,40 @@ class TestIssueLockStore(unittest.TestCase):
self.assertIn("live foreign issue lock", block or "") self.assertIn("live foreign issue lock", block or "")
def test_expired_lease_allows_takeover_with_conflict_check(self): def test_expired_lease_allows_takeover_with_conflict_check(self):
# #601: expired + dead pid / missing worktree → sanctioned reclaim (no block).
existing = _lock_record( existing = _lock_record(
branch_name="feat/issue-420-other", branch_name="feat/issue-420-other",
worktree_path="/tmp/other", worktree_path="/tmp/other-does-not-exist-420",
session_pid=1,
work_lease=_lease("2000-01-01T00:00:00Z"), work_lease=_lease("2000-01-01T00:00:00Z"),
) )
incoming = _lock_record(worktree_path="/tmp/mine") incoming = _lock_record(worktree_path="/tmp/mine")
self.assertIsNone(ils.assess_foreign_lock_overwrite(existing, incoming)) self.assertIsNone(ils.assess_foreign_lock_overwrite(existing, incoming))
with mock.patch.object(ils, "is_process_alive", return_value=False):
block = ils.assess_same_issue_lease_conflict( block = ils.assess_same_issue_lease_conflict(
existing, existing,
issue_number=420, issue_number=420,
branch_name="feat/issue-420-server-code-parity", branch_name="feat/issue-420-server-code-parity",
worktree_path="/tmp/mine", worktree_path="/tmp/mine",
) )
self.assertIn("Recovery review is required", block or "") self.assertIsNone(block)
# Expired but still-live owner pid AND present worktree → fail closed.
present_wt = self.lock_dir # exists
sticky = _lock_record(
branch_name="feat/issue-420-other",
worktree_path=present_wt,
session_pid=os.getpid(),
work_lease=_lease("2000-01-01T00:00:00Z"),
)
with mock.patch.object(ils, "is_process_alive", return_value=True):
blocked = ils.assess_same_issue_lease_conflict(
sticky,
issue_number=420,
branch_name="feat/issue-420-server-code-parity",
worktree_path="/tmp/mine",
)
self.assertIn("Recovery review is required", blocked or "")
def test_same_owner_lease_conflict_allows_refresh(self): def test_same_owner_lease_conflict_allows_refresh(self):
worktree = "/tmp/wt-420" worktree = "/tmp/wt-420"
+392
View File
@@ -0,0 +1,392 @@
"""Tests for first-class control-plane lease lifecycle (#601)."""
from __future__ import annotations
import os
import tempfile
import unittest
from datetime import timedelta
from unittest import mock
from allocator_service import allocate_next_work, WorkCandidate
from control_plane_db import ControlPlaneDB, ForeignLeaseError, _ts, _utc_now
import lease_lifecycle as ll
import merger_lease_adoption as mla
import reviewer_pr_lease as rpl
from datetime import datetime, timezone
class LeaseLifecycleTest(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
self.db = ControlPlaneDB(self.db_path)
self.db.upsert_session(session_id="owner", role="author", profile="prgs-author", pid=111)
self.db.upsert_session(session_id="other", role="author", profile="prgs-author", pid=222)
def tearDown(self) -> None:
self._tmp.cleanup()
def _assign(self, session_id="owner", number=601, **kwargs):
return self.db.assign_and_lease(
session_id=session_id,
role="author",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
kind="issue",
number=number,
allowed_actions=("implement", "comment", "push", "create_pr"),
forbidden_actions=("approve", "merge", "self_select_without_assignment"),
worktree_path=kwargs.pop("worktree_path", self._tmp.name),
owner_pid=kwargs.pop("owner_pid", os.getpid()),
**kwargs,
)
def test_list_active_leases_as_workflow_state(self) -> None:
a = self._assign(number=601)
self._assign(session_id="other", number=602)
listed = ll.list_active_leases(
self.db, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools"
)
self.assertTrue(listed["success"])
self.assertEqual(listed["count"], 2)
self.assertEqual(listed["authoritative_source"], "control_plane_db")
self.assertFalse(listed["file_lock_only"])
self.assertFalse(listed["comment_lease_only"])
numbers = sorted(x["work_number"] for x in listed["leases"])
self.assertEqual(numbers, [601, 602])
self.assertIsNotNone(a.lease_id)
def test_adopt_lease_owner_resume_preserves_provenance(self) -> None:
a = self._assign()
res = ll.adopt_lease(
self.db,
lease_id=a.lease_id,
adopter_session_id="owner",
role="author",
worktree_path=self._tmp.name,
)
self.assertTrue(res["success"])
self.assertTrue(res["same_owner"])
prov = res["provenance"]
self.assertEqual(prov["adopted_from_session_id"], "owner")
self.assertEqual(prov["adopted_by_session_id"], "owner")
self.assertEqual(prov["work_number"], 601)
self.assertEqual(prov["worktree_path"], self._tmp.name)
state = self.db.get_lease_workflow_state(a.lease_id)
self.assertIsNotNone(state)
self.assertEqual(state["lease"]["status"], "active")
def test_release_lease_explicit_recorded(self) -> None:
a = self._assign()
res = ll.release_lease(self.db, lease_id=a.lease_id, session_id="owner")
self.assertEqual(res["outcome"], "released")
self.assertIn("release_proof", res)
self.assertEqual(res["release_proof"]["status"], "released")
state = self.db.get_lease_workflow_state(a.lease_id)
self.assertEqual(state["lease"]["status"], "released")
def test_expire_and_reclaim_expired_lease(self) -> None:
a = self._assign(lease_ttl_seconds=1)
past = _utc_now() - timedelta(hours=2)
import sqlite3
conn = sqlite3.connect(self.db_path)
try:
conn.execute(
"UPDATE leases SET expires_at = ? WHERE lease_id = ?",
(_ts(past), a.lease_id),
)
conn.commit()
finally:
conn.close()
exp = ll.expire_leases(self.db)
self.assertGreaterEqual(exp["expired_count"], 1)
reclaimed = ll.reclaim_expired_lease(
self.db,
lease_id=a.lease_id,
session_id="other",
role="author",
worktree_path=self._tmp.name,
)
self.assertEqual(reclaimed["outcome"], "reclaimed")
self.assertEqual(reclaimed["assignment"]["session_id"], "other")
self.assertEqual(
reclaimed["provenance"]["adopted_from_session_id"], "owner"
)
self.assertEqual(
reclaimed["provenance"]["adopted_by_session_id"], "other"
)
def test_abandon_stale_lease_with_required_proof(self) -> None:
a = self._assign(owner_pid=99999999, worktree_path="/nonexistent/path/for-601")
# Force active but dead pid + missing worktree
proof = ll.AbandonProof(
dead_process=True,
missing_worktree=True,
no_open_pr=True,
no_live_mutation_risk=True,
owner_pid=99999999,
worktree_path="/nonexistent/path/for-601",
)
res = ll.abandon_lease(
self.db,
lease_id=a.lease_id,
requester_session_id="other",
proof=proof,
)
self.assertEqual(res["outcome"], "abandoned")
self.assertEqual(res["prior_owner_session_id"], "owner")
self.assertTrue(res["abandon_proof"]["dead_process"])
state = self.db.get_lease_workflow_state(a.lease_id)
self.assertEqual(state["lease"]["status"], "abandoned")
def test_refuse_steal_active_foreign_lease(self) -> None:
a = self._assign()
with self.assertRaises(ll.LeaseLifecycleError) as ctx:
ll.adopt_lease(
self.db,
lease_id=a.lease_id,
adopter_session_id="other",
role="author",
)
self.assertIn("steal", str(ctx.exception).lower())
decision = ll.inspect_lease(
self.db, a.lease_id, caller_session_id="other"
)
self.assertEqual(decision["safe_next_action"], ll.SAFE_WAIT_FOREIGN)
self.assertTrue(decision["block"])
def test_refuse_ambiguous_lease_ownership_stale_id(self) -> None:
decision = ll.inspect_lease(
self.db, "lease-does-not-exist", caller_session_id="owner"
)
self.assertFalse(decision["found"])
self.assertEqual(decision["safe_next_action"], ll.SAFE_STALE_PROMPT)
with self.assertRaises(ll.LeaseLifecycleError):
ll.adopt_lease(
self.db,
lease_id="lease-does-not-exist",
adopter_session_id="owner",
role="author",
)
def test_provenance_on_adopt_release_abandon(self) -> None:
a = self._assign()
adopted = ll.adopt_lease(
self.db,
lease_id=a.lease_id,
adopter_session_id="owner",
role="author",
worktree_path=self._tmp.name,
expected_head_sha="abc",
)
self.assertIn("adopted_from_session_id", adopted["provenance"])
self.assertIn("adopted_by_session_id", adopted["provenance"])
released = ll.release_lease(
self.db, lease_id=a.lease_id, session_id="owner"
)
self.assertEqual(released["release_proof"]["session_id"], "owner")
b = self._assign(number=700, owner_pid=1, worktree_path="/no/such/wt")
abandoned = ll.abandon_lease(
self.db,
lease_id=b.lease_id,
requester_session_id="owner",
proof=ll.AbandonProof(
dead_process=True,
missing_worktree=True,
no_live_mutation_risk=True,
no_open_pr=True,
),
)
self.assertIn("abandon_proof", abandoned)
self.assertEqual(abandoned["abandon_proof"]["dead_process"], True)
def test_allocator_assignment_lease_compatible(self) -> None:
"""Allocator assign+lease still works and is listable/inspectable."""
cands = [
WorkCandidate(
kind="issue",
number=601,
labels=("status:ready",),
title="leases",
priority=20,
)
]
res = allocate_next_work(
self.db,
session_id="alloc-s",
role="author",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
candidates=cands,
apply=True,
profile_name="prgs-author",
username="jcwalker3",
)
self.assertEqual(res["outcome"], "assigned_work")
lid = res["assignment"]["lease_id"]
listed = ll.list_active_leases(self.db, remote="prgs")
ids = [x["lease_id"] for x in listed["leases"]]
self.assertIn(lid, ids)
insp = ll.inspect_lease(self.db, lid, caller_session_id="alloc-s")
self.assertTrue(insp["found"])
self.assertIn(
insp["safe_next_action"],
(ll.SAFE_OWNER_RESUME, ll.SAFE_ABANDON_ALLOWED),
)
def test_reviewer_merger_lease_handoff_still_works(self) -> None:
"""#536 merger adoption path is independent and must not regress."""
rpl.clear_session_lease()
body = mla.format_adoption_body(
repo="Scaled-Tech-Consulting/Gitea-Tools",
pr_number=999,
issue_number=601,
adopter_identity="sysadmin",
adopter_profile="prgs-merger",
adopter_session_id="merger-1",
worktree="branches/merge-pr999",
candidate_head="a" * 40,
target_branch="master",
target_branch_sha="b" * 40,
adopted_from_session_id="reviewer-1",
adopted_from_profile="prgs-reviewer",
adopted_from_reviewer_identity="sysadmin",
adopted_from_comment_id=42,
)
self.assertIn(mla.ADOPTION_MARKER, body)
self.assertIn("adopted_from_session_id: reviewer-1", body)
self.assertIn("adopted_by_profile: prgs-merger", body)
prov = mla.build_lease_provenance(
source=mla.SOURCE_ADOPT,
comment_id=42,
adopted_from_session_id="reviewer-1",
adoption_reason=mla.DEFAULT_ADOPTION_REASON,
)
rpl.record_session_lease(
{
"pr_number": 999,
"session_id": "merger-1",
"comment_id": 42,
},
lease_provenance=prov,
)
proof = mla.describe_session_lease_proof(rpl.get_session_lease())
self.assertTrue(proof["lease_proof_sanctioned"])
self.assertEqual(proof["lease_proof_source"], mla.SOURCE_ADOPT)
rpl.clear_session_lease()
def test_missing_worktree_and_dead_pid_handled_safely(self) -> None:
a = self._assign(owner_pid=1, worktree_path="/definitely/missing/wt-601")
with mock.patch.object(ll, "is_process_alive", return_value=False):
fr = ll.classify_lease_freshness(
self.db.get_lease_workflow_state(a.lease_id)["lease"]
)
self.assertIn(fr["freshness"], ("stale_dead_process", "stale_missing_worktree"))
# Insufficient abandon proof fails closed
with self.assertRaises(ll.LeaseLifecycleError):
ll.abandon_lease(
self.db,
lease_id=a.lease_id,
requester_session_id="other",
proof=ll.AbandonProof(dead_process=True), # missing no_live_mutation_risk
)
def test_file_lock_or_comment_not_authoritative_alone(self) -> None:
report = ll.non_db_lease_authority_report(
file_lock_present=True,
comment_lease_present=False,
db_lease_present=False,
)
self.assertEqual(report["safe_next_action"], ll.SAFE_NO_AUTHORITY)
self.assertTrue(report["file_lock_only"])
self.assertIsNone(report["authoritative_source"])
report2 = ll.non_db_lease_authority_report(
file_lock_present=True,
comment_lease_present=True,
db_lease_present=True,
)
self.assertEqual(report2["authoritative_source"], "control_plane_db")
self.assertFalse(report2["file_lock_only"])
self.assertFalse(report2["comment_lease_only"])
def test_abandon_proof_is_sufficient_rules(self) -> None:
self.assertFalse(ll.AbandonProof(dead_process=True).is_sufficient())
self.assertTrue(
ll.AbandonProof(
dead_process=True,
missing_worktree=True,
no_live_mutation_risk=True,
no_open_pr=True,
).is_sufficient()
)
self.assertTrue(
ll.AbandonProof(
dead_process=True,
no_live_mutation_risk=True,
same_owner=True,
).is_sufficient()
)
self.assertTrue(
ll.AbandonProof(
missing_worktree=True,
no_live_mutation_risk=True,
operator_authorized=True,
).is_sufficient()
)
class IssueLockExpiredReclaimTest(unittest.TestCase):
def test_expired_lock_reclaim_allowed_when_dead_and_missing_wt(self) -> None:
import issue_lock_store as ils
lock = {
"issue_number": 601,
"branch_name": "feat/issue-601-x",
"worktree_path": "/no/such/worktree-601",
"session_pid": 1,
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2000-01-01T00:00:00Z",
},
}
with mock.patch.object(ils, "is_process_alive", return_value=False):
res = ils.assess_expired_lock_reclaim(lock)
self.assertTrue(res["reclaim_allowed"])
# Conflict assessor allows takeover
block = ils.assess_same_issue_lease_conflict(
lock,
issue_number=601,
branch_name="feat/issue-601-first-class-leases",
worktree_path="/tmp/new",
)
self.assertIsNone(block)
def test_live_lock_reclaim_refused(self) -> None:
import issue_lock_store as ils
from datetime import datetime, timedelta, timezone
future = (datetime.now(timezone.utc) + timedelta(hours=2)).strftime(
"%Y-%m-%dT%H:%M:%SZ"
)
lock = {
"issue_number": 601,
"branch_name": "feat/issue-601-x",
"worktree_path": tempfile.gettempdir(),
"session_pid": os.getpid(),
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": future,
"last_heartbeat_at": future,
},
}
res = ils.assess_expired_lock_reclaim(lock)
self.assertFalse(res["reclaim_allowed"])
if __name__ == "__main__":
unittest.main()