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.

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(

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),

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,