Implement ACTIVITY-WP-0026 ops_run claim queue (T01–T06).

Add durable claimable ops_runs table, emit dual-write on TaskSpec, REST claim/lease/complete/fail API, ops status visibility, and consumer docs aligned with ACT-ADR-005. T07 railiance rollout remains deploy-side.
2026-08-03 19:22:50 +02:00 · 2026-08-03 19:22:50 +02:00 · 15eb3a2066
commit 15eb3a2066
parent a145cc4027
15 changed files with 1294 additions and 27 deletions
--- a/.env.example
+++ b/.env.example
@ -25,6 +25,19 @@ ISSUE_CORE_API_KEY=
 # state-hub (default, no Forgejo) | null (dry-run) | rest (issue-core opt-in)
 ISSUE_SINK_TYPE=state-hub

+# ── Ops run claim queue (ACTIVITY-WP-0026 / ACT-ADR-005) ───────────────────────
+# Create claimable ops_run rows on emit_tasks (default true). Harness claims via
+# POST /ops-runs/claim — not issue-core / Forgejo.
+OPS_RUN_QUEUE_ENABLED=true
+# Default claim lease (seconds); heartbeat extends.
+OPS_RUN_LEASE_SECONDS=900
+# Permanent fail after this many claims when fail+reopen is requested.
+OPS_RUN_MAX_ATTEMPTS=3
+# Stuck open/claimed threshold for /ops/automations/status ops_runs.sla_hours
+OPS_RUN_SLA_HOURS=1
+# Worker credential for claim/complete/fail (X-Worker-Token or Bearer).
+ACTIVITY_CORE_WORKER_TOKEN=
+
 # ── Activity definitions ───────────────────────────────────────────────────────
 # Colon-separated paths to additional activity-definitions/ directories.
 # The local activity-definitions/ directory is always scanned.
--- a/docs/issue-core-emission-boundary.md
+++ b/docs/issue-core-emission-boundary.md
@ -3,6 +3,20 @@
 activity-core owns the decision to spawn a task and the audit trail that says
 why it spawned. It does not own downstream task lifecycle state after emission.

+## Ops claim vs issue-core (ACT-ADR-005 / ACTIVITY-WP-0026)
+
+| Concern | Home |
+| ------- | ---- |
+| **Claimable automation run** | `ops_runs` table + `POST /ops-runs/claim` in activity-core |
+| **Fleet visibility / progress** | IssueSink (`state-hub` default) |
+| **External tracker ticket** | issue-core → **Forgejo** (opt-in `ISSUE_SINK_TYPE=rest` only) |
+
+**`ops_run` is not issue-core.** Harnesses must claim ops runs; they must not
+poll issue-core or Forgejo for scheduled FI/Binky-style work. Full contract:
+`docs/ops-run-queue.md` and `docs/task-emission-consumer-contract.md`.
+
+Self-hosted forge product name: **Forgejo** only (no Gitea support path).
+
 ## Sink matrix (ACTIVITY-WP-0022)

 | `ISSUE_SINK_TYPE` | Destination | Default? |
@ -72,6 +86,9 @@ activity-core.
 Failed to connect to backend 'forgejo-inbox': Failed to connect to Gitea API
 ```

+(Historical error string from the Forgejo-compatible client library — the
+deployed forge is **Forgejo**, not Gitea.)
+
 **Disposition (ACTIVITY-WP-0023-T06):** activity-core keeps global default
 `state-hub` and does **not** flip production to `rest`. Path A is owned by
 **issue-core**: rotate/fix `GITEA_BACKEND_TOKEN` (Forgejo backend PAT for the
@ -79,6 +96,9 @@ forgejo-inbox connector — not the activity-core `ISSUE_CORE_API_KEY`
 ingestion key). After issue-core proves `POST /issues/` → **201**, operators may
 opt in per definition / overlay only (WP-0022).

+Internal scheduled automation should use **ops_run claim** (WP-0026), not wait
+on this rest path.
+
 Smoke from worker (does not change sink env):

 ```bash
