From 15eb3a2066c5724073808b5da22a7e908346480a Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 3 Aug 2026 19:22:50 +0200 Subject: [PATCH] =?UTF-8?q?Implement=20ACTIVITY-WP-0026=20ops=5Frun=20clai?= =?UTF-8?q?m=20queue=20(T01=E2=80=93T06).?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .env.example | 13 + docs/issue-core-emission-boundary.md | 20 ++ docs/ops-run-queue.md | 111 ++++++ docs/recurring-automations-playbook.md | 24 +- docs/runbook.md | 37 ++ docs/task-emission-consumer-contract.md | 36 +- migrations/versions/0007_create_ops_runs.py | 95 +++++ src/activity_core/activities.py | 20 ++ src/activity_core/api.py | 3 + src/activity_core/ops_api.py | 33 +- src/activity_core/ops_run_queue.py | 299 +++++++++++++++ src/activity_core/ops_runs_api.py | 339 ++++++++++++++++++ src/activity_core/orm.py | 42 +++ tests/test_ops_run_queue.py | 223 ++++++++++++ .../ACTIVITY-WP-0026-ops-run-claim-queue.md | 26 +- 15 files changed, 1294 insertions(+), 27 deletions(-) create mode 100644 docs/ops-run-queue.md create mode 100644 migrations/versions/0007_create_ops_runs.py create mode 100644 src/activity_core/ops_run_queue.py create mode 100644 src/activity_core/ops_runs_api.py create mode 100644 tests/test_ops_run_queue.py diff --git a/.env.example b/.env.example index 8073f6a..44a085e 100644 --- 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. diff --git a/docs/issue-core-emission-boundary.md b/docs/issue-core-emission-boundary.md index 2dc806e..b452da7 100644 --- 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 diff --git a/docs/ops-run-queue.md b/docs/ops-run-queue.md new file mode 100644 index 0000000..aaf77d2 --- /dev/null +++ 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 | diff --git a/docs/recurring-automations-playbook.md b/docs/recurring-automations-playbook.md index 5aae481..5b25125 100644 --- 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 diff --git a/docs/runbook.md b/docs/runbook.md index a3fe63d..77722e8 100644 --- 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 diff --git a/docs/task-emission-consumer-contract.md b/docs/task-emission-consumer-contract.md index 5fef3ca..8d7c4ad 100644 --- 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. diff --git a/migrations/versions/0007_create_ops_runs.py b/migrations/versions/0007_create_ops_runs.py new file mode 100644 index 0000000..dfc0853 --- /dev/null +++ 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") diff --git a/src/activity_core/activities.py b/src/activity_core/activities.py index fd9eb6b..8b7efac 100644 --- 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) diff --git a/src/activity_core/api.py b/src/activity_core/api.py index a03f99f..c0923c3 100644 --- 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]: diff --git a/src/activity_core/ops_api.py b/src/activity_core/ops_api.py index 8e21a02..456953a 100644 --- 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}") diff --git a/src/activity_core/ops_run_queue.py b/src/activity_core/ops_run_queue.py new file mode 100644 index 0000000..cb3e866 --- /dev/null +++ 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 diff --git a/src/activity_core/ops_runs_api.py b/src/activity_core/ops_runs_api.py new file mode 100644 index 0000000..8a1e8c5 --- /dev/null +++ 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} diff --git a/src/activity_core/orm.py b/src/activity_core/orm.py index 6818a51..5293468 100644 --- 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(), + ) diff --git a/tests/test_ops_run_queue.py b/tests/test_ops_run_queue.py new file mode 100644 index 0000000..4e72caf --- /dev/null +++ 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 diff --git a/workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md b/workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md index 6687bba..920f20c 100644 --- 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