From 21dc228cb1eb193e4ed3a1f207178356c1cef45f Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 5 Aug 2026 15:30:00 +0200 Subject: [PATCH] fix(ops_run): unique triggering_event_id per cron fire Cron schedules always passed trigger_key="scheduled", so ops_run idempotency collapsed every weekday into one key. After the first fire, create_ops_run was a silent no-op, the claim loop starved, and dual-clock host timers produced empty FI briefs. Map scheduled fires to run_id (or scheduled:{iso}) via emit_triggering_event_id; log duplicate skips; document the contract. --- WORK-RECORDS.md | 4 +-- docs/ops-run-queue.md | 9 ++++- src/activity_core/activities.py | 17 ++++++++-- src/activity_core/ops_run_queue.py | 29 ++++++++++++++++ src/activity_core/workflows.py | 12 ++++++- tests/test_ops_run_queue.py | 53 ++++++++++++++++++++++++++++++ 6 files changed, 117 insertions(+), 7 deletions(-) diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 4905c43..6c95747 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -29,7 +29,7 @@ | workplan | ACTIVITY-WP-0023 | finished | — | workplans/ACTIVITY-WP-0023-intent-scope-gap-closure.md | | workplan | ACTIVITY-WP-0024 | finished | — | workplans/ACTIVITY-WP-0024-operator-automation-console.md | | workplan | ACTIVITY-WP-0025 | finished | — | workplans/ACTIVITY-WP-0025-ops-ui-sso-access.md | -| workplan | ACTIVITY-WP-0026 | in_progress | — | workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md | +| workplan | ACTIVITY-WP-0026 | done | — | workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md | | workplan | ADHOC-2026-06-01 | finished | — | workplans/ADHOC-2026-06-01.md | | workplan | custodian-WP-0001 | done | — | workplans/custodian-WP-0001-temporal-backbone.md | | workplan | custodian-WP-0002 | done | — | workplans/custodian-WP-0002-triggers-ops.md | @@ -158,7 +158,7 @@ | task | ACTIVITY-WP-0026-T04 | done | — | workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md | | task | ACTIVITY-WP-0026-T05 | done | — | workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md | | task | ACTIVITY-WP-0026-T06 | done | — | workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md | -| task | ACTIVITY-WP-0026-T07 | todo | — | workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md | +| task | ACTIVITY-WP-0026-T07 | done | — | workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md | | task | ADHOC-2026-06-01-T01 | done | — | workplans/ADHOC-2026-06-01.md | | task | ADHOC-2026-06-01-T02 | done | — | workplans/ADHOC-2026-06-01.md | | task | ADHOC-2026-06-01-T03 | done | — | workplans/ADHOC-2026-06-01.md | diff --git a/docs/ops-run-queue.md b/docs/ops-run-queue.md index dccf918..6d1ee57 100644 --- a/docs/ops-run-queue.md +++ b/docs/ops-run-queue.md @@ -13,7 +13,7 @@ workplan task file. Not an issue-core or Forgejo ticket. | ----- | ---- | ----- | | `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}` | +| `idempotency_key` | text | **Unique**; default `{activity_id}:{source_id}:{triggering_event_id}` — for cron fires, `triggering_event_id` must be **per-fire** (`run_id` or `scheduled:{iso}`), never bare `scheduled` | | `target_repo` | text | From TaskSpec | | `title` | text | | | `description` | text | | @@ -80,6 +80,13 @@ On `emit_tasks` (when `OPS_RUN_QUEUE_ENABLED` is truthy, **default true**): 2. Dual-write existing IssueSink (`state-hub` progress by default). 3. Write `task_spawn_log` audit as today. +**Cron trigger keys:** `RunActivityWorkflow` maps the schedule sentinel +`trigger_key="scheduled"` to a per-fire `triggering_event_id` via +`emit_triggering_event_id()` (`scheduled:{scheduled_for}` when known, else +`run_id`). Using bare `scheduled` for every fire made only the first weekday +create an `ops_run`; later days recorded `activity_runs` but left the claim +queue empty. + Never requires Forgejo or issue-core for the claim path. ## Auth diff --git a/src/activity_core/activities.py b/src/activity_core/activities.py index 8b7efac..8e5e9d4 100644 --- a/src/activity_core/activities.py +++ b/src/activity_core/activities.py @@ -473,6 +473,11 @@ async def emit_tasks(payload: dict) -> list[str]: try: # ACTIVITY-WP-0026: claimable ops_run (primary for harness) try: + from activity_core.ops_run_queue import ( + build_idempotency_key, + ops_run_queue_enabled, + ) + ops_id = await create_ops_run_from_spec( session, spec, @@ -480,10 +485,16 @@ async def emit_tasks(payload: dict) -> list[str]: ) if ops_id is not None: activity.logger.info( - "emit_tasks: ops_run created id=%s key=%s:%s", + "emit_tasks: ops_run created id=%s key=%s", ops_id, - spec.source_id, - triggering_event_id, + build_idempotency_key(spec), + ) + elif ops_run_queue_enabled(): + # Unique constraint hit: redelivery of same fire (ok) + # OR a non-unique trigger key (bug — see emit_triggering_event_id). + activity.logger.info( + "emit_tasks: ops_run not inserted (duplicate or skip) key=%s", + build_idempotency_key(spec), ) except Exception as ops_exc: activity.logger.warning( diff --git a/src/activity_core/ops_run_queue.py b/src/activity_core/ops_run_queue.py index cb3e866..55651cd 100644 --- a/src/activity_core/ops_run_queue.py +++ b/src/activity_core/ops_run_queue.py @@ -45,6 +45,35 @@ def build_idempotency_key(spec: TaskSpec) -> str: ) +# Sentinel used by Temporal Schedule starters (see schedule_manager.SCHEDULED_TRIGGER_KEY). +SCHEDULED_TRIGGER_KEY = "scheduled" + + +def emit_triggering_event_id( + trigger_key: str, + run_id: str, + *, + scheduled_for: str | None = None, +) -> str: + """Return a per-fire key for task_spawn_log / ops_run idempotency. + + Event and manual triggers already pass a unique ``trigger_key``. Cron + schedules always pass the sentinel ``scheduled``; using that bare value as + ``triggering_event_id`` made every weekday fire share one ops_run + idempotency key, so only the first day created a claimable row. + + Preference for scheduled fires: + 1. ``scheduled_for`` when the starter supplied it (one-shot / future cron) + 2. ``run_id`` (uuid5 of workflow_id — unique per Temporal schedule fire, + stable on redelivery) + """ + if trigger_key != SCHEDULED_TRIGGER_KEY: + return trigger_key + if scheduled_for: + return f"scheduled:{scheduled_for}" + return run_id + + def ops_run_to_dict(row: OpsRun) -> dict[str, Any]: return { "id": str(row.id), diff --git a/src/activity_core/workflows.py b/src/activity_core/workflows.py index 0195d3a..1c2b90c 100644 --- a/src/activity_core/workflows.py +++ b/src/activity_core/workflows.py @@ -30,6 +30,7 @@ with workflow.unsafe.imports_passed_through(): persist_task_instance, resolve_context, ) + from activity_core.ops_run_queue import emit_triggering_event_id from activity_core.schedule_manager import SCHEDULED_TRIGGER_KEY # T32: Custom search attributes for Temporal visibility (must be registered in Temporal first). @@ -197,12 +198,21 @@ class RunActivityWorkflow: # ── 6. Emit tasks (may fail independently of run audit) ─────────────── if task_spec_dicts: + # Cron schedules pass trigger_key="scheduled" for *every* fire. + # ops_run idempotency is {def}:{source}:{triggering_event_id}, so + # a bare "scheduled" key collides after the first day and silently + # skips create (claim loop starves; dual-clock timers paper over it). + # run_id is uuid5(workflow_id) for schedules and is unique per fire + # while stable on Temporal redelivery of the same fire. + emit_trigger_id = emit_triggering_event_id( + trigger_key, run_id, scheduled_for=scheduled_for + ) await workflow.execute_activity( emit_tasks, { "task_specs": task_spec_dicts, "activity_id": activity_id, - "triggering_event_id": trigger_key, + "triggering_event_id": emit_trigger_id, "run_id": run_id, }, start_to_close_timeout=_ACTIVITY_TIMEOUT, diff --git a/tests/test_ops_run_queue.py b/tests/test_ops_run_queue.py index 4e72caf..c45fd7c 100644 --- a/tests/test_ops_run_queue.py +++ b/tests/test_ops_run_queue.py @@ -11,6 +11,7 @@ import pytest from activity_core.ops_run_queue import ( build_idempotency_key, + emit_triggering_event_id, max_attempts, ops_run_queue_enabled, ops_run_to_dict, @@ -39,6 +40,58 @@ def test_build_idempotency_key() -> None: ) +def test_emit_triggering_event_id_scheduled_uses_run_id() -> None: + """Bare 'scheduled' must not be the ops_run key for every weekday fire.""" + day1 = emit_triggering_event_id("scheduled", "run-day-1") + day2 = emit_triggering_event_id("scheduled", "run-day-2") + assert day1 == "run-day-1" + assert day2 == "run-day-2" + assert day1 != day2 + + +def test_emit_triggering_event_id_prefers_scheduled_for() -> None: + got = emit_triggering_event_id( + "scheduled", + "run-fallback", + scheduled_for="2026-08-05T05:30:00+00:00", + ) + assert got == "scheduled:2026-08-05T05:30:00+00:00" + + +def test_emit_triggering_event_id_passes_through_events_and_manual() -> None: + assert emit_triggering_event_id("evt-abc", "run-x") == "evt-abc" + assert ( + emit_triggering_event_id("manual-3da4cf06", "run-y") == "manual-3da4cf06" + ) + + +def test_scheduled_fires_produce_distinct_ops_keys() -> None: + """Regression: two cron days must not share one ops_run idempotency key.""" + def_id = "3169ab1f-882b-59c7-9763-d014dc96f4fc" + source = "emit-fi-daily-brief-task" + key_aug4 = build_idempotency_key( + TaskSpec( + title="FI", + activity_definition_id=def_id, + source_id=source, + triggering_event_id=emit_triggering_event_id("scheduled", "run-aug4"), + ) + ) + key_aug5 = build_idempotency_key( + TaskSpec( + title="FI", + activity_definition_id=def_id, + source_id=source, + triggering_event_id=emit_triggering_event_id("scheduled", "run-aug5"), + ) + ) + # Old bug: both would be "...:scheduled" + legacy = f"{def_id}:{source}:scheduled" + assert key_aug4 != key_aug5 + assert key_aug4 != legacy + assert key_aug5 != legacy + + def test_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("OPS_RUN_MAX_ATTEMPTS", raising=False) assert max_attempts() == 3