--- a/docs/ops-run-queue.md
+++ b/docs/ops-run-queue.md
@ -0,0 +1,111 @@
+# Ops run claim queue
+
+**Status:** implemented (ACTIVITY-WP-0026 code path; railiance rollout = T07)  
+**Architecture:** [ACT-ADR-005](adr/adr-005-ops-runs-vs-dev-work-records.md)
+
+Durable, claimable work instances for **scheduled / automation** runs. Not a
+workplan task file. Not an issue-core or Forgejo ticket.
+
+## Model
+
+| Field | Type | Notes |
+| ----- | ---- | ----- |
+| `id` | UUID | Primary key (uuid4; UUIDv7 optional later) |
+| `activity_definition_id` | UUID | FK → activity_definitions |
+| `idempotency_key` | text | **Unique**; default `{activity_id}:{source_id}:{triggering_event_id}` |
+| `target_repo` | text | From TaskSpec |
+| `title` | text | |
+| `description` | text | |
+| `labels` | JSONB array | e.g. `["automated","research-brief"]` |
+| `priority` | text | low \| medium \| high |
+| `state` | text | `open` \| `claimed` \| `succeeded` \| `failed` \| `expired` |
+| `claim_owner` | text nullable | Worker identity |
+| `lease_until` | timestamptz nullable | Claim lease deadline |
+| `attempt` | int | Starts at 0; incremented on each claim |
+| `source_type` | text | rule \| instruction |
+| `source_id` | text | Rule id |
+| `triggering_event_id` | text | Event or workflow key |
+| `approach_hint` | text nullable | Optional from rule |
+| `result` | JSONB | Completion metadata |
+| `created_at` / `updated_at` | timestamptz | |
+
+## API (actcore-api)
+
+| Method | Path | Role |
+| ------ | ---- | ---- |
+| `GET` | `/ops-runs` | List/filter |
+| `GET` | `/ops-runs/{id}` | One row |
+| `POST` | `/ops-runs/claim` | Claim open runs (lease) |
+| `POST` | `/ops-runs/{id}/heartbeat` | Extend lease |
+| `POST` | `/ops-runs/{id}/complete` | succeeded |
+| `POST` | `/ops-runs/{id}/fail` | failed (+ optional reopen) |
+| `POST` | `/ops-runs/expire-leases` | Reopen or expire stale claims |
+
+### Claim body
+
+```json
+{
+  "worker_id": "rein-aharness@railiance01",
+  "labels": ["automated"],
+  "labels_mode": "any",
+  "limit": 1,
+  "lease_seconds": 900
+}
+```
+
+- `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.
+
+### Complete / fail body
+
+```json
+{ "result": { "path": "briefs/…", "ok": true }, "worker_id": "…" }
+```
+
+```json
+{ "error": "llm timeout", "worker_id": "…", "reopen": false }
+```
+
+If `reopen: true` and `attempt < max_attempts` (env `OPS_RUN_MAX_ATTEMPTS`, default 3),
+state returns to `open`; else `failed`.
+
+## Emit path
+
+On `emit_tasks` (when `OPS_RUN_QUEUE_ENABLED` is truthy, **default true**):
+
+1. Insert `ops_run` with `state=open` (idempotent on unique key).  
+2. Dual-write existing IssueSink (`state-hub` progress by default).  
+3. Write `task_spawn_log` audit as today.
+
+Never requires Forgejo or issue-core for the claim path.
+
+## Auth
+
+- **Worker:** `ACTIVITY_CORE_WORKER_TOKEN` via `X-Worker-Token` or
+  `Authorization: Bearer` (same value accepted on claim/complete/fail/heartbeat).  
+- **Operator:** existing ops SSO / `ACTIVITY_CORE_OPERATOR_TOKEN` for list/status.  
+- **Local dev:** `ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS=1` when no tokens set.
+
+## Env
+
+| Variable | Default | Meaning |
+| -------- | ------- | ------- |
+| `OPS_RUN_QUEUE_ENABLED` | `true` | Create ops_run on emit |
+| `OPS_RUN_LEASE_SECONDS` | `900` | Default claim lease |
+| `OPS_RUN_MAX_ATTEMPTS` | `3` | Fail permanently after N claims |
+| `ACTIVITY_CORE_WORKER_TOKEN` | unset | Harness claim credential |
+
+## Consumer (rein-aharness)
+
+See REIN-A-0002. Claim loop → approach table → execute → complete + domain
+completion event (`fi_daily_brief`, etc.).
+
+## Not this queue
+
+| Concern | Home |
+| ------- | ---- |
+| Multi-day engineering tasks | Workplan files + State Hub |
+| External tracker tickets | issue-core → **Forgejo** (optional projection) |
+| Schedule truth | Temporal + activity definitions |
--- a/docs/recurring-automations-playbook.md
+++ b/docs/recurring-automations-playbook.md
@ -20,14 +20,15 @@ issue-core projection — never as the default ops claim queue.
 ┌─────────────────────────────────────────────────────────────┐
 │ 1. activity-core  — WHEN / WHAT / WHERE                     │
 │    Temporal schedule · context resolvers · rules            │
-│    → emits activity_task_spawn (state-hub sink by default)  │
+│    → INSERT ops_run (open) + dual-write activity_task_spawn │
+│    Claim API: POST /ops-runs/claim  (docs/ops-run-queue.md) │
 └────────────────────────────┬────────────────────────────────┘
                             │
 ┌────────────────────────────▼────────────────────────────────┐
 │ 2. rein-aharness (or domain executor) — DOES THE WORK       │
-│    llm-connect · repo checkout · commit · completion event  │
-│    Host timers on railiance are *interim* until poll/claim  │
-│    of spawns is fully wired (state-hub sink is not claimable)│
+│    claim ops_run · llm-connect · checkout · complete/fail   │
+│    Host timers on railiance are *interim* until REIN-A-0002 │
+│    claim loop is live (state-hub spawn alone is not claimable)│
 └────────────────────────────┬────────────────────────────────┘
                             │
 ┌────────────────────────────▼────────────────────────────────┐
@ -151,17 +152,22 @@ Do not flip production to `rest` globally.

 ## Operator daily habit

-1. Ops console: did automations run? (`/ops/automations/status`)  
-2. State Hub: `activity_task_spawn` + domain completion events  
-3. Domain repo: brief/artifact files + git history  
-4. If something is missing, **fix activity-core first** (schedule paused?
-   definition disabled? resolver due stuck? sink errors?) — do not invent a new cron  
+1. Ops console: did automations run? (`/ops/automations/status`) — includes
+   `ops_runs.counts` and `stuck_open_or_claimed` (SLA, default 1h)  
+2. Claim queue: `GET /ops-runs?state=open` (or failed) — harness backlog  
+3. State Hub: `activity_task_spawn` + domain completion events  
+4. Domain repo: brief/artifact files + git history  
+5. If something is missing, **fix activity-core first** (schedule paused?
+   definition disabled? resolver due stuck? open ops_run with no claim?) —
+   do not invent a new cron  

 ---

 ## Related

 - `INTENT.md` — when/what/where boundary  
+- `docs/adr/adr-005-ops-runs-vs-dev-work-records.md` — ops vs dev work  
+- `docs/ops-run-queue.md` — claim API  
 - `docs/task-emission-consumer-contract.md` — spawn payload + consumer duties  
 - `docs/runbook.md` — sync, trigger, ops UI  
 - `docs/adr/adr-002-definition-format.md` — definition files  
--- a/docs/runbook.md
+++ b/docs/runbook.md
@ -396,6 +396,43 @@ Default: **`ISSUE_SINK_TYPE=state-hub`** (ACTIVITY-WP-0022). See
 `review_required` on instructions is **metadata only** until a downstream
 review queue exists (issue-core / work-record lane) — see ACTIVITY-WP-0023-T09.

+## Ops run claim queue (ACTIVITY-WP-0026)
+
+Durable claimable instances for scheduled automation. Spec:
+`docs/ops-run-queue.md`. Architecture: ACT-ADR-005.
+
+```bash
+# Visibility (counts + stuck SLA in status)
+curl -sS "http://localhost:8010/ops/automations/status?since=today" \
+  | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get('ops_runs'))"
+
+# Open runs
+curl -sS "http://localhost:8010/ops-runs?state=open" | python3 -m json.tool
+
+# Claim (worker)
+curl -sS -X POST "http://localhost:8010/ops-runs/claim" \
+  -H "Content-Type: application/json" \
+  -H "X-Worker-Token: $ACTIVITY_CORE_WORKER_TOKEN" \
+  -d '{"worker_id":"rein-aharness@railiance01","labels":["automated"],"limit":1}'
+
+# Reopen stale leases
+curl -sS -X POST "http://localhost:8010/ops-runs/expire-leases" \
+  -H "X-Worker-Token: $ACTIVITY_CORE_WORKER_TOKEN"
+```
+
+| Env | Default | Meaning |
+| --- | --- | --- |
+| `OPS_RUN_QUEUE_ENABLED` | `true` | Insert ops_run on emit |
+| `OPS_RUN_LEASE_SECONDS` | `900` | Claim lease |
+| `OPS_RUN_MAX_ATTEMPTS` | `3` | Fail permanently after N claims |
+| `OPS_RUN_SLA_HOURS` | `1` | Stuck threshold in status |
+| `ACTIVITY_CORE_WORKER_TOKEN` | unset | Harness claim auth |
+
+**Railiance rollout (T07):** apply migration `0007`, set
+`OPS_RUN_QUEUE_ENABLED=true` on worker/api, smoke-trigger FI definition, confirm
+open row via `GET /ops-runs?state=open`. Keep host timers until REIN-A-0002 claim
+loop is proven.
+
 Example distinction from the June 2026 daily triage evidence:

 ```text
--- a/docs/task-emission-consumer-contract.md
+++ b/docs/task-emission-consumer-contract.md
@ -2,12 +2,34 @@

 **Audience:** agent-harness, per-repo Temporal workers, operators.  
 **Owners:** activity-core (producer), consumers (executors).  
