fix(scheduler): return a real execution_id, and 409 when nothing started (#1968)

docs/memory/learnings.md

@@ -284,3 +284,7 @@ plan or review. `/autoplan` reads this before planning; write for that reader.
 ## 2026-08-03 — pitfall — A shared helper imported from `services.*` becomes a MagicMock in half the suite
 **Context**: trinity-enterprise#314. I put the `parse_template_yaml` policy wrapper in `services/template_service.py` and imported it into `services/agent_service/crud.py`. Several harnesses (`test_1759`, the #1484 characterization fixture) stub `services.*` wholesale via `patch.dict("sys.modules", ...)`, so at the call site the helper resolved to a MagicMock, the parse returned a MagicMock, and `crud.py`'s own `isinstance(template_data, dict)` check rejected every local template as "empty or malformed template.yaml" — 7 failures that looked like a parser bug and were an import-location bug.
 **Lesson**: a helper that CROSS-CUTS services (a parser, a validator, a formatter) belongs in `utils/*`, not in whichever service happened to need it first. `utils.*` is import-safe from anywhere; `services.*` is stubbed wholesale by any test that wants to isolate a sibling, and the failure mode is not an ImportError but a silently-truthy Mock flowing into a type check several frames away. Bonus: it also avoids the new import edge (`crud -> template_service`) that the layering does not otherwise have.
+
+## 2026-08-04 — pitfall — A fire-and-forget `create_task` gets more dangerous the more state you create before it
+**Context**: trinity#1968 self-review. `_trigger_handler` was already spawning its run with a bare `asyncio.create_task(...)` whose result nobody kept — the event loop holds only a WEAK reference, so the task can be collected mid-flight (the asyncio docs say so outright). That was survivable while the handler created nothing first: a collected task meant the run silently didn't happen. The fix for #1968 moved the lock acquisition and the execution-row INSERT to BEFORE the spawn so the response could carry a real `execution_id` — and in doing so turned the same latent footgun into a stranded `running` row whose id the caller is holding, plus a schedule lock pinned until its Redis TTL. Nothing about the `create_task` line changed; what changed is what it now owns. `agent_server/services/result_callback.py` (#1083) already had the `_inflight` set + `add_done_callback(discard)` shape for exactly this.
+**Lesson**: (1) When a change moves resource acquisition EARLIER — before an async handoff, a return, or a commit — re-audit the handoff itself even though its code is untouched. The diff shows the moved lines, not the line whose blast radius they changed; a review that only reads the diff cannot see this. (2) A bare `asyncio.create_task` is a latent bug whose severity is set by what the caller allocated before it: nothing → a lost no-op; a lock or a DB row → a leak that outlives the request. Keep a strong reference (module- or instance-level set, discarded in a done-callback) at every spawn site that owns state, not just the ones that look important. (3) When the codebase already contains the guard pattern somewhere (here `_inflight` from #1083), the second site is not a new design decision — grep for the pattern before inventing one, and cite the precedent so the third site is obvious.

src/backend/routers/schedules.py

@@ -483,6 +483,16 @@ async def trigger_schedule(
                     status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                     detail="Scheduler service unavailable"
                 )
+            elif response.status_code == 409:
+                # #1968: the scheduler declined because this schedule is
+                # already running. Relayed as a 409 rather than flattened into
+                # the 500 below, because it is not a failure — it is the answer
+                # to the caller's question, and the old handler's silent
+                # `"status": "triggered"` for this case is the bug.
+                raise HTTPException(
+                    status_code=status.HTTP_409_CONFLICT,
+                    detail="Schedule is already executing"
+                )
             elif response.status_code != 200:
                 logger.error(f"Scheduler trigger failed: {response.status_code} - {response.text}")
                 raise HTTPException(
@@ -505,12 +515,20 @@ async def trigger_schedule(
                     "schedule_id": schedule_id,
                     "schedule_name": result.get("schedule_name"),
                     "triggered_by": "manual",
+                    # #1968: the audit row can now name the execution it
+                    # started, so a trigger and its run are joinable after the
+                    # fact rather than only correlatable by timestamp.
+                    "execution_id": result.get("execution_id"),
                 },
             )
 
             return {
                 "status": "triggered",
                 "schedule_id": schedule_id,
+                # #1968: the field the MCP tool has always read and never
+                # found. It was absent here because the scheduler responded
+                # before the row existed; it now creates the row first.
+                "execution_id": result.get("execution_id"),
                 "schedule_name": result.get("schedule_name"),
                 "agent_name": result.get("agent_name"),
                 "message": result.get("message", "Execution started")

src/cli/trinity_cli/commands/schedules.py

@@ -41,4 +41,13 @@ def trigger_schedule(agent, schedule_id):
     """Trigger a schedule immediately."""
     client = TrinityClient()
     data = client.post(f"/api/agents/{agent}/schedules/{schedule_id}/trigger")
-    click.echo(f"Triggered schedule {schedule_id} on '{agent}'")
+    # #1968: `data` was fetched and thrown away, so the command could not tell
+    # the user which run it had just started. The response now carries a real
+    # execution_id; print it, guarded, since an older backend still omits it.
+    execution_id = (data or {}).get("execution_id")
+    if execution_id:
+        click.echo(
+            f"Triggered schedule {schedule_id} on '{agent}' (execution {execution_id})"
+        )
+    else:
+        click.echo(f"Triggered schedule {schedule_id} on '{agent}'")

src/frontend/src/components/SchedulesPanel.vue

@@ -1337,7 +1337,20 @@ async function triggerSchedule(schedule) {
       await loadExecutions(schedule.id)
     }
   } catch (error) {
-    reportActionFailure(error, `run the schedule "${schedule.name}" now`)
+    // #1968: the backend now answers 409 when the schedule is already running,
+    // instead of reporting a success that started nothing. That is not the
+    // "nothing was changed — try again" case reportActionFailure describes: a
+    // run IS in flight, and retrying only hits the same lock. Say what is
+    // actually true, and reload so the user can see the run in question.
+    if (error?.response?.status === 409) {
+      actionError.value = `"${schedule.name}" is already running — no new run was started.`
+      actionErrorDetail.value = ''
+      if (expandedSchedule.value === schedule.id) {
+        await loadExecutions(schedule.id)
+      }
+    } else {
+      reportActionFailure(error, `run the schedule "${schedule.name}" now`)
+    }
   } finally {
     triggerLoading.value = null
   }

src/mcp-server/src/tools/schedules.ts

@@ -5,7 +5,7 @@
  */
 
 import { z } from "zod";
-import { TrinityClient } from "../client.js";
+import { ApiError, TrinityClient } from "../client.js";
 import type { McpAuthContext } from "../types.js";
 
 /**
@@ -536,17 +536,52 @@ export function createScheduleTools(
 
         // #1970: carry the caller identity through so the resulting execution
         // row is attributable to this key/agent, not just to the key owner.
-        const result = await apiClient.triggerAgentSchedule(
-          agent_name,
-          schedule_id,
-          authContext?.agentName,
-          authContext
-            ? { keyId: authContext.keyId, keyName: authContext.keyName }
-            : undefined
-        );
+        let result;
+        try {
+          result = await apiClient.triggerAgentSchedule(
+            agent_name,
+            schedule_id,
+            authContext?.agentName,
+            authContext
+              ? { keyId: authContext.keyId, keyName: authContext.keyName }
+              : undefined
+          );
+        } catch (error) {
+          // #1968: "already running" is an ANSWER, not a failure. Surfacing the
+          // raw ApiError would hand the calling agent a stack-shaped string it
+          // has to parse; a structured status lets it decide to wait and poll
+          // instead of retrying into the same lock.
+          if (error instanceof ApiError && error.status === 409) {
+            console.log(`[trigger_agent_schedule] Schedule '${schedule_id}' is already executing`);
+            return JSON.stringify({
+              status: "already_running",
+              schedule_id,
+              agent_name,
+              message:
+                "This schedule is already executing, so no new run was started. " +
+                "Poll list_recent_executions for the run in flight.",
+            }, null, 2);
+          }
+          throw error;
+        }
 
         console.log(`[trigger_agent_schedule] Triggered schedule '${schedule_id}' for agent '${agent_name}', execution_id: ${result.execution_id}`);
 
+        // #1968: `execution_id` used to be absent from the response, so this
+        // string always read "...with ID 'undefined'". Guard the wording rather
+        // than assume — an old scheduler behind a new MCP server still omits it,
+        // and a confident lie about the id is what this issue is about.
+        if (!result.execution_id) {
+          return JSON.stringify({
+            status: "triggered",
+            schedule_id,
+            execution_id: null,
+            message:
+              "Schedule triggered, but the backend returned no execution id. " +
+              "Find the run via list_recent_executions.",
+          }, null, 2);
+        }
+
         return JSON.stringify({
           status: "triggered",
           schedule_id,

src/mcp-server/src/types.ts

@@ -270,7 +270,12 @@ export interface ScheduleToggleResult {
 export interface ScheduleTriggerResult {
   status: "triggered";
   schedule_id: string;
-  execution_id: string;
+  // #1968: optional, and that is the honest declaration. This was typed as a
+  // required `string` while the backend never sent the field at all — which is
+  // precisely why the compiler stayed happy while every trigger interpolated
+  // `undefined` into its success message. A current backend fills it in; an
+  // older one still omits it, so callers must check rather than trust the type.
+  execution_id?: string;
   message?: string;
 }
 

src/scheduler/main.py

@@ -64,6 +64,10 @@ def __init__(self):
         self.health_app: Optional[web.Application] = None
         self.health_runner: Optional[web.AppRunner] = None
         self._shutdown_event = asyncio.Event()
+        # #1968: strong refs to the fire-and-forget trigger tasks. The loop keeps
+        # only a weak one, and these tasks now own a held lock and a live
+        # execution row — see `_trigger_handler` for why that matters.
+        self._inflight_triggers: set = set()
 
     async def start(self):
         """Start the scheduler application."""
@@ -210,17 +214,89 @@ async def _trigger_handler(self, request: web.Request) -> web.Response:
             f"source_user_id={origin.user_id} source_agent={origin.agent_name}"
         )
 
-        # Execute in background (fire-and-forget)
-        asyncio.create_task(
+        # #1968: acquire the lock and create the execution row HERE, before
+        # responding. Both facts the caller needs — whether the trigger took
+        # effect, and which execution it produced — only exist after these two
+        # steps, and the old handler answered before either had happened.
+        #
+        # It reported `"status": "triggered"` with no id even when the lock was
+        # denied and nothing ran, so a suppressed trigger and a real one were
+        # byte-identical to the caller. The MCP tool then interpolated the
+        # missing key and told every agent `Execution started with ID
+        # 'undefined'`.
+        lock = self.scheduler_service.lock_manager.try_acquire_schedule_lock(schedule_id)
+        if not lock:
+            logger.warning(
+                f"Trigger for {schedule_id} (triggered_by={triggered_by}): "
+                "schedule already executing"
+            )
+            return web.json_response({
+                "status": "already_running",
+                "schedule_id": schedule_id,
+                "schedule_name": schedule.name,
+                "agent_name": schedule.agent_name,
+                "message": "Schedule is already executing",
+            }, status=409)
+
+        # From here the lock is HELD. Every path below must either hand it to
+        # the background task or release it — a leak would wedge the schedule
+        # until the Redis TTL expires.
+        try:
+            execution = self.scheduler_service.db.create_execution(
+                schedule_id=schedule.id,
+                agent_name=schedule.agent_name,
+                message=schedule.message,
+                triggered_by=triggered_by,
+                model_used=schedule.model,
+                source_user_id=origin.user_id,
+                source_user_email=origin.user_email,
+                source_agent_name=origin.agent_name,
+                source_mcp_key_id=origin.mcp_key_id,
+                source_mcp_key_name=origin.mcp_key_name,
+            )
+        except Exception as exc:
+            lock.release()
+            logger.error(f"Trigger for {schedule_id}: could not create execution: {exc}")
+            return web.json_response(
+                {"error": "Failed to create execution record"}, status=500
+            )
+
+        if not execution:
+            lock.release()
+            logger.error(f"Trigger for {schedule_id}: could not create execution record")
+            return web.json_response(
+                {"error": "Failed to create execution record"}, status=500
+            )
+
+        # Execute in background, handing over the lock we already hold and the
+        # row we already created.
+        #
+        # The strong reference is load-bearing, not defensive tidiness. The
+        # event loop holds only a WEAK reference to a task, so a bare
+        # `create_task(...)` whose result nobody keeps can be garbage-collected
+        # mid-flight (the asyncio docs say so outright). That was survivable
+        # before this change — a dropped task meant the run silently didn't
+        # happen. It is not survivable now: the row and the lock are created
+        # BEFORE the task, so a collected task strands a `running` execution
+        # whose id the caller already holds and pins the schedule's lock until
+        # its TTL. Same `_inflight` shape the #1083 result-callback path uses
+        # (`agent_server/services/result_callback.py`).
+        task = asyncio.create_task(
             self._execute_manual_trigger(
-                schedule_id, triggered_by=triggered_by, origin=origin
+                schedule_id,
+                triggered_by=triggered_by,
+                origin=origin,
+                lock=lock,
+                execution=execution,
             )
         )
+        self._inflight_triggers.add(task)
+        task.add_done_callback(self._inflight_triggers.discard)
 
-        # Return immediately; execution creates its own record asynchronously
         return web.json_response({
             "status": "triggered",
             "schedule_id": schedule_id,
+            "execution_id": execution.id,
             "schedule_name": schedule.name,
             "agent_name": schedule.agent_name,
             "triggered_by": triggered_by,
@@ -232,31 +308,55 @@ async def _execute_manual_trigger(
         schedule_id: str,
         triggered_by: str = "manual",
         origin: Optional[ExecutionOrigin] = None,
+        lock=None,
+        execution=None,
     ):
-        """Execute a manually or webhook-triggered schedule."""
-        try:
-            # Acquire lock (prevents concurrent execution)
-            lock = self.scheduler_service.lock_manager.try_acquire_schedule_lock(schedule_id)
+        """Execute a manually or webhook-triggered schedule.
+
+        #1968: `lock` and `execution` are acquired/created by `_trigger_handler`
+        BEFORE it responds, so the response can carry a real `execution_id` and
+        a denied lock can be reported as 409 instead of a false "triggered".
+        Both are passed in rather than taken here; this coroutine's job is to
+        run the schedule and, whatever happens, release the lock.
+
+        They stay optional so the method keeps working if called without them
+        (it then acquires its own lock, as before, and lets the service create
+        the row) — a caller that forgets is degraded, not broken.
+
+        Whichever way the lock arrived, this coroutine owns it from here and
+        releases it exactly once. Structured as acquire-then-single-`finally`
+        rather than a release in both the `finally` and an error branch: the
+        second release is the dangerous one, since a lock re-acquired by the
+        next run in between would be freed out from under it.
+        """
+        if lock is None:
+            try:
+                lock = self.scheduler_service.lock_manager.try_acquire_schedule_lock(
+                    schedule_id
+                )
+            except Exception as exc:
+                logger.error(f"Trigger for {schedule_id}: lock acquisition failed: {exc}")
+                return
             if not lock:
                 logger.warning(
                     f"Trigger for {schedule_id} (triggered_by={triggered_by}): "
                     "schedule already executing"
                 )
                 return
 
-            try:
-                await self.scheduler_service._execute_schedule_with_lock(
-                    schedule_id,
-                    triggered_by=triggered_by,
-                    origin=origin
-                )
-            finally:
-                lock.release()
-
+        try:
+            await self.scheduler_service._execute_schedule_with_lock(
+                schedule_id,
+                triggered_by=triggered_by,
+                origin=origin,
+                execution=execution
+            )
         except Exception as e:
             logger.error(
                 f"Trigger execution failed for {schedule_id} (triggered_by={triggered_by}): {e}"
             )
+        finally:
+            lock.release()
 
     async def _run_until_shutdown(self):
         """Run the scheduler until shutdown signal."""