From c44948ef968a398be2fc1ef6dbd931294df11f28 Mon Sep 17 00:00:00 2001 From: Oleksii Dolhov Date: Tue, 4 Aug 2026 11:06:57 +0300 Subject: [PATCH 1/5] fix(scheduler): record who initiated an execution (#1970) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `schedule_executions` has carried five origin columns for audit since AUDIT-001, and the backend populates them on every path it owns. The scheduler is a separate service with its own DB module, and its `create_execution()` listed none of them in the INSERT — nor accepted them in its signature, so there was nowhere to put a caller even if one had been forwarded. Every scheduler-created row was written with all five NULL. `triggered_by='manual'` therefore recorded *that* a human ran something and never *who*. The attribution lived only in backend/MCP-server logs, bounded by log retention, so past a few weeks the durable record could not answer "did anyone trigger this run, and who?". The identity was dropped at three points, not one: 1. the backend's delegating POST sent no body at all, so the authenticated caller — in scope right there — never crossed the hop; 2. `_trigger_handler` had no parameter to receive one; 3. `create_execution()` had nowhere to put it. An `ExecutionOrigin` value object is threaded through all three. One object rather than five parallel parameters at four call depths: five positional siblings is how one of them silently stops being forwarded. Two paths the DB fix alone would have left blank are covered too. A retry inherits the original run's origin — it has no caller of its own, but a chain of retries that drops the initiator makes the first attempt the only attributable one; the read is fail-open, since an audit lookup must not be able to stop a retry from running. A reminder inherits the provenance #1296 already persisted. Cron ticks stay NULL. Attributing an autonomous fire to, say, the schedule's owner would make the column actively misleading — a blank reads as "unknown", a wrong name does not. Also hardened while here: - the untrusted trigger body is validated at the scheduler boundary. `source_user_id` is dropped rather than coerced when it is not an int: `bool` IS an `int` in Python, so `True` would have persisted as user 1, a real account attributed to a run it had nothing to do with. Strings are length-capped and blank-to-None, so "" and NULL are not two spellings of "unknown". - the backend prefers the validated `current_user.agent_name` over the raw `X-Source-Agent` header — the reverse of chat.py's precedence, which is fine for a collaboration hint but would let a caller pin its run on a sibling agent in an audit column. - the MCP trigger tool forwards the origin headers `chat()` already sends (Invariant #13). Without it an MCP-triggered run attributes to the key OWNER but not to which key or agent fired it — the part that identifies the actor when one human owns many of both. Not a vulnerability: nothing authorizes on these columns. Backward compatible in both rolling-deploy directions — an old scheduler ignores the new body fields, and a new scheduler treats a bodyless POST as an unattributed manual trigger. tests/unit/test_1970_execution_origin.py — 27 checks, 25 of which fail against the pre-fix tree. Related to #1970 Co-Authored-By: Claude Opus 5 (1M context) --- src/scheduler/models.py | 58 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/src/scheduler/models.py b/src/scheduler/models.py index 4c97294f4..c83c78e3e 100644 --- a/src/scheduler/models.py +++ b/src/scheduler/models.py @@ -183,6 +183,64 @@ def _str(key: str) -> Optional[str]: ) +@dataclass +class ExecutionOrigin: + """Who initiated an execution (AUDIT-001, #1970). + + One value object instead of five parallel parameters threaded through + ``_trigger_handler → _execute_manual_trigger → _execute_schedule_with_lock + → create_execution`` — five positional siblings at four call depths is how + one of them silently stops being forwarded. + + Attribution only; **nothing authorizes on these fields.** An all-``None`` + origin is the correct, honest record for a cron tick, which has no caller. + """ + user_id: Optional[int] = None + user_email: Optional[str] = None + agent_name: Optional[str] = None + mcp_key_id: Optional[str] = None + mcp_key_name: Optional[str] = None + + def is_empty(self) -> bool: + return not any( + (self.user_id, self.user_email, self.agent_name, + self.mcp_key_id, self.mcp_key_name) + ) + + @classmethod + def from_payload(cls, body: object) -> "ExecutionOrigin": + """Build from an untrusted JSON body (the scheduler's trigger endpoint). + + Validates at the boundary: wrong types are dropped rather than coerced, + and strings are length-capped, so a malformed or oversized payload + cannot write junk into an append-only-in-spirit audit column. A caller + that lies about its identity is not a new exposure — ``triggered_by`` + has been caller-supplied on this same endpoint all along, and the + scheduler is reachable only from the platform network. + """ + if not isinstance(body, dict): + return cls() + + def _str(key: str) -> Optional[str]: + value = body.get(key) + if not isinstance(value, str): + return None + value = value.strip() + return value[:255] or None + + user_id = body.get("source_user_id") + if isinstance(user_id, bool) or not isinstance(user_id, int): + user_id = None + + return cls( + user_id=user_id, + user_email=_str("source_user_email"), + agent_name=_str("source_agent_name"), + mcp_key_id=_str("source_mcp_key_id"), + mcp_key_name=_str("source_mcp_key_name"), + ) + + @dataclass class Reminder: """A durable one-shot agent self-reminder (#1296). From 3225c94772f58d82f92b7626e77f014931470f31 Mon Sep 17 00:00:00 2001 From: Oleksii Dolhov Date: Tue, 4 Aug 2026 11:41:40 +0300 Subject: [PATCH 2/5] fix(scheduler): return a real execution_id, and 409 when nothing started (#1968) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_trigger_handler` was fire-and-forget: it spawned the run with `asyncio.create_task` and responded immediately, *before* the execution record existed. So it had no id to return. The backend relayed the same id-less fields, and the MCP tool interpolated the missing key — telling every agent `Execution started with ID 'undefined'`, on every trigger, while the execution ran fine. Callers could not correlate a trigger with its run, poll it, or fetch its result; the workaround was to guess from `list_recent_executions` by timestamp. The same ordering hid a second problem. The response was emitted before `_execute_manual_trigger` had even attempted the distributed lock, so a trigger suppressed because the schedule was already running still answered `"status": "triggered"`. A suppressed trigger and a real one were byte-identical to the caller. The handler now acquires the lock and creates the row synchronously, then hands both to the background task. That makes two facts sayable that simply did not exist yet at response time: which execution this is, and whether one was started at all. * 200 carries a real `execution_id`, valid the moment the caller receives it — a fast poller must not 404. * 409 `already_running` replaces the false "triggered", with no id and no row, because nothing ran. Exactly one row per trigger: `_execute_schedule_with_lock` takes the pre-created execution and skips its own create. Two rows would hand the caller an id naming a row that never runs while a second did the work. Because a row can now exist before a gate decides not to run, an abandoned run FAILs its pre-created row rather than leaving it `running` forever — canary E-01's exact signature, and a task the UI would show indefinitely. The handler also now holds the lock across a DB write, which is new, so every exit from that window releases it: creation raising, creation returning None, the run raising, and normal completion — exactly once each. A second release is the dangerous one, since a lock re-acquired by the next run in between would be freed out from under it. Relayed through the remaining surfaces: * the backend forwards `execution_id` (and records it on the audit row, so a trigger and its run are joinable after the fact) and maps 409 rather than flattening it into "Failed to trigger schedule" — a worse lie than the original, since it claims failure where the schedule is healthily busy; * the MCP tool returns a structured `already_running` instead of throwing, so an agent gets a decision it can act on, and GUARDS the id instead of interpolating it — an older backend still omits the field, and swapping one confident lie for another is not a fix; * `ScheduleTriggerResult.execution_id` becomes optional. Typing it as a required `string` while the wire never sent it is precisely why the compiler stayed happy through every `undefined`; * the UI reads 409 as "already running" rather than "nothing was changed — try again", and the CLI prints the id it was already fetching and discarding. tests/unit/test_1968_trigger_execution_id.py — 22 checks, 17 of which fail against the pre-fix tree. Related to #1968 Co-Authored-By: Claude Opus 5 (1M context) --- src/backend/routers/schedules.py | 18 + src/cli/trinity_cli/commands/schedules.py | 11 +- .../src/components/SchedulesPanel.vue | 15 +- src/mcp-server/src/tools/schedules.ts | 53 +- src/mcp-server/src/types.ts | 7 +- src/scheduler/main.py | 115 +++- src/scheduler/service.py | 86 ++- tests/registry.json | 13 + tests/unit/test_1968_trigger_execution_id.py | 607 ++++++++++++++++++ 9 files changed, 880 insertions(+), 45 deletions(-) create mode 100644 tests/unit/test_1968_trigger_execution_id.py diff --git a/src/backend/routers/schedules.py b/src/backend/routers/schedules.py index b7de036e2..c302f0656 100644 --- a/src/backend/routers/schedules.py +++ b/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") diff --git a/src/cli/trinity_cli/commands/schedules.py b/src/cli/trinity_cli/commands/schedules.py index 2c5f3f454..fc5c7edb2 100644 --- a/src/cli/trinity_cli/commands/schedules.py +++ b/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}'") diff --git a/src/frontend/src/components/SchedulesPanel.vue b/src/frontend/src/components/SchedulesPanel.vue index 10213f17a..49c0c41b4 100644 --- a/src/frontend/src/components/SchedulesPanel.vue +++ b/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 } diff --git a/src/mcp-server/src/tools/schedules.ts b/src/mcp-server/src/tools/schedules.ts index e31e69002..f24f00d95 100644 --- a/src/mcp-server/src/tools/schedules.ts +++ b/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, diff --git a/src/mcp-server/src/types.ts b/src/mcp-server/src/types.ts index 3df659c00..18807e0fb 100644 --- a/src/mcp-server/src/types.ts +++ b/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; } diff --git a/src/scheduler/main.py b/src/scheduler/main.py index 292579024..16e53551d 100644 --- a/src/scheduler/main.py +++ b/src/scheduler/main.py @@ -210,17 +210,76 @@ 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) + # #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. 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, ) ) - # 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,11 +291,35 @@ 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}): " @@ -244,19 +327,19 @@ async def _execute_manual_trigger( ) 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.""" diff --git a/src/scheduler/service.py b/src/scheduler/service.py index e661f050b..642bd7049 100644 --- a/src/scheduler/service.py +++ b/src/scheduler/service.py @@ -851,11 +851,34 @@ async def _execute_schedule(self, schedule_id: str): finally: lock.release() + def _abandon_precreated_execution(self, execution, reason: str) -> None: + """Fail an execution row created before this run could be aborted (#1968). + + A manual trigger now creates its row synchronously so the caller gets a + real id back, which means the row can already exist when a gate below + decides not to run. Leaving it `running` forever is the worse outcome: + canary E-01 flags exactly that, and the UI would show a task that never + ends. Fail it with the reason instead. Best-effort — an abort path must + not raise. + """ + if execution is None: + return + try: + self.db.update_execution_status( + execution.id, ExecutionStatus.FAILED, error=reason + ) + logger.info(f"Abandoned pre-created execution {execution.id}: {reason}") + except Exception as exc: + logger.error( + f"Could not abandon pre-created execution {execution.id}: {exc}" + ) + async def _execute_schedule_with_lock( self, schedule_id: str, triggered_by: str = "schedule", origin: Optional[ExecutionOrigin] = None, + execution=None, ): """Execute schedule after acquiring lock. @@ -865,14 +888,26 @@ async def _execute_schedule_with_lock( origin: Who initiated the run (#1970). None for a cron tick — there is no caller, and the execution row records that honestly as NULLs rather than inventing an owner. + execution: A row already created by the caller (#1968). The manual + trigger endpoint creates it synchronously so it can return a + real `execution_id`; passing it here is what stops this method + creating a SECOND row for the same run. None on the cron path, + which still creates its own. """ schedule = self.db.get_schedule(schedule_id) if not schedule: logger.error(f"Schedule {schedule_id} not found") + # Deleted between the trigger response and this task starting. + self._abandon_precreated_execution( + execution, "Schedule was deleted before the run started" + ) return if not schedule.enabled and triggered_by == "schedule": logger.info(f"Schedule {schedule_id} is disabled, skipping") + self._abandon_precreated_execution( + execution, "Schedule was disabled before the run started" + ) return # Check if agent has autonomy enabled (only for cron-triggered, not manual) @@ -885,6 +920,9 @@ async def _execute_schedule_with_lock( # the default-OFF fleet) and no last_run_at (nothing ran; setting it # would suppress the _get_missed_schedules catch-up). self._advance_next_run_only(schedule) + self._abandon_precreated_execution( + execution, "Agent autonomy was disabled before the run started" + ) return # #1808: git-sync freeze gate. The owner opted in via @@ -918,34 +956,48 @@ async def _execute_schedule_with_lock( # Same projection advance as the autonomy branch (#1472) so the # schedule never renders a receding "Next: Nd ago" while frozen. self._advance_next_run_only(schedule) + self._abandon_precreated_execution( + execution, "Git sync started failing before the run started" + ) return # Agent-owned pre-check gate (#454): may skip the firing entirely or # override the message. Manual triggers always fire. should_fire, effective_message = await self._apply_pre_check_gate(schedule, triggered_by) if not should_fire: + # Unreachable for a pre-created row: the gate short-circuits to + # (True, schedule.message) for anything but triggered_by="schedule", + # and only manual/webhook triggers pre-create. Abandon anyway — a + # future gate change must not silently strand a running row. + self._abandon_precreated_execution( + execution, "Agent pre-check declined the run" + ) return logger.info(f"Executing schedule: {schedule.name} for agent {schedule.agent_name} (triggered_by={triggered_by})") - # Create execution record - origin = origin or ExecutionOrigin() - execution = self.db.create_execution( - schedule_id=schedule.id, - agent_name=schedule.agent_name, - message=effective_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 - ) + # #1968: the manual path already created the row, synchronously, so its + # id could be returned to the caller. Creating another here would give + # one trigger two executions — the caller's id would name a row that + # never runs while a second one did the work. + if execution is None: + origin = origin or ExecutionOrigin() + execution = self.db.create_execution( + schedule_id=schedule.id, + agent_name=schedule.agent_name, + message=effective_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 + ) - if not execution: - logger.error(f"Failed to create execution record for schedule {schedule_id}") - return + if not execution: + logger.error(f"Failed to create execution record for schedule {schedule_id}") + return # #1472: advance the run-time projection ONCE, at fire time, regardless of # the eventual outcome — the single "this window was consumed" write. It diff --git a/tests/registry.json b/tests/registry.json index eb177c29f..a06bd6653 100644 --- a/tests/registry.json +++ b/tests/registry.json @@ -1553,6 +1553,19 @@ "parity" ], "description": "The agent server was outside ent#314's YAML sweep (#1965). utils/safe_yaml.py (PR #1961) put every author-controlled YAML reader in the backend behind one hardened loader, and its AST guard walks the whole backend with an EMPTY allowlist - but it walked _BACKEND.rglob only, so docker/base-image/agent_server/ kept six bare yaml.safe_load calls on documents the backend itself assigns REJECT: template.yaml (x2, credential_requirements_service), skill frontmatter (skill_packaging), dashboard.yaml (compatibility/static_checks) and .trinity/persistent-state.yaml. The vector is amplification at SERIALIZATION, not parse - a 416 B level-6 anchor bomb resolves in ~0.001 s and blows up to ~110 MB when something walks the graph - and the backend proxies /info and /dashboard, so the walk happens in-container then again across the wire. Covers: byte-parity of the vendored loader (the credential_paths.py shape, Invariant #5) plus proof the vendored COPY actually behaves - refuses a level-6 bomb under BUDGET, any alias under REJECT, duplicate keys, and still parses an honest document (byte parity is not behaviour parity if the file never imports); each of the four agent-authored sites on the shared loader with its backend counterpart's kind AND policy; /config/agent-config.yaml deliberately BUDGET not REJECT, stated as an exception because the platform writes it and bind-mounts it mode:'ro' so the agent cannot author it, and yaml.dump emits an anchor for any shared object reference - REJECT there would be a self-inflicted outage for no security; no bare safe_load left anywhere in the tree; and HardenedYamlError named in the except arms that previously caught only yaml.YAMLError (it is a ValueError, so without its own arm a refused bomb escapes to the generic handler and surfaces as the unnamed 500 the AC rules out - the trap static_checks._parse_yaml records backend-side). AC #4 end-to-end: a level-6 bomb in a container's template.yaml is refused by BOTH template.yaml readers with the expanded graph never reaching the response, an honest template still serves, and the metrics assertion checks the NAMED refusal rather than has_metrics:False - a bomb parses fine under bare safe_load and yields no metrics: key, so the flag alone passes against the very tree this issue reports. The AST-guard widening itself lives in test_ent314_hardened_yaml.py (both trees, still empty allowlist) rather than here, because splitting a guard across two files is how the second copy stops being run." + }, + { + "file": "unit/test_1968_trigger_execution_id.py", + "feature": "#1968", + "added": "2026-08-04", + "categories": [ + "scheduler", + "backend", + "mcp", + "unit", + "devex" + ], + "description": "Manual schedule trigger must name the execution it started, and admit when it started nothing (#1968). _trigger_handler was fire-and-forget: it spawned the run with asyncio.create_task and responded BEFORE the execution record existed, so it had no id to return, the backend relayed the same id-less fields, and the MCP tool interpolated the missing key - telling every agent \"Execution started with ID 'undefined'\" on every trigger. The same ordering hid a second thing: the response was emitted before the distributed lock was even attempted, so a suppressed trigger answered status='triggered' and was byte-identical to a real one. The fix acquires the lock and creates the row synchronously, then hands both to the background task. Covers: a real execution_id naming the row actually created; the id valid at RESPONSE time, not eventually (a fast poller must not 404); EXACTLY ONE row per trigger, asserted at both ends - the handler's, and the service not re-creating when passed one (two rows would give the caller an id naming a row that never runs while a second does the work); 409 already_running with no id and no row and no run. Lock hygiene, since the handler now holds a lock across a DB write: released exactly once on success (a double release would free a lock the NEXT run re-acquired), on a raising run, on create_execution raising, and on create_execution returning None; and no lock acquired at all for an unknown schedule. Orphan safety: a pre-created row whose run is then abandoned (schedule deleted in the window) is FAILED, not left running - canary E-01's exact signature - and the abandon helper never raises on an abort path. #1970 interaction: create_execution moved from the service into the handler, so the origin columns are pinned to move with it or that fix silently regresses to all-NULL. Relay: the backend forwards execution_id and maps 409 instead of flattening it to a 500 'Failed to trigger' (a worse lie - it claims failure where the schedule is healthily busy), the MCP tool answers 409 structurally instead of throwing and GUARDS the id rather than interpolating it unguarded (an older backend still omits it - do not swap one confident lie for another), ScheduleTriggerResult admits the field can be absent (typing it as a required string is why the compiler never flagged the undefined), and the UI reads 409 as 'already running' rather than 'nothing was changed - try again'." } ] } diff --git a/tests/unit/test_1968_trigger_execution_id.py b/tests/unit/test_1968_trigger_execution_id.py new file mode 100644 index 000000000..f44bff039 --- /dev/null +++ b/tests/unit/test_1968_trigger_execution_id.py @@ -0,0 +1,607 @@ +"""#1968 — a manual trigger must name the execution it started, and admit when +it started nothing. + +`_trigger_handler` was fire-and-forget: it spawned the run with +`asyncio.create_task` and responded immediately, *before* the execution record +existed. So it had no id to return, the backend relayed the same id-less five +fields, and the MCP tool interpolated the missing key — telling every agent +`Execution started with ID 'undefined'` on every single trigger. A caller could +not correlate its trigger with a run, poll it, or fetch its result. + +The same ordering hid a second thing. The response was emitted before +`_execute_manual_trigger` had even attempted the distributed lock, so a trigger +suppressed because the schedule was already running still answered +`"status": "triggered"`. A suppressed trigger and a real one were byte-identical +to the caller; the only trace was a scheduler-side WARNING. + +The fix acquires the lock and creates the row synchronously in the handler, then +hands both to the background task. That makes two facts sayable that previously +did not exist yet at response time: which execution this is, and whether one was +started at all (409 `already_running`). + +What is pinned here, beyond the happy path: + * exactly ONE execution row per trigger — the handler creating one and + `_execute_schedule_with_lock` creating another would give the caller an id + naming a row that never runs while a second row does the work; + * the lock is released exactly once on every path, including the failure ones + — the handler now holds it across a DB write, which is new; + * a pre-created row is never left `running` when a gate aborts the run (canary + E-01 flags exactly that shape); + * the MCP tool answers 409 with a structured `already_running` rather than + throwing, and never claims an id it did not receive. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import re +import sqlite3 +import sys +from pathlib import Path + +import pytest + +_REPO = Path(__file__).resolve().parents[2] + +# The `src.scheduler` namespace import resolves only with the repo root on +# sys.path — true for a repo-root `pytest` run but NOT in CI, whose rootdir is +# `tests/`. Appended (never inserted at 0) so the repo root cannot shadow the +# conftest-managed `src/backend` entries. Mirrors test_1808. +if str(_REPO) not in sys.path: + sys.path.append(str(_REPO)) + +os.environ.setdefault("REDIS_URL", "redis://test:test@redis:6379") +os.environ.setdefault("REDIS_PASSWORD", "test") +os.environ.setdefault("REDIS_BACKEND_PASSWORD", "test") + + +def _main_module(): + import src.scheduler.main as scheduler_main + + return scheduler_main + + +def _service_module(): + import src.scheduler.service as scheduler_service + + return scheduler_service + + +def _database_module(): + import src.scheduler.database as scheduler_database + + return scheduler_database + + +# --------------------------------------------------------------------------- +# Fakes: the handler is aiohttp-shaped, so drive it with the minimum surface. +# --------------------------------------------------------------------------- + + +class _FakeRequest: + def __init__(self, schedule_id: str, body: dict | None = None): + self.match_info = {"schedule_id": schedule_id} + self._body = body + self.content_type = "application/json" if body is not None else "text/plain" + self.content_length = len(json.dumps(body)) if body is not None else 0 + + async def json(self): + return self._body + + +class _FakeLock: + def __init__(self): + self.release_count = 0 + + def release(self): + self.release_count += 1 + + +class _FakeLockManager: + def __init__(self, *, grant: bool): + self.lock = _FakeLock() if grant else None + self.acquire_count = 0 + + def try_acquire_schedule_lock(self, schedule_id): + self.acquire_count += 1 + return self.lock + + +def _seed_db(db_path: Path) -> None: + conn = sqlite3.connect(db_path) + conn.execute( + """ + CREATE TABLE agent_schedules ( + id TEXT PRIMARY KEY, agent_name TEXT, name TEXT, cron_expression TEXT, + message TEXT, enabled INTEGER, timezone TEXT, description TEXT, + owner_id INTEGER, created_at TEXT, updated_at TEXT, + last_run_at TEXT, next_run_at TEXT, model TEXT, deleted_at TEXT + ) + """ + ) + conn.execute( + """ + CREATE TABLE schedule_executions ( + id TEXT PRIMARY KEY, schedule_id TEXT, agent_name TEXT, status TEXT, + started_at TEXT, completed_at TEXT, duration_ms INTEGER, message TEXT, + response TEXT, error TEXT, triggered_by TEXT, model_used TEXT, + attempt_number INTEGER DEFAULT 1, retry_of_execution_id TEXT, + source_user_id INTEGER, source_user_email TEXT, source_agent_name TEXT, + source_mcp_key_id TEXT, source_mcp_key_name TEXT, + -- Written by update_execution_status. Omitting them made the + -- abandon path raise OperationalError, which its fail-safe + -- swallowed, so the row stayed `running` and the test "found" a bug + -- that was the fixture's. Keep this column set in step with the + -- UPDATE in db.update_execution_status. + context_used INTEGER, context_max INTEGER, cost REAL, + tool_calls TEXT, execution_log TEXT, claude_session_id TEXT + ) + """ + ) + conn.execute( + "INSERT INTO agent_schedules VALUES ('sch-1','a1','nightly','0 3 * * *'," + "'do the thing',1,'UTC',NULL,1,'2026-01-01T00:00:00Z','2026-01-01T00:00:00Z'," + "NULL,NULL,NULL,NULL)" + ) + conn.commit() + conn.close() + + +def _executions(db_path: Path) -> list[dict]: + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + rows = [dict(r) for r in conn.execute("SELECT * FROM schedule_executions")] + conn.close() + return rows + + +def _app(db_path: Path, *, grant_lock: bool = True, ran: list | None = None): + """A SchedulerApp whose service is real but whose lock manager is a fake, + with `_execute_schedule_with_lock` stubbed so no agent is contacted.""" + app = _main_module().SchedulerApp() + service = _service_module().SchedulerService( + database=_database_module().SchedulerDatabase(str(db_path)), + lock_manager=_FakeLockManager(grant=grant_lock), + redis_url="redis://test:test@redis:6379", + ) + + async def _stub(schedule_id, triggered_by="schedule", origin=None, execution=None): + if ran is not None: + ran.append( + { + "schedule_id": schedule_id, + "triggered_by": triggered_by, + "origin": origin, + "execution": execution, + } + ) + + service._execute_schedule_with_lock = _stub + app.scheduler_service = service + return app + + +async def _trigger(app, schedule_id="sch-1", body=None): + """Call the handler and let the spawned background task finish.""" + response = await app._trigger_handler(_FakeRequest(schedule_id, body)) + # The handler fires the run via create_task; give it a turn to complete so + # lock-release assertions see the final state. + await asyncio.sleep(0) + await asyncio.sleep(0) + return response + + +def _payload(response) -> dict: + return json.loads(response.body.decode()) + + +# --------------------------------------------------------------------------- +# The headline defect. +# --------------------------------------------------------------------------- + + +def test_trigger_returns_a_real_execution_id(tmp_path): + """The bug, directly: the response had no id because it was sent before the + row existed.""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path) + + response = asyncio.run(_trigger(app)) + body = _payload(response) + + assert response.status == 200 + assert body["status"] == "triggered" + assert body.get("execution_id"), "the response still carries no execution_id" + + rows = _executions(db_path) + assert len(rows) == 1 + assert body["execution_id"] == rows[0]["id"], ( + "the returned id does not name the row that was created" + ) + + +def test_the_returned_id_is_usable_before_the_run_finishes(tmp_path): + """The id must be valid at RESPONSE time, not eventually. + + The whole point is that a caller can poll it. If the row only appeared once + the background task got around to it, the id would name nothing for a + while, and a fast poller would 404 — a subtler version of the same bug. + """ + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path) + + async def _drive(): + # Deliberately NOT awaiting the background task before looking. + response = await app._trigger_handler(_FakeRequest("sch-1")) + return _payload(response), _executions(db_path) + + body, rows = asyncio.run(_drive()) + + assert len(rows) == 1, "the row did not exist by the time the caller was answered" + assert rows[0]["id"] == body["execution_id"] + assert rows[0]["status"] == "running" + assert rows[0]["triggered_by"] == "manual" + + +def test_exactly_one_execution_row_per_trigger(tmp_path): + """Handler-creates + service-creates would be two rows for one trigger, and + the caller's id would name the one that never runs.""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + ran = [] + app = _app(db_path, ran=ran) + + asyncio.run(_trigger(app)) + + assert len(_executions(db_path)) == 1, "one trigger produced more than one row" + # And the run was handed the row rather than left to make its own. + assert ran and ran[0]["execution"] is not None + + +def test_service_does_not_recreate_a_passed_execution(tmp_path): + """The other half of the same invariant, at the service boundary.""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + db = _database_module().SchedulerDatabase(str(db_path)) + service = _service_module().SchedulerService( + database=db, + lock_manager=_FakeLockManager(grant=True), + redis_url="redis://test:test@redis:6379", + ) + + pre = db.create_execution( + schedule_id="sch-1", agent_name="a1", message="m", triggered_by="manual" + ) + + dispatched = [] + + async def _no_dispatch(schedule, execution, message, triggered_by): + dispatched.append(execution.id) + + service._dispatch_and_record_outcome = _no_dispatch + service._publish_event = lambda *a, **k: asyncio.sleep(0) + + asyncio.run( + service._execute_schedule_with_lock( + "sch-1", triggered_by="manual", execution=pre + ) + ) + + rows = _executions(db_path) + assert len(rows) == 1, "the service created a second row for a pre-created run" + assert dispatched == [pre.id], "the pre-created row was not the one dispatched" + + +# --------------------------------------------------------------------------- +# Honest suppression. +# --------------------------------------------------------------------------- + + +def test_lock_denied_returns_409_not_a_false_triggered(tmp_path): + """A suppressed trigger used to be byte-identical to a real one.""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path, grant_lock=False) + + response = asyncio.run(_trigger(app)) + body = _payload(response) + + assert response.status == 409 + assert body["status"] == "already_running" + assert "execution_id" not in body, ( + "a suppressed trigger must not hand back an id — nothing was started" + ) + + +def test_lock_denied_creates_no_execution_row(tmp_path): + """Nothing ran, so nothing may be recorded as running. (Auditing the + *cron*-side suppression is #1969's job and uses a `skipped` row; this path + simply must not invent a `running` one.)""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path, grant_lock=False) + + asyncio.run(_trigger(app)) + + assert _executions(db_path) == [] + + +def test_lock_denied_does_not_run_the_schedule(tmp_path): + db_path = tmp_path / "t.db" + _seed_db(db_path) + ran = [] + app = _app(db_path, grant_lock=False, ran=ran) + + asyncio.run(_trigger(app)) + + assert ran == [] + + +# --------------------------------------------------------------------------- +# Lock hygiene — the handler now holds a lock across a DB write. +# --------------------------------------------------------------------------- + + +def test_lock_released_exactly_once_on_success(tmp_path): + """Released twice, a lock re-acquired by the next run in between would be + freed out from under it.""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path) + + asyncio.run(_trigger(app)) + + assert app.scheduler_service.lock_manager.lock.release_count == 1 + + +def test_lock_released_when_the_run_raises(tmp_path): + """A failing run must not strand the lock until its Redis TTL.""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path) + + async def _boom(*args, **kwargs): + raise RuntimeError("agent exploded") + + app.scheduler_service._execute_schedule_with_lock = _boom + + asyncio.run(_trigger(app)) + + assert app.scheduler_service.lock_manager.lock.release_count == 1 + + +def test_lock_released_when_the_row_cannot_be_created(tmp_path): + """The new failure window: the lock is held across `create_execution`. If + that write fails we must release before returning, or the schedule is + wedged until the TTL for a run that never started.""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path) + + def _fail(*args, **kwargs): + raise sqlite3.OperationalError("disk I/O error") + + app.scheduler_service.db.create_execution = _fail + + response = asyncio.run(_trigger(app)) + + assert response.status == 500 + assert app.scheduler_service.lock_manager.lock.release_count == 1 + + +def test_lock_released_when_row_creation_returns_none(tmp_path): + """Same window, non-raising variant.""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path) + app.scheduler_service.db.create_execution = lambda *a, **k: None + + response = asyncio.run(_trigger(app)) + + assert response.status == 500 + assert app.scheduler_service.lock_manager.lock.release_count == 1 + + +def test_no_lock_acquired_for_an_unknown_schedule(tmp_path): + """The 404 gate must stay ahead of the lock — locking a schedule that does + not exist would block nothing and leak a key.""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path) + + response = asyncio.run(_trigger(app, schedule_id="nope")) + + assert response.status == 404 + assert app.scheduler_service.lock_manager.acquire_count == 0 + + +# --------------------------------------------------------------------------- +# A pre-created row must never be stranded `running`. +# --------------------------------------------------------------------------- + + +def test_gate_abort_fails_the_precreated_row(tmp_path): + """The schedule is deleted between the response and the task starting. + + The row already exists and its id is already in the caller's hands. Left + `running` it never terminates — canary E-01's exact signature, and a task + the UI shows forever. + """ + db_path = tmp_path / "t.db" + _seed_db(db_path) + db = _database_module().SchedulerDatabase(str(db_path)) + service = _service_module().SchedulerService( + database=db, + lock_manager=_FakeLockManager(grant=True), + redis_url="redis://test:test@redis:6379", + ) + + pre = db.create_execution( + schedule_id="sch-1", agent_name="a1", message="m", triggered_by="manual" + ) + + conn = sqlite3.connect(db_path) + conn.execute("DELETE FROM agent_schedules WHERE id = 'sch-1'") + conn.commit() + conn.close() + + asyncio.run( + service._execute_schedule_with_lock( + "sch-1", triggered_by="manual", execution=pre + ) + ) + + rows = _executions(db_path) + assert len(rows) == 1 + assert rows[0]["status"] == "failed", ( + "a pre-created row was left running after the run was abandoned (#1968)" + ) + assert rows[0]["error"] + + +def test_abandon_helper_never_raises(tmp_path): + """It runs on abort paths. A raise there would replace a clean abandon with + an exception in a background task nobody awaits.""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + service = _service_module().SchedulerService( + database=_database_module().SchedulerDatabase(str(db_path)), + lock_manager=_FakeLockManager(grant=True), + redis_url="redis://test:test@redis:6379", + ) + + def _fail(*args, **kwargs): + raise sqlite3.OperationalError("gone") + + service.db.update_execution_status = _fail + + class _Row: + id = "exec-x" + + service._abandon_precreated_execution(_Row(), "reason") # must not raise + service._abandon_precreated_execution(None, "reason") # None is a no-op + + +# --------------------------------------------------------------------------- +# #1970 interaction: attribution must survive the move. +# --------------------------------------------------------------------------- + + +def test_precreated_row_still_carries_the_caller_identity(tmp_path): + """`create_execution` moved from the service into the handler. The origin + columns must move with it, or #1970 silently regresses to all-NULL.""" + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path) + + asyncio.run( + _trigger( + app, + body={ + "triggered_by": "manual", + "source_user_id": 7, + "source_user_email": "operator@example.com", + "source_mcp_key_id": "key-abc", + "source_mcp_key_name": "ops laptop", + }, + ) + ) + + row = _executions(db_path)[0] + assert row["source_user_id"] == 7 + assert row["source_user_email"] == "operator@example.com" + assert row["source_mcp_key_id"] == "key-abc" + assert row["source_mcp_key_name"] == "ops laptop" + + +def test_webhook_trigger_still_records_its_trigger_type(tmp_path): + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path) + + response = asyncio.run(_trigger(app, body={"triggered_by": "webhook"})) + + assert _payload(response)["triggered_by"] == "webhook" + assert _executions(db_path)[0]["triggered_by"] == "webhook" + + +# --------------------------------------------------------------------------- +# Backend + MCP relay. +# --------------------------------------------------------------------------- + + +def _backend_trigger_endpoint() -> str: + source = (_REPO / "src" / "backend" / "routers" / "schedules.py").read_text( + encoding="utf-8" + ) + marker = "async def trigger_schedule(" + return source[source.index(marker):][:5000] + + +def test_backend_relays_the_execution_id(): + """The backend passed through five fields and dropped the one the MCP tool + reads.""" + endpoint = _backend_trigger_endpoint() + assert re.search(r'"execution_id":\s*result\.get\("execution_id"\)', endpoint), ( + "the backend does not relay execution_id to its caller (#1968)" + ) + + +def test_backend_maps_409_instead_of_flattening_it_to_500(): + """Without this the honest 409 reaches the caller as 'Failed to trigger + schedule' — a worse lie than the one being fixed, since it claims failure + where the schedule is healthily busy.""" + endpoint = _backend_trigger_endpoint() + assert "409" in endpoint + assert "HTTP_409_CONFLICT" in endpoint + + +def test_mcp_tool_handles_409_without_throwing(): + """An agent should get a decision it can act on, not an exception string.""" + tool = ( + _REPO / "src" / "mcp-server" / "src" / "tools" / "schedules.ts" + ).read_text(encoding="utf-8") + trigger = tool[tool.index("trigger_agent_schedule"):] + assert "error.status === 409" in trigger + assert "already_running" in trigger + + +def test_mcp_tool_does_not_claim_an_id_it_did_not_get(): + """An older backend still omits the field. Interpolating it anyway is + literally the reported bug — do not swap one confident lie for another.""" + tool = ( + _REPO / "src" / "mcp-server" / "src" / "tools" / "schedules.ts" + ).read_text(encoding="utf-8") + trigger = tool[tool.index("trigger_agent_schedule"):] + assert "!result.execution_id" in trigger, ( + "the MCP tool interpolates execution_id unguarded — an old backend " + "still yields 'undefined' (#1968)" + ) + + +def test_trigger_result_type_admits_the_field_can_be_absent(): + """`execution_id: string` (required) is why the compiler never flagged the + `undefined` in the first place — the type asserted a guarantee the wire + never made.""" + types = (_REPO / "src" / "mcp-server" / "src" / "types.ts").read_text( + encoding="utf-8" + ) + block = types[types.index("export interface ScheduleTriggerResult"):][:600] + assert "execution_id?" in block, ( + "ScheduleTriggerResult still declares execution_id as always present" + ) + + +def test_ui_distinguishes_already_running_from_a_failure(): + """The 409 is not the 'nothing was changed — try again' case: a run IS in + flight and retrying hits the same lock.""" + panel = ( + _REPO / "src" / "frontend" / "src" / "components" / "SchedulesPanel.vue" + ).read_text(encoding="utf-8") + trigger = panel[panel.index("async function triggerSchedule("):][:1600] + assert "409" in trigger + assert "already running" in trigger.lower() From 251436b44458c4c9f7a3d0e4be3b8e214b0586ce Mon Sep 17 00:00:00 2001 From: Oleksii Dolhov Date: Tue, 4 Aug 2026 12:21:03 +0300 Subject: [PATCH 3/5] fix(scheduler): keep a strong reference to the spawned trigger task (#1968) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review finding on this PR. The event loop holds only a WEAK reference to a task, so a bare `asyncio.create_task(...)` whose result nobody keeps can be garbage-collected mid-flight — the asyncio docs say so outright. The bare call predates this PR, but this PR changes what it costs. Before, a collected task meant the run silently did not happen. Now the lock is acquired and the execution row created BEFORE the task is spawned, so a collected task strands a `running` execution whose id the caller already holds and pins the schedule's lock until its Redis TTL. Uses the `_inflight` set + `add_done_callback(discard)` shape the #1083 result-callback path already established (`agent_server/services/result_callback.py`), so the set cannot grow without bound. Guarded by `test_the_spawned_task_is_strongly_referenced`, which checks the reference is held in the window BEFORE the task runs — the only window where collection is possible — and released after. Related to #1968 Co-Authored-By: Claude Opus 5 (1M context) --- src/scheduler/main.py | 19 +++++++++- tests/unit/test_1968_trigger_execution_id.py | 37 ++++++++++++++++++++ 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/src/scheduler/main.py b/src/scheduler/main.py index 16e53551d..e157f4e29 100644 --- a/src/scheduler/main.py +++ b/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.""" @@ -266,7 +270,18 @@ async def _trigger_handler(self, request: web.Request) -> web.Response: # Execute in background, handing over the lock we already hold and the # row we already created. - asyncio.create_task( + # + # 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, @@ -275,6 +290,8 @@ async def _trigger_handler(self, request: web.Request) -> web.Response: execution=execution, ) ) + self._inflight_triggers.add(task) + task.add_done_callback(self._inflight_triggers.discard) return web.json_response({ "status": "triggered", diff --git a/tests/unit/test_1968_trigger_execution_id.py b/tests/unit/test_1968_trigger_execution_id.py index f44bff039..c3554fdfc 100644 --- a/tests/unit/test_1968_trigger_execution_id.py +++ b/tests/unit/test_1968_trigger_execution_id.py @@ -406,6 +406,43 @@ def test_lock_released_when_row_creation_returns_none(tmp_path): assert app.scheduler_service.lock_manager.lock.release_count == 1 +def test_the_spawned_task_is_strongly_referenced(tmp_path): + """The event loop keeps only a WEAK reference to a task, so a bare + `create_task(...)` nobody holds can be collected mid-flight (asyncio says + so outright). + + That was survivable before this change — a dropped task meant the run + silently didn't happen. It is not survivable now: the lock and the row are + created BEFORE the task, so a collected task strands a `running` execution + whose id the caller already holds and pins the lock until its TTL. Same + `_inflight` shape as the #1083 result-callback path. + """ + import asyncio + + db_path = tmp_path / "t.db" + _seed_db(db_path) + app = _app(db_path) + + seen = {} + + async def _drive(): + # Look BEFORE the task has run — that is the window where a weakly + # referenced task can be collected. + await app._trigger_handler(_FakeRequest("sch-1")) + seen["held"] = len(app._inflight_triggers) + await asyncio.sleep(0) + await asyncio.sleep(0) + seen["after"] = len(app._inflight_triggers) + + asyncio.run(_drive()) + + assert seen["held"] == 1, "the spawned trigger task is not strongly referenced" + assert seen["after"] == 0, ( + "the done-callback does not discard the task — the set grows without " + "bound for the life of the process" + ) + + def test_no_lock_acquired_for_an_unknown_schedule(tmp_path): """The 404 gate must stay ahead of the lock — locking a schedule that does not exist would block nothing and leak a key.""" From 27f46329a968b97ed4583f75b1a56614d3ecf4cd Mon Sep 17 00:00:00 2001 From: Oleksii Dolhov Date: Tue, 4 Aug 2026 12:23:49 +0300 Subject: [PATCH 4/5] docs(learnings): record the create_task-owns-state class surfaced by #1968 review Co-Authored-By: Claude Opus 5 (1M context) --- docs/memory/learnings.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/memory/learnings.md b/docs/memory/learnings.md index a05947612..55da32ea0 100644 --- a/docs/memory/learnings.md +++ b/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. From 8cbfc1f48131404ea1429860722af669b1e9c33b Mon Sep 17 00:00:00 2001 From: Oleksii Dolhov Date: Wed, 5 Aug 2026 12:50:27 +0300 Subject: [PATCH 5/5] fix(scheduler): remove the duplicate ExecutionOrigin my rebase introduced MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit My rebase of this branch onto dev left `src/scheduler/models.py` with TWO `class ExecutionOrigin` definitions: this PR's at line 110 and dev's copy (#1974, as merged) at 187. Python keeps the last one, so the second shadowed the first and both of this PR's fixes became dead code — the None-comparing `is_empty` and the SQLite-range guard on `source_user_id`. Three `test_execution_origin_properties` cases went red on head, correctly. Git did not flag it. Both sides added the class at different offsets, so the textual auto-merge took both hunks and reported no conflict — a semantic duplicate that only a reader or an importer would notice. Two things on my side let it through: - I re-read only the files git marked as conflicted, not the whole merged result of a 4-commit rebase. - My post-rebase check was `-k "1968 or ent326 or timeline or scheduler or executions"`, which does not match `test_execution_origin_properties`. The filter was narrower than the blast radius, so 129 tests passed and said nothing about the file I had just broken. Kept the first copy: it is a strict superset (identical fields, `is_empty` comparing against None so `user_id=0` is not reported empty, and the `_SQLITE_INT_MIN/MAX` range check that stops an out-of-range id reaching the INSERT and taking the dispatch down). Deleted dev's older copy. Swept the other six branches I rebased for the same class of damage — duplicate top-level defs in any changed .py — all clean. 350 passed across origin/scheduler/1968/1969/1970/execution. Co-Authored-By: Claude Opus 5 (1M context) --- src/scheduler/models.py | 58 ----------------------------------------- 1 file changed, 58 deletions(-) diff --git a/src/scheduler/models.py b/src/scheduler/models.py index c83c78e3e..4c97294f4 100644 --- a/src/scheduler/models.py +++ b/src/scheduler/models.py @@ -183,64 +183,6 @@ def _str(key: str) -> Optional[str]: ) -@dataclass -class ExecutionOrigin: - """Who initiated an execution (AUDIT-001, #1970). - - One value object instead of five parallel parameters threaded through - ``_trigger_handler → _execute_manual_trigger → _execute_schedule_with_lock - → create_execution`` — five positional siblings at four call depths is how - one of them silently stops being forwarded. - - Attribution only; **nothing authorizes on these fields.** An all-``None`` - origin is the correct, honest record for a cron tick, which has no caller. - """ - user_id: Optional[int] = None - user_email: Optional[str] = None - agent_name: Optional[str] = None - mcp_key_id: Optional[str] = None - mcp_key_name: Optional[str] = None - - def is_empty(self) -> bool: - return not any( - (self.user_id, self.user_email, self.agent_name, - self.mcp_key_id, self.mcp_key_name) - ) - - @classmethod - def from_payload(cls, body: object) -> "ExecutionOrigin": - """Build from an untrusted JSON body (the scheduler's trigger endpoint). - - Validates at the boundary: wrong types are dropped rather than coerced, - and strings are length-capped, so a malformed or oversized payload - cannot write junk into an append-only-in-spirit audit column. A caller - that lies about its identity is not a new exposure — ``triggered_by`` - has been caller-supplied on this same endpoint all along, and the - scheduler is reachable only from the platform network. - """ - if not isinstance(body, dict): - return cls() - - def _str(key: str) -> Optional[str]: - value = body.get(key) - if not isinstance(value, str): - return None - value = value.strip() - return value[:255] or None - - user_id = body.get("source_user_id") - if isinstance(user_id, bool) or not isinstance(user_id, int): - user_id = None - - return cls( - user_id=user_id, - user_email=_str("source_user_email"), - agent_name=_str("source_agent_name"), - mcp_key_id=_str("source_mcp_key_id"), - mcp_key_name=_str("source_mcp_key_name"), - ) - - @dataclass class Reminder: """A durable one-shot agent self-reminder (#1296).