-**Related:** ACTIVITY-WP-0022 (sink policy), ACTIVITY-WP-0023-T02 (executor gap).
+**Related:** ACTIVITY-WP-0022 (sink policy), ACTIVITY-WP-0023-T02 (executor gap),
+**ACT-ADR-005 / ACTIVITY-WP-0026** (ops_run claim queue).

 activity-core answers **when / what / where**. It does **not** execute work.
 Consumers must pick up emitted tasks and produce domain evidence.

-## Sink matrix (ACTIVITY-WP-0022)
+## Primary claim path: `ops_run` (ACTIVITY-WP-0026)
+
+For **scheduled / automation** work (FI daily brief, Binky rhythm, etc.), the
+authoritative claimable instance is an **`ops_run`** row in activity-core — not
+issue-core and not a Forgejo ticket.
+
+| Step | Call | Notes |
+| ---- | ---- | ----- |
+| List / poll | `GET /ops-runs?state=open` | Filter by labels via claim body |
+| Claim | `POST /ops-runs/claim` | `{ worker_id, labels?, limit?, lease_seconds? }` |
+| Heartbeat | `POST /ops-runs/{id}/heartbeat` | Extend lease during long runs |
+| Complete | `POST /ops-runs/{id}/complete` | `{ worker_id, result }` |
+| Fail | `POST /ops-runs/{id}/fail` | `{ worker_id, error, reopen? }` |
+
+Full field list, auth, and env: **`docs/ops-run-queue.md`**.  
+Consumer implementation (rein-aharness): **REIN-A-0002**.
+
+Emit dual-writes: `ops_run` (claim) + existing IssueSink progress (`state-hub`
+by default) + `task_spawn_log` audit. Do **not** treat `activity_task_spawn`
+progress as a claim queue — it is visibility only.
+
+## Sink matrix (ACTIVITY-WP-0022) — dual-write / projection

 | `ISSUE_SINK_TYPE` | Behaviour | When to use |
 | --- | --- | --- |
@ -15,12 +37,13 @@ Consumers must pick up emitted tasks and produce domain evidence.
 | **`null`** | Synthetic `null-*` refs in `task_spawn_log` only | Dry-run / contract review |
 | **`rest`** | POST issue-core `/issues/` (may project to Forgejo) | **Explicit opt-in** only when external tracker issues are intended and backend is healthy |

-Unset or unknown values fall back to **`state-hub`** (safe default).
+Unset or unknown values fall back to **`state-hub`** (safe default).  
+Ops claim does **not** require `rest` or issue-core.

 ## Payload: `activity_task_spawn` (State Hub)

 Produced by `StateHubProgressSink`. Consumers should treat `detail` as the
