From 42347da0126de363454dd167595ed5297cb68603 Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 19 Oct 2026 15:30:29 +0200 Subject: [PATCH] feat(ops-runs): expire claims that keep losing their lease Lapsed claims were reopened with no attempt ceiling, so a run whose worker lost its lease (including a 409 before dispatch) could cycle forever. Activity Core now reopens a lapsed claim below OPS_RUN_MAX_ATTEMPTS and sets terminal 'expired' at the ceiling. The owner is kept for audit, and /ops-runs/expire-leases reports reopened and expired counts. Answers the REINAH-WP-0003-T05 open point on thread 27cd4c47 (ACTIVITY-WP-ADHOC-2026-10-19). Co-Authored-By: Claude Opus 5.5 Assistant: claude-code Assistant-Model: opus Assistant-Process: 244572@bnt-lap001 Assistant-Session: 3b235e1a-ae45-4d0f-ae70-17a9bd78ece8 --- WORK-RECORDS.md | 3 + docs/ops-run-queue.md | 9 ++- src/activity_core/ops_run_queue.py | 52 ++++++++++++++---- src/activity_core/ops_runs_api.py | 6 +- tests/test_ops_run_queue.py | 88 ++++++++++++++++++++++++++++++ workplans/ADHOC-2026-10-19.md | 48 ++++++++++++++++ 6 files changed, 192 insertions(+), 14 deletions(-) create mode 100644 workplans/ADHOC-2026-10-19.md diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 81d58ac..ad100e5 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -54,6 +54,7 @@ | workplan | ACTIVITY-WP-ADHOC-2026-08-20 | finished | — | workplans/ADHOC-2026-08-20.md | | workplan | ACTIVITY-WP-ADHOC-2026-08-23 | finished | — | workplans/ADHOC-2026-08-23.md | | workplan | ACTIVITY-WP-ADHOC-2026-09-14 | finished | — | workplans/ADHOC-2026-09-14.md | +| workplan | ACTIVITY-WP-ADHOC-2026-10-19 | active | — | workplans/ADHOC-2026-10-19.md | | workplan | custodian-WP-0001 | done | — | workplans/custodian-WP-0001-temporal-backbone.md | | workplan | custodian-WP-0002 | done | — | workplans/custodian-WP-0002-triggers-ops.md | | workplan | custodian-WP-0003a | done | — | workplans/custodian-WP-0003a-event-bridge-model-rules-registry.md | @@ -279,6 +280,8 @@ | task | ACTIVITY-WP-ADHOC-2026-08-23-T01 | done | — | workplans/ADHOC-2026-08-23.md | | task | ACTIVITY-WP-ADHOC-2026-09-14-T01 | done | — | workplans/ADHOC-2026-09-14.md | | task | ACTIVITY-WP-ADHOC-2026-09-14-T02 | done | — | workplans/ADHOC-2026-09-14.md | +| task | ACTIVITY-WP-ADHOC-2026-10-19-T01 | done | — | workplans/ADHOC-2026-10-19.md | +| task | ACTIVITY-WP-ADHOC-2026-10-19-T02 | wait | — | workplans/ADHOC-2026-10-19.md | | task | T01 | done | — | workplans/custodian-WP-0001-temporal-backbone.md | | task | T02 | done | — | workplans/custodian-WP-0001-temporal-backbone.md | | task | T03 | done | — | workplans/custodian-WP-0001-temporal-backbone.md | diff --git a/docs/ops-run-queue.md b/docs/ops-run-queue.md index e7573e7..a04685c 100644 --- a/docs/ops-run-queue.md +++ b/docs/ops-run-queue.md @@ -61,7 +61,14 @@ workplan task file. Not an issue-core or Forgejo ticket. - `labels_mode`: `any` (default) — run must contain at least one listed label; `all` — run must contain every listed label; omit `labels` to claim any open run. - Claim uses `FOR UPDATE SKIP LOCKED` for concurrency safety. -- Stale claims (`state=claimed` and `lease_until <= now()`) are reopened before select. +- Stale claims (`state=claimed` and `lease_until <= now()`) are recovered by + Activity Core before every claim select and by `POST /ops-runs/expire-leases` + (returns `{"reopened": n, "expired": m}`). Below `OPS_RUN_MAX_ATTEMPTS` the row + returns to `open` with owner and lease cleared; at the ceiling it becomes + terminal `expired` (owner kept for audit, `result.error` set). A worker whose + lease was lost, including one that got a 409 before dispatch, must not close + the row; it leaves recovery to this expiry path. A late close on an expired + row is refused as `state_conflict`. - Heartbeat, complete, and fail lock the row and require an active lease (`lease_until > now()`). An expired worker cannot revive or close its claim. - A first terminal close returns `close_disposition: applied`. If its response diff --git a/src/activity_core/ops_run_queue.py b/src/activity_core/ops_run_queue.py index a3a3832..c3d3fc3 100644 --- a/src/activity_core/ops_run_queue.py +++ b/src/activity_core/ops_run_queue.py @@ -219,17 +219,47 @@ async def create_ops_run_from_spec( async def reopen_stale_claims(session: AsyncSession) -> int: - """Return claimed rows with expired leases to open.""" + """Return claimed rows with expired leases to open; expire exhausted ones. + + A claim whose lease lapsed without a terminal close (crash, lost lease, + or a worker that refused dispatch after a 409) is Activity Core's to + recover: the worker that lost the lease must not close it. Below the + attempt ceiling the row returns to ``open``; at or above + ``OPS_RUN_MAX_ATTEMPTS`` it becomes terminal ``expired`` so a row that + keeps losing its lease cannot cycle forever. Returns the number of rows + reopened (see :func:`expire_stale_claims` for both counts). + """ + counts = await expire_stale_claims(session) + return counts["reopened"] + + +async def expire_stale_claims(session: AsyncSession) -> dict[str, int]: + """Reopen or terminally expire claimed rows whose lease has lapsed.""" now = _utcnow() - stmt = ( + stale = and_( + OpsRun.state == "claimed", + OpsRun.lease_until.is_not(None), + OpsRun.lease_until <= now, + ) + ceiling = max_attempts() + expired = await session.execute( update(OpsRun) - .where( - and_( - OpsRun.state == "claimed", - OpsRun.lease_until.is_not(None), - OpsRun.lease_until <= now, - ) + .where(stale, OpsRun.attempt >= ceiling) + .values( + state="expired", + lease_until=None, + result={ + "error": ( + f"lease expired without terminal close after {ceiling} " + "attempt(s); expired by activity-core" + ) + }, + updated_at=now, ) + ) + reopened = await session.execute( + update(OpsRun) + .where(stale, OpsRun.attempt < ceiling) .values( state="open", claim_owner=None, @@ -237,8 +267,10 @@ async def reopen_stale_claims(session: AsyncSession) -> int: updated_at=now, ) ) - result = await session.execute(stmt) - return int(result.rowcount or 0) + return { + "reopened": int(reopened.rowcount or 0), + "expired": int(expired.rowcount or 0), + } async def claim_ops_runs( diff --git a/src/activity_core/ops_runs_api.py b/src/activity_core/ops_runs_api.py index 31797ce..4d9e229 100644 --- a/src/activity_core/ops_runs_api.py +++ b/src/activity_core/ops_runs_api.py @@ -30,7 +30,7 @@ from activity_core.ops_run_queue import ( list_ops_runs, ops_run_counts, ops_run_to_dict, - reopen_stale_claims, + expire_stale_claims, ) router = APIRouter(prefix="/ops-runs", tags=["ops-runs"]) @@ -479,5 +479,5 @@ async def post_expire_leases( Session = _db() async with Session() as session: async with session.begin(): - n = await reopen_stale_claims(session) - return {"reopened": n} + counts = await expire_stale_claims(session) + return counts diff --git a/tests/test_ops_run_queue.py b/tests/test_ops_run_queue.py index 3860645..2db32aa 100644 --- a/tests/test_ops_run_queue.py +++ b/tests/test_ops_run_queue.py @@ -632,3 +632,91 @@ def test_label_filter_any_vs_all() -> None: want = {"research-brief", "missing"} assert bool(want & row_labels) # any assert not want.issubset(row_labels) # all + + +@pytest.mark.asyncio +async def test_expire_stale_claims_reopens_below_and_expires_at_ceiling( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A lapsed lease returns to open below the ceiling, else terminal expired.""" + from sqlalchemy.dialects import postgresql + + from activity_core import ops_run_queue as oq + + monkeypatch.setenv("OPS_RUN_MAX_ATTEMPTS", "3") + statements: list = [] + + async def execute(stmt, *args, **kwargs): + statements.append(stmt) + res = MagicMock() + res.rowcount = 2 if len(statements) == 1 else 5 + return res + + session = AsyncMock() + session.execute = AsyncMock(side_effect=execute) + + counts = await oq.expire_stale_claims(session) + assert counts == {"expired": 2, "reopened": 5} + assert len(statements) == 2 + + compiled = [ + s.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": False}) + for s in statements + ] + expire_sql, reopen_sql = (str(c) for c in compiled) + expire_params, reopen_params = (c.params for c in compiled) + + # Both statements only touch claimed rows whose lease has lapsed. + for sql in (expire_sql, reopen_sql): + assert "ops_runs.state = " in sql + assert "ops_runs.lease_until IS NOT NULL" in sql + assert "ops_runs.lease_until <= " in sql + assert "ops_runs.attempt >= " in expire_sql + assert "ops_runs.attempt < " in reopen_sql + + assert expire_params["state"] == "expired" + assert "claimed" in expire_params.values() + assert 3 in expire_params.values() + assert expire_params["lease_until"] is None + assert "expired by activity-core" in expire_params["result"]["error"] + # The original claim owner stays on the expired row for audit. + assert "claim_owner" not in expire_params + + assert reopen_params["state"] == "open" + assert reopen_params["claim_owner"] is None + assert reopen_params["lease_until"] is None + + +@pytest.mark.asyncio +async def test_reopen_stale_claims_returns_reopened_count() -> None: + from activity_core import ops_run_queue as oq + + session = AsyncMock() + with patch.object( + oq, + "expire_stale_claims", + AsyncMock(return_value={"reopened": 4, "expired": 1}), + ): + assert await oq.reopen_stale_claims(session) == 4 + + +@pytest.mark.asyncio +async def test_lost_owner_cannot_close_an_expired_row() -> None: + """After activity-core expires a lapsed claim, a late close is refused.""" + from activity_core import ops_run_queue as oq + + row = MagicMock() + row.state = "expired" + row.claim_owner = "worker-1" + row.lease_until = None + row.result = {"error": "lease expired"} + row.repository_grant = None + row.close_intent_digest = None + session = AsyncMock() + session.get = AsyncMock(return_value=row) + + outcome = await oq.close_ops_run( + session, uuid.uuid4(), worker_id="worker-1", action="fail", error="late" + ) + assert outcome.disposition == "state_conflict" + assert row.state == "expired" diff --git a/workplans/ADHOC-2026-10-19.md b/workplans/ADHOC-2026-10-19.md new file mode 100644 index 0000000..829e822 --- /dev/null +++ b/workplans/ADHOC-2026-10-19.md @@ -0,0 +1,48 @@ +--- +id: ACTIVITY-WP-ADHOC-2026-10-19 +type: workplan +title: Activity Core owns recovery of lapsed ops_run leases +domain: infotech +repo: activity-core +status: active +owner: claude-activity-core +created: "2026-10-19" +updated: "2026-10-19" +related: [REINAH-WP-0003] +--- + +Origin: rein-aharness REINAH-WP-0003-T05 open point on thread 27cd4c47. When +the worker's first heartbeat gets a 409 before dispatch, the run stays +`claimed`, because a worker must not close a lease it has lost. Activity +Core's position: that is correct, and recovery is Activity Core's job. + +## Expire claims that keep losing their lease + +```task +id: ACTIVITY-WP-ADHOC-2026-10-19-T01 +status: done +priority: high +``` + +`reopen_stale_claims` returned every lapsed claim to `open` with no attempt +ceiling, so a row that kept losing its lease could cycle forever. New +`expire_stale_claims` in `ops_run_queue.py`, which runs before each claim and +behind `POST /ops-runs/expire-leases`, now reopens lapsed claims below +`OPS_RUN_MAX_ATTEMPTS`. At the ceiling it sets terminal `expired`, keeps the +owner for audit and records `result.error`. A late close on an expired row is +refused as `state_conflict`. The endpoint returns +`{"reopened": n, "expired": m}`. Contract documented in +`docs/ops-run-queue.md`, with three new tests in `tests/test_ops_run_queue.py`. + +## Release the expiry change to production + +```task +id: ACTIVITY-WP-ADHOC-2026-10-19-T02 +status: wait +blocked_on: human-gate:image-release +priority: medium +``` + +Wait: this ships with the next image release through the normal GitOps path, +the same release that carries repository grant v2 (ACTIVITY-WP-0046-T03). No +rollout from this session.