-authoritative task spec.
+**visibility** task spec; claim authority is `ops_run` when the queue is enabled.

 ```json
 {
@ -75,4 +98,7 @@ and still post the progress event — spawn without completion leaves `due=true`
 - Using `TaskExecutorWorkflow` in activity-core for real work (disabled by
  default; ACTIVITY-WP-0023-T08).
 - Global `ISSUE_SINK_TYPE=rest` for all definitions (reintroduces Forgejo spam).
- Treating `task_spawn_log` as task status authority.
+- Treating `task_spawn_log` or State Hub `activity_task_spawn` as claim authority
+  (use `POST /ops-runs/claim` — ACT-ADR-005).
+- Using issue-core or Forgejo as the ops automation claim queue.
+- Product references to **Gitea** — self-hosted forge is **Forgejo** only.
--- a/migrations/versions/0007_create_ops_runs.py
+++ b/migrations/versions/0007_create_ops_runs.py
@ -0,0 +1,95 @@
+"""create_ops_runs
+
+Revision ID: 0007
+Revises: 0006
+Create Date: 2026-08-03
+
+ACTIVITY-WP-0026 / ACT-ADR-005 — claimable ops run queue.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "0007"
+down_revision: Union[str, Sequence[str], None] = "0006"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "ops_runs",
+        sa.Column(
+            "id",
+            sa.UUID(),
+            nullable=False,
+            server_default=sa.text("gen_random_uuid()"),
+        ),
+        sa.Column("activity_definition_id", sa.UUID(), nullable=False),
+        sa.Column("idempotency_key", sa.Text(), nullable=False),
+        sa.Column("target_repo", sa.Text(), nullable=True),
+        sa.Column("title", sa.Text(), nullable=False),
+        sa.Column("description", sa.Text(), nullable=False, server_default=""),
+        sa.Column(
+            "labels",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=False,
+            server_default=sa.text("'[]'::jsonb"),
+        ),
+        sa.Column("priority", sa.Text(), nullable=False, server_default="medium"),
+        sa.Column("state", sa.Text(), nullable=False, server_default="open"),
+        sa.Column("claim_owner", sa.Text(), nullable=True),
+        sa.Column("lease_until", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("attempt", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("source_type", sa.Text(), nullable=False),
+        sa.Column("source_id", sa.Text(), nullable=False),
+        sa.Column("triggering_event_id", sa.Text(), nullable=False),
+        sa.Column("approach_hint", sa.Text(), nullable=True),
+        sa.Column(
+            "result",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=False,
+            server_default=sa.text("'{}'::jsonb"),
+        ),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.ForeignKeyConstraint(
+            ["activity_definition_id"],
+            ["activity_definitions.id"],
+            ondelete="RESTRICT",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("idempotency_key", name="uq_ops_runs_idempotency_key"),
+    )
+    op.create_index("idx_ops_runs_state", "ops_runs", ["state"])
+    op.create_index("idx_ops_runs_lease", "ops_runs", ["lease_until"])
+    op.create_index(
+        "idx_ops_runs_activity",
+        "ops_runs",
+        ["activity_definition_id"],
+    )
+    op.create_index(
+        "idx_ops_runs_trigger",
+        "ops_runs",
+        ["triggering_event_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("idx_ops_runs_trigger", table_name="ops_runs")
+    op.drop_index("idx_ops_runs_activity", table_name="ops_runs")
+    op.drop_index("idx_ops_runs_lease", table_name="ops_runs")
+    op.drop_index("idx_ops_runs_state", table_name="ops_runs")
+    op.drop_table("ops_runs")
--- a/src/activity_core/activities.py
+++ b/src/activity_core/activities.py
@ -26,6 +26,7 @@ from activity_core.db import make_engine
 from activity_core.issue_sink import get_issue_sink
 from activity_core.orm import ActivityDefinition as ActivityDefinitionRow
 from activity_core.orm import ActivityRun, TaskInstance, TaskSpawnLog
+from activity_core.ops_run_queue import create_ops_run_from_spec
 from activity_core.llm_client import get_llm_client
 from activity_core.models import InstructionDef
 from activity_core.ops_evidence_sinks import persist_ops_inventory_evidence
@ -470,6 +471,25 @@ async def emit_tasks(payload: dict) -> list[str]:
                    activity_definition_id=activity_id,
                )
                try:
+                    # ACTIVITY-WP-0026: claimable ops_run (primary for harness)
+                    try:
+                        ops_id = await create_ops_run_from_spec(
+                            session,
+                            spec,
+                            approach_hint=spec_dict.get("approach_hint"),
+                        )
+                        if ops_id is not None:
+                            activity.logger.info(
+                                "emit_tasks: ops_run created id=%s key=%s:%s",
+                                ops_id,
+                                spec.source_id,
+                                triggering_event_id,
+                            )
+                    except Exception as ops_exc:
+                        activity.logger.warning(
+                            "emit_tasks: ops_run insert failed — %s", ops_exc
+                        )
+
                    ref = sink.emit(spec)
                    refs.append(ref.external_id)

--- a/src/activity_core/api.py
+++ b/src/activity_core/api.py
@ -41,6 +41,7 @@ from temporalio.client import Client

 from activity_core.models import ActivityDefinition, CronTriggerConfig
 from activity_core.ops_api import bind_ops_deps, router as ops_router
+from activity_core.ops_runs_api import bind_ops_runs_deps, router as ops_runs_router
 from activity_core.orm import ActivityDefinition as ActivityDefinitionRow, EventType as EventTypeRow
 from activity_core.schedule_manager import delete_schedule, upsert_schedule
 from activity_core.sync_service import run_sync
@ -73,6 +74,7 @@ async def lifespan(app: FastAPI):  # type: ignore[type-arg]
    _session_factory = async_sessionmaker(engine, expire_on_commit=False)
    _temporal_client = await Client.connect(TEMPORAL_HOST, namespace=TEMPORAL_NAMESPACE)
    bind_ops_deps(_get_db, _get_temporal)
+    bind_ops_runs_deps(_get_db)

    yield

@ -82,6 +84,7 @@ async def lifespan(app: FastAPI):  # type: ignore[type-arg]
 app = FastAPI(title="activity-core API", lifespan=lifespan)
 app.include_router(webhook_router)
 app.include_router(ops_router)
+app.include_router(ops_runs_router)


 def _get_db() -> async_sessionmaker[AsyncSession]:
--- a/src/activity_core/ops_api.py
+++ b/src/activity_core/ops_api.py
@ -90,7 +90,7 @@ async def automations_status(
    timezone_name: str = Query(default="Europe/Berlin", alias="timezone"),
    activity_id: str | None = Query(default=None),
 ) -> dict[str, Any]:
-    return await ops_status(
+    report = await ops_status(
        since=since,
        until=until,
        timezone_name=timezone_name,
@ -100,6 +100,37 @@ async def automations_status(
        state_hub_url=os.environ.get("STATE_HUB_URL"),
        activity_id=activity_id,
    )
+    # ACTIVITY-WP-0026 T05: ops_run claim-queue visibility + SLA signals
+    try:
+        from datetime import timedelta
+
+        from sqlalchemy import and_, func, select
+
+        from activity_core.orm import OpsRun
+        from activity_core.ops_run_queue import ops_run_counts
+
+        sla_hours = float(os.environ.get("OPS_RUN_SLA_HOURS", "1") or "1")
+        now = datetime.now(timezone.utc)
+        sla_cutoff = now - timedelta(hours=max(0.1, sla_hours))
+        Session = _db()
+        async with Session() as session:
+            counts = await ops_run_counts(session)
+            stuck_stmt = select(func.count()).where(
+                and_(
+                    OpsRun.state.in_(("open", "claimed")),
+                    OpsRun.created_at < sla_cutoff,
+                )
+            )
+            stuck = int((await session.execute(stuck_stmt)).scalar_one() or 0)
+        report["ops_runs"] = {
+            "counts": counts,
+            "stuck_open_or_claimed": stuck,
+            "sla_hours": sla_hours,
+            "list_url": "/ops-runs?state=open",
+        }
+    except Exception as exc:  # table may not exist pre-migration
+        report["ops_runs"] = {"error": str(exc), "counts": {}}
+    return report


@router.get("/automations/{definition_id}")
--- a/src/activity_core/ops_run_queue.py
+++ b/src/activity_core/ops_run_queue.py
@ -0,0 +1,299 @@
+"""Ops run claim queue service (ACTIVITY-WP-0026 / ACT-ADR-005)."""
+
+from __future__ import annotations
+
+import os
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from sqlalchemy import Select, and_, func, select, update
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from activity_core.orm import OpsRun
+from activity_core.rules.models import TaskSpec
+
+OPS_RUN_STATES = frozenset(
+    {"open", "claimed", "succeeded", "failed", "expired"}
+)
+
+
+def ops_run_queue_enabled() -> bool:
+    raw = (os.environ.get("OPS_RUN_QUEUE_ENABLED") or "true").strip().lower()
+    return raw in {"1", "true", "yes", "on", ""}
+
+
+def default_lease_seconds() -> int:
+    try:
+        return max(30, int(os.environ.get("OPS_RUN_LEASE_SECONDS", "900")))
+    except ValueError:
+        return 900
+
+
+def max_attempts() -> int:
+    try:
+        return max(1, int(os.environ.get("OPS_RUN_MAX_ATTEMPTS", "3")))
+    except ValueError:
+        return 3
+
+
+def build_idempotency_key(spec: TaskSpec) -> str:
+    """Stable key for one emit; prevents double open rows on redelivery."""
+    return (
+        f"{spec.activity_definition_id}:{spec.source_id}:{spec.triggering_event_id}"
+    )
+
+
+def ops_run_to_dict(row: OpsRun) -> dict[str, Any]:
+    return {
+        "id": str(row.id),
+        "activity_definition_id": str(row.activity_definition_id),
+        "idempotency_key": row.idempotency_key,
+        "target_repo": row.target_repo,
+        "title": row.title,
+        "description": row.description,
+        "labels": list(row.labels or []),
+        "priority": row.priority,
+        "state": row.state,
+        "claim_owner": row.claim_owner,
+        "lease_until": row.lease_until.isoformat() if row.lease_until else None,
+        "attempt": row.attempt,
+        "source_type": row.source_type,
+        "source_id": row.source_id,
+        "triggering_event_id": row.triggering_event_id,
+        "approach_hint": row.approach_hint,
+        "result": dict(row.result or {}),
+        "created_at": row.created_at.isoformat() if row.created_at else None,
+        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
+    }
+
+
+async def create_ops_run_from_spec(
+    session: AsyncSession,
+    spec: TaskSpec,
+    *,
+    approach_hint: str | None = None,
+) -> uuid.UUID | None:
+    """Insert ops_run if queue enabled; return id or None if disabled/duplicate."""
+    if not ops_run_queue_enabled():
+        return None
+    if not spec.activity_definition_id:
+        return None
+
+    try:
+        def_id = uuid.UUID(str(spec.activity_definition_id))
+    except ValueError:
+        return None
+
+    key = build_idempotency_key(spec)
+    now = datetime.now(timezone.utc)
+    stmt = (
+        pg_insert(OpsRun)
+        .values(
+            id=uuid.uuid4(),
+            activity_definition_id=def_id,
+            idempotency_key=key,
+            target_repo=spec.target_repo,
+            title=spec.title or "(untitled)",
+            description=spec.description or "",
+            labels=list(spec.labels or []),
+            priority=spec.priority or "medium",
+            state="open",
+            attempt=0,
+            source_type=spec.source_type or "rule",
+            source_id=spec.source_id or "",
+            triggering_event_id=spec.triggering_event_id or "",
+            approach_hint=approach_hint,
+            result={},
+            created_at=now,
+            updated_at=now,
+        )
+        .on_conflict_do_nothing(index_elements=["idempotency_key"])
+        .returning(OpsRun.id)
+    )
+    result = await session.execute(stmt)
+    row_id = result.scalar_one_or_none()
+    return row_id
+
+
+async def reopen_stale_claims(session: AsyncSession) -> int:
+    """Return claimed rows with expired leases to open."""
+    now = datetime.now(timezone.utc)
+    stmt = (
+        update(OpsRun)
+        .where(
+            and_(
+                OpsRun.state == "claimed",
+                OpsRun.lease_until.is_not(None),
+                OpsRun.lease_until < now,
+            )
+        )
+        .values(
+            state="open",
+            claim_owner=None,
+            lease_until=None,
+            updated_at=now,
+        )
+    )
+    result = await session.execute(stmt)
+    return int(result.rowcount or 0)
+
+
+async def claim_ops_runs(
+    session: AsyncSession,
+    *,
+    worker_id: str,
+    labels: list[str] | None = None,
+    labels_mode: str = "any",
+    limit: int = 1,
+    lease_seconds: int | None = None,
+) -> list[OpsRun]:
+    """Claim up to ``limit`` open ops_runs for worker_id."""
+    if not worker_id.strip():
+        raise ValueError("worker_id is required")
+    limit = max(1, min(limit, 20))
+    lease_seconds = lease_seconds or default_lease_seconds()
+    now = datetime.now(timezone.utc)
+    lease_until = now + timedelta(seconds=lease_seconds)
+
+    await reopen_stale_claims(session)
+
+    # Candidate ids with SKIP LOCKED
+    filters = [OpsRun.state == "open"]
+    # labels filter applied in Python after fetch for JSONB portability,
+    # but prefer SQL when possible — use jsonb containment for "all"
+    stmt: Select[tuple[uuid.UUID]] = (
+        select(OpsRun.id)
+        .where(*filters)
+        .order_by(OpsRun.created_at.asc())
+        .limit(limit * 5)  # over-fetch if label filter drops rows
+        .with_for_update(skip_locked=True)
+    )
+    result = await session.execute(stmt)
+    candidate_ids = list(result.scalars().all())
+    if not candidate_ids:
+        return []
+
+    claimed: list[OpsRun] = []
+    for run_id in candidate_ids:
+        if len(claimed) >= limit:
+            break
+        row = await session.get(OpsRun, run_id)
+        if row is None or row.state != "open":
+            continue
+        if labels:
+            row_labels = {str(x) for x in (row.labels or [])}
+            want = {str(x) for x in labels}
+            if labels_mode == "all":
+                if not want.issubset(row_labels):
+                    continue
+            else:  # any
+                if not (want & row_labels):
+                    continue
+        row.state = "claimed"
+        row.claim_owner = worker_id.strip()
+        row.lease_until = lease_until
+        row.attempt = int(row.attempt or 0) + 1
+        row.updated_at = now
+        claimed.append(row)
+
+    return claimed
+
+
+async def heartbeat_ops_run(
+    session: AsyncSession,
+    run_id: uuid.UUID,
+    *,
+    worker_id: str,
+    lease_seconds: int | None = None,
+) -> OpsRun | None:
+    row = await session.get(OpsRun, run_id)
+    if row is None:
+        return None
+    if row.state != "claimed" or row.claim_owner != worker_id:
+        return None
+    lease_seconds = lease_seconds or default_lease_seconds()
+    now = datetime.now(timezone.utc)
+    row.lease_until = now + timedelta(seconds=lease_seconds)
+    row.updated_at = now
+    return row
+
+
+async def complete_ops_run(
+    session: AsyncSession,
+    run_id: uuid.UUID,
+    *,
+    worker_id: str,
+    result: dict[str, Any] | None = None,
+) -> OpsRun | None:
+    row = await session.get(OpsRun, run_id)
+    if row is None:
+        return None
+    if row.state != "claimed" or row.claim_owner != worker_id:
+        return None
+    now = datetime.now(timezone.utc)
+    row.state = "succeeded"
+    row.lease_until = None
+    row.result = dict(result or {})
+    row.updated_at = now
+    return row
+
+
+async def fail_ops_run(
+    session: AsyncSession,
+    run_id: uuid.UUID,
+    *,
+    worker_id: str,
+    error: str = "",
+    reopen: bool = False,
+    result: dict[str, Any] | None = None,
+) -> OpsRun | None:
+    row = await session.get(OpsRun, run_id)
+    if row is None:
+        return None
+    if row.state != "claimed" or row.claim_owner != worker_id:
+        return None
+    now = datetime.now(timezone.utc)
+    payload = dict(result or {})
+    if error:
+        payload["error"] = error[:2000]
+    row.result = payload
+    row.updated_at = now
+    if reopen and int(row.attempt or 0) < max_attempts():
+        row.state = "open"
+        row.claim_owner = None
+        row.lease_until = None
+    else:
+        row.state = "failed"
+        row.lease_until = None
+    return row
+
+
+async def list_ops_runs(
+    session: AsyncSession,
+    *,
+    state: str | None = None,
+    activity_definition_id: uuid.UUID | None = None,
+    since: datetime | None = None,
+    limit: int = 50,
+) -> list[OpsRun]:
+    limit = max(1, min(limit, 200))
+    stmt = select(OpsRun).order_by(OpsRun.created_at.desc()).limit(limit)
+    if state:
+        stmt = stmt.where(OpsRun.state == state)
+    if activity_definition_id:
+        stmt = stmt.where(OpsRun.activity_definition_id == activity_definition_id)
+    if since:
+        stmt = stmt.where(OpsRun.created_at >= since)
+    result = await session.execute(stmt)
+    return list(result.scalars().all())
+
+
+async def ops_run_counts(session: AsyncSession) -> dict[str, int]:
+    stmt = select(OpsRun.state, func.count()).group_by(OpsRun.state)
+    result = await session.execute(stmt)
+    counts = {s: 0 for s in OPS_RUN_STATES}
+    for state, n in result.all():
+        counts[str(state)] = int(n)
+    return counts
--- a/src/activity_core/ops_runs_api.py
+++ b/src/activity_core/ops_runs_api.py
@ -0,0 +1,339 @@
+"""REST API for ops_run claim queue (ACTIVITY-WP-0026)."""
+
+from __future__ import annotations
+
+import os
+import uuid
+from datetime import datetime
+from typing import Any, Callable
+
+from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request
+from pydantic import BaseModel, Field
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from activity_core.ops_auth import (
+    allow_unauth_mutations,
+    extract_operator_token,
+    extract_sso_principal,
+    operator_token_configured,
+)
+from activity_core.ops_run_queue import (
+    claim_ops_runs,
+    complete_ops_run,
+    default_lease_seconds,
+    fail_ops_run,
+    heartbeat_ops_run,
+    list_ops_runs,
+    ops_run_counts,
+    ops_run_to_dict,
+    reopen_stale_claims,
+)
+
+router = APIRouter(prefix="/ops-runs", tags=["ops-runs"])
+
+_get_db: Callable[[], async_sessionmaker[AsyncSession]] | None = None
+
+WORKER_TOKEN_ENV = "ACTIVITY_CORE_WORKER_TOKEN"
+
+
+def bind_ops_runs_deps(
+    get_db: Callable[[], async_sessionmaker[AsyncSession]],
+) -> None:
+    global _get_db
+    _get_db = get_db
+
+
+def _db() -> async_sessionmaker[AsyncSession]:
+    if _get_db is None:
+        raise RuntimeError("ops_runs API not bound")
+    return _get_db()
+
+
+def _worker_token_configured() -> bool:
+    return bool((os.environ.get(WORKER_TOKEN_ENV) or "").strip())
+
+
+def _extract_worker_token(
+    *,
+    x_worker_token: str | None,
+    authorization: str | None,
+) -> str | None:
+    if x_worker_token and x_worker_token.strip():
+        return x_worker_token.strip()
+    if authorization and authorization.lower().startswith("bearer "):
+        return authorization[7:].strip() or None
+    return None
+
+
+def require_worker_or_operator(
+    request: Request,
+    *,
+    x_worker_token: str | None = None,
+    x_operator_token: str | None = None,
+    authorization: str | None = None,
+) -> str:
+    """Return principal string for worker or operator."""
+    worker_tok = _extract_worker_token(
+        x_worker_token=x_worker_token, authorization=authorization
+    )
+    expected_worker = (os.environ.get(WORKER_TOKEN_ENV) or "").strip()
+    if expected_worker and worker_tok and worker_tok == expected_worker:
+        return f"worker:{worker_tok[:8]}"
+
+    sso = extract_sso_principal(request)
+    if sso:
+        return f"sso:{sso}"
+
+    op_tok = extract_operator_token(
+        x_operator_token=x_operator_token, authorization=authorization
+    )
+    expected_op = (os.environ.get("ACTIVITY_CORE_OPERATOR_TOKEN") or "").strip()
+    if expected_op and op_tok and op_tok == expected_op:
+        return "operator:token"
+
+    if not _worker_token_configured() and not operator_token_configured():
+        if allow_unauth_mutations():
+            return "dev:unauth"
+        # Dev convenience when no tokens configured at all
+        return "dev:open"
+
+    raise HTTPException(status_code=401, detail="worker or operator auth required")
+
+
+class ClaimBody(BaseModel):
+    worker_id: str = Field(..., min_length=1, max_length=256)
+    labels: list[str] | None = None
+    labels_mode: str = Field(default="any", pattern="^(any|all)$")
+    limit: int = Field(default=1, ge=1, le=20)
+    lease_seconds: int | None = Field(default=None, ge=30, le=86400)
+
+
+class HeartbeatBody(BaseModel):
+    worker_id: str = Field(..., min_length=1, max_length=256)
+    lease_seconds: int | None = Field(default=None, ge=30, le=86400)
+
+
+class CompleteBody(BaseModel):
+    worker_id: str = Field(..., min_length=1, max_length=256)
+    result: dict[str, Any] | None = None
+
+
+class FailBody(BaseModel):
+    worker_id: str = Field(..., min_length=1, max_length=256)
+    error: str = ""
+    reopen: bool = False
+    result: dict[str, Any] | None = None
+
+
+@router.get("")
+async def get_ops_runs(
+    request: Request,
+    state: str | None = Query(default=None),
+    activity_definition_id: str | None = Query(default=None),
+    since: datetime | None = Query(default=None),
+    limit: int = Query(default=50, ge=1, le=200),
+    x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
+    x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
+    authorization: str | None = Header(default=None),
+) -> dict[str, Any]:
+    require_worker_or_operator(
+        request,
+        x_worker_token=x_worker_token,
+        x_operator_token=x_operator_token,
+        authorization=authorization,
+    )
+    def_id = None
+    if activity_definition_id:
+        try:
+            def_id = uuid.UUID(activity_definition_id)
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail="invalid activity_definition_id") from exc
+
+    Session = _db()
+    async with Session() as session:
+        rows = await list_ops_runs(
+            session,
+            state=state,
+            activity_definition_id=def_id,
+            since=since,
+            limit=limit,
+        )
+        counts = await ops_run_counts(session)
+        await session.commit()
+    return {
+        "items": [ops_run_to_dict(r) for r in rows],
+        "counts": counts,
+    }
+
+
+@router.get("/{run_id}")
+async def get_ops_run(
+    run_id: uuid.UUID,
+    request: Request,
+    x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
+    x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
+    authorization: str | None = Header(default=None),
+) -> dict[str, Any]:
+    require_worker_or_operator(
+        request,
+        x_worker_token=x_worker_token,
+        x_operator_token=x_operator_token,
+        authorization=authorization,
+    )
+    Session = _db()
+    async with Session() as session:
+        from activity_core.orm import OpsRun
+
+        row = await session.get(OpsRun, run_id)
+        if row is None:
+            raise HTTPException(status_code=404, detail="ops_run not found")
+        return ops_run_to_dict(row)
+
+
+@router.post("/claim")
+async def post_claim(
+    body: ClaimBody,
+    request: Request,
+    x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
+    x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
+    authorization: str | None = Header(default=None),
+) -> dict[str, Any]:
+    require_worker_or_operator(
+        request,
+        x_worker_token=x_worker_token,
+        x_operator_token=x_operator_token,
+        authorization=authorization,
+    )
+    Session = _db()
+    async with Session() as session:
+        async with session.begin():
+            claimed = await claim_ops_runs(
+                session,
+                worker_id=body.worker_id,
+                labels=body.labels,
+                labels_mode=body.labels_mode,
+                limit=body.limit,
+                lease_seconds=body.lease_seconds,
+            )
+        return {
+            "items": [ops_run_to_dict(r) for r in claimed],
+            "lease_seconds": body.lease_seconds or default_lease_seconds(),
+        }
+
+
+@router.post("/{run_id}/heartbeat")
+async def post_heartbeat(
+    run_id: uuid.UUID,
+    body: HeartbeatBody,
+    request: Request,
+    x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
+    x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
+    authorization: str | None = Header(default=None),
+) -> dict[str, Any]:
+    require_worker_or_operator(
+        request,
+        x_worker_token=x_worker_token,
+        x_operator_token=x_operator_token,
+        authorization=authorization,
+    )
+    Session = _db()
+    async with Session() as session:
+        async with session.begin():
+            row = await heartbeat_ops_run(
+                session,
+                run_id,
+                worker_id=body.worker_id,
+                lease_seconds=body.lease_seconds,
+            )
+            if row is None:
+                raise HTTPException(
+                    status_code=409,
+                    detail="not claimed by this worker or not found",
+                )
+            return ops_run_to_dict(row)
+
+
+@router.post("/{run_id}/complete")
+async def post_complete(
+    run_id: uuid.UUID,
+    body: CompleteBody,
+    request: Request,
+    x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
+    x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
+    authorization: str | None = Header(default=None),
+) -> dict[str, Any]:
+    require_worker_or_operator(
+        request,
+        x_worker_token=x_worker_token,
+        x_operator_token=x_operator_token,
+        authorization=authorization,
+    )
+    Session = _db()
+    async with Session() as session:
+        async with session.begin():
+            row = await complete_ops_run(
+                session,
+                run_id,
+                worker_id=body.worker_id,
+                result=body.result,
+            )
+            if row is None:
+                raise HTTPException(
+                    status_code=409,
+                    detail="not claimed by this worker or not found",
+                )
+            return ops_run_to_dict(row)
+
+
+@router.post("/{run_id}/fail")
+async def post_fail(
+    run_id: uuid.UUID,
+    body: FailBody,
+    request: Request,
+    x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
+    x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
+    authorization: str | None = Header(default=None),
+) -> dict[str, Any]:
+    require_worker_or_operator(
+        request,
+        x_worker_token=x_worker_token,
+        x_operator_token=x_operator_token,
+        authorization=authorization,
+    )
+    Session = _db()
+    async with Session() as session:
+        async with session.begin():
+            row = await fail_ops_run(
+                session,
+                run_id,
+                worker_id=body.worker_id,
+                error=body.error,
+                reopen=body.reopen,
+                result=body.result,
+            )
+            if row is None:
+                raise HTTPException(
+                    status_code=409,
+                    detail="not claimed by this worker or not found",
+                )
+            return ops_run_to_dict(row)
+
+
+@router.post("/expire-leases")
+async def post_expire_leases(
+    request: Request,
+    x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"),
+    x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"),
+    authorization: str | None = Header(default=None),
+) -> dict[str, Any]:
+    require_worker_or_operator(
+        request,
+        x_worker_token=x_worker_token,
+        x_operator_token=x_operator_token,
+        authorization=authorization,
+    )
+    Session = _db()
+    async with Session() as session:
+        async with session.begin():
+            n = await reopen_stale_claims(session)
+        return {"reopened": n}
--- a/src/activity_core/orm.py
+++ b/src/activity_core/orm.py
@ -138,3 +138,45 @@ class TaskInstance(Base):
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
+
+
+class OpsRun(Base):
+    """Claimable automation run instance (ACT-ADR-005 / ACTIVITY-WP-0026)."""
+
+    __tablename__ = "ops_runs"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    activity_definition_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("activity_definitions.id", ondelete="RESTRICT"),
+        nullable=False,
+        index=True,
+    )
+    idempotency_key: Mapped[str] = mapped_column(Text, nullable=False, unique=True)
+    target_repo: Mapped[str | None] = mapped_column(Text, nullable=True)
+    title: Mapped[str] = mapped_column(Text, nullable=False)
+    description: Mapped[str] = mapped_column(Text, nullable=False, default="")
+    labels: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
+    priority: Mapped[str] = mapped_column(Text, nullable=False, default="medium")
+    state: Mapped[str] = mapped_column(Text, nullable=False, default="open", index=True)
+    claim_owner: Mapped[str | None] = mapped_column(Text, nullable=True)
+    lease_until: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, index=True
+    )
+    attempt: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    source_type: Mapped[str] = mapped_column(Text, nullable=False)
+    source_id: Mapped[str] = mapped_column(Text, nullable=False)
+    triggering_event_id: Mapped[str] = mapped_column(Text, nullable=False, index=True)
+    approach_hint: Mapped[str | None] = mapped_column(Text, nullable=True)
+    result: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        server_default=func.now(),
+        onupdate=func.now(),
+    )
--- a/tests/test_ops_run_queue.py
+++ b/tests/test_ops_run_queue.py
@ -0,0 +1,223 @@
+"""Tests for ops_run claim queue (ACTIVITY-WP-0026)."""
+
+from __future__ import annotations
+
+import os
+import uuid
+from datetime import datetime, timedelta, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from activity_core.ops_run_queue import (
+    build_idempotency_key,
+    max_attempts,
+    ops_run_queue_enabled,
+    ops_run_to_dict,
+)
+from activity_core.rules.models import TaskSpec
+
+
+def test_ops_run_queue_enabled_default(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("OPS_RUN_QUEUE_ENABLED", raising=False)
+    assert ops_run_queue_enabled() is True
+    monkeypatch.setenv("OPS_RUN_QUEUE_ENABLED", "false")
+    assert ops_run_queue_enabled() is False
+    monkeypatch.setenv("OPS_RUN_QUEUE_ENABLED", "1")
+    assert ops_run_queue_enabled() is True
+
+
+def test_build_idempotency_key() -> None:
+    spec = TaskSpec(
+        title="t",
+        activity_definition_id="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
+        source_id="emit-fi",
+        triggering_event_id="wf-1",
+    )
+    assert build_idempotency_key(spec) == (
+        "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee:emit-fi:wf-1"
+    )
+
+
+def test_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("OPS_RUN_MAX_ATTEMPTS", raising=False)
+    assert max_attempts() == 3
+    monkeypatch.setenv("OPS_RUN_MAX_ATTEMPTS", "5")
+    assert max_attempts() == 5
+
+
+def test_ops_run_to_dict_shape() -> None:
+    now = datetime.now(timezone.utc)
+    row = MagicMock()
+    row.id = uuid.uuid4()
+    row.activity_definition_id = uuid.uuid4()
+    row.idempotency_key = "k"
+    row.target_repo = "freedom-intelligence"
+    row.title = "FI brief"
+    row.description = "d"
+    row.labels = ["automated", "research-brief"]
+    row.priority = "medium"
+    row.state = "open"
+    row.claim_owner = None
+    row.lease_until = None
+    row.attempt = 0
+    row.source_type = "rule"
+    row.source_id = "emit"
+    row.triggering_event_id = "e1"
+    row.approach_hint = None
+    row.result = {}
+    row.created_at = now
+    row.updated_at = now
+    d = ops_run_to_dict(row)
+    assert d["target_repo"] == "freedom-intelligence"
+    assert d["state"] == "open"
+    assert "research-brief" in d["labels"]
+    assert d["created_at"]
+
+
+@pytest.mark.asyncio
+async def test_create_ops_run_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
+    from activity_core.ops_run_queue import create_ops_run_from_spec
+
+    monkeypatch.setenv("OPS_RUN_QUEUE_ENABLED", "false")
+    session = AsyncMock()
+    spec = TaskSpec(
+        title="t",
+        activity_definition_id=str(uuid.uuid4()),
+        source_id="s",
+        triggering_event_id="e",
+    )
+    assert await create_ops_run_from_spec(session, spec) is None
+    session.execute.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_create_ops_run_inserts(monkeypatch: pytest.MonkeyPatch) -> None:
+    from activity_core.ops_run_queue import create_ops_run_from_spec
+
+    monkeypatch.setenv("OPS_RUN_QUEUE_ENABLED", "true")
+    run_id = uuid.uuid4()
+    result = MagicMock()
+    result.scalar_one_or_none.return_value = run_id
+    session = AsyncMock()
+    session.execute = AsyncMock(return_value=result)
+    spec = TaskSpec(
+        title="FI daily",
+        target_repo="freedom-intelligence",
+        labels=["automated", "research-brief"],
+        activity_definition_id=str(uuid.uuid4()),
+        source_id="emit-fi-daily-brief-task",
+        triggering_event_id="manual-1",
+    )
+    got = await create_ops_run_from_spec(session, spec)
+    assert got == run_id
+    session.execute.assert_awaited()
+
+
+@pytest.mark.asyncio
+async def test_claim_and_complete_roundtrip(monkeypatch: pytest.MonkeyPatch) -> None:
+    """In-memory style: claim filters labels and complete transitions state."""
+    from activity_core import ops_run_queue as oq
+
+    now = datetime.now(timezone.utc)
+    open_run = MagicMock()
+    open_run.id = uuid.uuid4()
+    open_run.state = "open"
+    open_run.labels = ["automated", "research-brief"]
+    open_run.attempt = 0
+    open_run.claim_owner = None
+    open_run.lease_until = None
+    open_run.result = {}
+
+    session = AsyncMock()
+
+    reopen_result = MagicMock()
+    reopen_result.rowcount = 0
+
+    id_result = MagicMock()
+    id_result.scalars.return_value.all.return_value = [open_run.id]
+
+    async def execute_side_effect(stmt, *args, **kwargs):
+        # reopen_stale_claims uses Update; claim select uses Select (may contain FOR UPDATE)
+        name = type(stmt).__name__.lower()
+        if name == "update":
+            return reopen_result
+        return id_result
+
+    session.execute = AsyncMock(side_effect=execute_side_effect)
+    session.get = AsyncMock(return_value=open_run)
+
+    claimed = await oq.claim_ops_runs(
+        session,
+        worker_id="worker-1",
+        labels=["research-brief"],
+        limit=1,
+        lease_seconds=60,
+    )
+    assert len(claimed) == 1
+    assert open_run.state == "claimed"
+    assert open_run.claim_owner == "worker-1"
+    assert open_run.attempt == 1
+    assert open_run.lease_until is not None
+    assert open_run.lease_until > now
+
+    done = await oq.complete_ops_run(
+        session,
+        open_run.id,
+        worker_id="worker-1",
+        result={"path": "briefs/x.md"},
+    )
+    assert done is not None
+    assert open_run.state == "succeeded"
+    assert open_run.result["path"] == "briefs/x.md"
+
+
+@pytest.mark.asyncio
+async def test_fail_reopen_under_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
+    from activity_core import ops_run_queue as oq
+
+    monkeypatch.setenv("OPS_RUN_MAX_ATTEMPTS", "3")
+    row = MagicMock()
+    row.id = uuid.uuid4()
+    row.state = "claimed"
+    row.claim_owner = "w1"
+    row.attempt = 1
+    row.result = {}
+    session = AsyncMock()
+    session.get = AsyncMock(return_value=row)
+
+    out = await oq.fail_ops_run(
+        session, row.id, worker_id="w1", error="timeout", reopen=True
+    )
+    assert out is not None
+    assert row.state == "open"
+    assert row.claim_owner is None
+
+
+@pytest.mark.asyncio
+async def test_fail_permanent_at_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
+    from activity_core import ops_run_queue as oq
+
+    monkeypatch.setenv("OPS_RUN_MAX_ATTEMPTS", "3")
+    row = MagicMock()
+    row.id = uuid.uuid4()
+    row.state = "claimed"
+    row.claim_owner = "w1"
+    row.attempt = 3
+    row.result = {}
+    session = AsyncMock()
+    session.get = AsyncMock(return_value=row)
+
+    out = await oq.fail_ops_run(
+        session, row.id, worker_id="w1", error="timeout", reopen=True
+    )
+    assert out is not None
+    assert row.state == "failed"
+
+
+def test_label_filter_any_vs_all() -> None:
+    """Document labels_mode semantics used by claim_ops_runs."""
+    row_labels = {"automated", "research-brief"}
+    want = {"research-brief", "missing"}
+    assert bool(want & row_labels)  # any
+    assert not want.issubset(row_labels)  # all
--- a/workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md
+++ b/workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md
@ -4,7 +4,7 @@ type: workplan
 title: "Ops run claim queue — durable claimable work for scheduled automation"
 domain: infotech
 repo: activity-core
-status: ready
+status: in_progress
 owner: grok
 topic_slug: activity-core
 priority: high
@ -54,7 +54,7 @@ Temporal schedule / trigger

 ```task
 id: ACTIVITY-WP-0026-T01
-status: todo
+status: done
 priority: high
 state_hub_task_id: "d2f92ac2-d900-43d8-83ae-62e77d7467d4"
 ```
@ -97,7 +97,7 @@ service credential for harness, SSO for humans later).

 ```task
 id: ACTIVITY-WP-0026-T02
-status: todo
+status: done
 priority: high
 state_hub_task_id: "65d0d28f-0be6-4ae5-9e12-a051ccebbb36"
 ```
@ -113,7 +113,7 @@ Alembic migration for `ops_runs` (+ indexes on `state`, `lease_until`,

 ```task
 id: ACTIVITY-WP-0026-T03
-status: todo
+status: done
 priority: high
 state_hub_task_id: "1615936e-e0a4-4221-8b93-ec7938bd523b"
 ```
@ -137,7 +137,7 @@ Binky-style definitions; tests with mocked sink.

 ```task
 id: ACTIVITY-WP-0026-T04
-status: todo
+status: done
 priority: high
 state_hub_task_id: "57cdd0ce-2d54-4148-89e2-9d89c33d7f8a"
 ```
@ -154,7 +154,7 @@ lease TTL (config: e.g. 15m). Expire job or on-claim sweep: `claimed` past

 ```task
 id: ACTIVITY-WP-0026-T05
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "08f53f8b-6a17-4a21-9639-88f6924a7c1a"
 ```
@ -171,7 +171,7 @@ state_hub_task_id: "08f53f8b-6a17-4a21-9639-88f6924a7c1a"

 ```task
 id: ACTIVITY-WP-0026-T06
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "f8ca5ab8-3343-4edb-9ae8-fdbc90e17aae"
 ```
@ -205,11 +205,13 @@ state_hub_task_id: "b1ce221b-edb2-42ed-91a1-e9f144a58d1e"

 ## Acceptance

- [ ] ACT-ADR-005 referenced in SCOPE gaps (G2 disposition → this WP)  
- [ ] ops_run durable + claimable without Forgejo  
- [ ] Emit dual-write progress for transition  
- [ ] Docs use Forgejo-only language for self-hosted forge  
- [ ] REIN-A-0002 unblocked  
+- [x] ACT-ADR-005 referenced in SCOPE gaps (G2 disposition → this WP)  
+- [x] ops_run durable + claimable without Forgejo  
+- [x] Emit dual-write progress for transition  
+- [x] Docs use Forgejo-only language for self-hosted forge  
+- [x] REIN-A-0002 unblocked (contract in docs/ops-run-queue.md + consumer contract)  
+
+**T07 remaining:** railiance migrate + flag + smoke (deploy, not code).

 ## Out of scope