fix(webapp,run-engine): three follow-ups on Devin review (PR #3754)

Three unresolved Devin threads, all addressed in this commit. Committed
locally only — not pushed.

1. `callWithoutTraceEvents` was inheriting the new `emitRunFailedEvent`
   default of `true`, so its `createFailedTaskRun` call would fire the
   `runFailed` bus emit and the listener would write a ClickHouse
   completion event row with empty `traceId`/`spanId` — orphan row,
   directly contradicting the method's "without trace events" contract.
   Pass `emitRunFailedEvent: false` and enqueue
   `PerformTaskRunAlertsService` directly, mirroring the `call()`
   pattern so customers' ERROR channels still see the failure.

2. The cjson empty-tags defense lived only on `createCancelledRun`, not
   on `engine.trigger`. When the mollifier buffer's mutate-side Lua
   re-serialises a payload (e.g. `append_tags` on a buffered run that
   never had tags), an empty Lua table encodes as `{}` and decodes
   back to a JS object — and the previous `tags.length === 0` check
   passes that object straight to Prisma's `String[]` column.
   Mirror the same `Array.isArray && tags.length > 0 ? tags : undefined`
   guard `createCancelledRun` already uses. The defense is symmetric
   with the existing tested case for createCancelledRun, so the same
   contract holds for the trigger replay path.

3. `runCancelled` handler's `cancelRunEvent` lookup fails for
   buffered-only runs (no primary trace event exists, since the
   mollifier gate skipped `repository.traceEvent` for the
   not-yet-materialised run). The handler's `tryCatch` swallowed the
   error, but the systematic `[runCancelled] Failed to cancel run
   event` log fired on every cancelled buffered run.
   Add `emitRunCancelledEvent: boolean = true` to `createCancelledRun`
   (symmetric with the existing `emitRunFailedEvent` flag on
   `createFailedTaskRun`); drainer handler passes `false`. CANCELED PG
   row still writes; only the trace-event mirror is skipped.

Tests:
- `RunEngine.createCancelledRun > emitRunCancelledEvent: false
  suppresses the bus emit but still writes the CANCELED PG row` —
  pins the new flag's semantics.
- `createDrainerHandler > calls createCancelledRun with
  emitRunCancelledEvent: false (suppresses orphan trace-event log
  noise)` — pins the call site's contract.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: d-cs

apps/webapp/app/runEngine/services/triggerFailedTask.server.ts

@@ -291,7 +291,7 @@ export class TriggerFailedTaskService {
         }
       }
 
-      await this.engine.createFailedTaskRun({
+      const failedRun = await this.engine.createFailedTaskRun({
         friendlyId: failedRunFriendlyId,
         environment: {
           id: opts.environmentId,
@@ -313,8 +313,32 @@ export class TriggerFailedTaskService {
         depth,
         resumeParentOnCompletion: opts.resumeParentOnCompletion,
         batch: opts.batch,
+        // Suppress the engine's `runFailed` bus emit — the listener
+        // (`runEngineHandlers.server.ts` `runFailed`) calls
+        // `completeFailedRunEvent`, which writes a ClickHouse trace event
+        // row keyed on (traceId, spanId). This caller has no trace
+        // context (the method name is literally `callWithoutTraceEvents`)
+        // so the emit would write a row with empty traceId/spanId —
+        // orphan event in the store. We still want alert coverage,
+        // though, so enqueue directly below.
+        emitRunFailedEvent: false,
       });
 
+      // Alerts side of `runFailed` — the engine emit was suppressed
+      // above so we don't create an orphan trace event; enqueue the
+      // alert directly so customers' ERROR channels still see the
+      // failure. Best-effort, mirroring the `call()` path.
+      try {
+        await PerformTaskRunAlertsService.enqueue(failedRun.id);
+      } catch (alertsError) {
+        logger.warn("TriggerFailedTaskService.callWithoutTraceEvents: alert enqueue failed", {
+          taskId: opts.taskId,
+          friendlyId: failedRun.friendlyId,
+          error:
+            alertsError instanceof Error ? alertsError.message : String(alertsError),
+        });
+      }
+
       return failedRunFriendlyId;
     } catch (createError) {
       logger.error("TriggerFailedTaskService: failed to create pre-failed TaskRun (no trace)", {

apps/webapp/app/v3/mollifier/mollifierDrainerHandler.server.ts

@@ -53,11 +53,13 @@ export function createDrainerHandler(deps: {
           })
         : context.active();
 
-    // Cancel-wins-over-trigger. If a cancel API call
-    // landed on this entry while it was QUEUED, the snapshot carries
-    // `cancelledAt` + `cancelReason`. Skip the normal materialise path
-    // and write a CANCELED PG row directly. The existing runCancelled
-    // handler writes the TaskEvent.
+    // Cancel-wins-over-trigger. If a cancel API call landed on this
+    // entry while it was QUEUED, the snapshot carries `cancelledAt` +
+    // `cancelReason`. Skip the normal materialise path and write a
+    // CANCELED PG row directly. The `runCancelled` bus emit is
+    // suppressed here because a buffered-only run never had a primary
+    // trace event written for it — the runCancelled handler's
+    // `cancelRunEvent` lookup would fail and log noise per cancel.
     const cancelledAtStr =
       typeof input.payload.cancelledAt === "string" ? input.payload.cancelledAt : undefined;
     if (cancelledAtStr) {
@@ -79,6 +81,7 @@ export function createDrainerHandler(deps: {
                 snapshot: input.payload as any,
                 cancelledAt: new Date(cancelledAtStr),
                 cancelReason,
+                emitRunCancelledEvent: false,
               },
               deps.prisma,
             );

apps/webapp/test/mollifierDrainerHandler.test.ts

@@ -222,6 +222,45 @@ describe("createDrainerHandler", () => {
     expect(createFailedTaskRun).toHaveBeenCalledOnce();
   });
 
+  it("calls createCancelledRun with emitRunCancelledEvent: false (suppresses orphan trace-event log noise)", async () => {
+    // Buffered-only runs never had a primary trace event written for
+    // them — the mollifier gate skipped `repository.traceEvent` since
+    // the run hadn't materialised in PG yet. The `runCancelled` handler
+    // would log `[runCancelled] Failed to cancel run event` for every
+    // cancelled buffered run if we let the emit fire. Suppress it.
+    const friendlyId = RunId.generate().friendlyId;
+    const createCancelledRun = vi.fn(async () => ({
+      id: "internal",
+      friendlyId,
+      status: "CANCELED",
+    }));
+    const handler = createDrainerHandler({
+      engine: { createCancelledRun } as any,
+      prisma: {} as any,
+    });
+
+    await handler({
+      runId: friendlyId,
+      envId: "env_a",
+      orgId: "org_1",
+      payload: {
+        friendlyId,
+        taskIdentifier: "t",
+        environment: envFixture,
+        cancelledAt: new Date().toISOString(),
+        cancelReason: "Canceled by user",
+      },
+      attempts: 0,
+      createdAt: new Date(),
+    } as any);
+
+    expect(createCancelledRun).toHaveBeenCalledOnce();
+    const arg = createCancelledRun.mock.calls[0][0] as {
+      emitRunCancelledEvent?: boolean;
+    };
+    expect(arg.emitRunCancelledEvent).toBe(false);
+  });
+
   it("honours the cancel when a buffered cancel races a materialised non-CANCELED row", async () => {
     // Cancel-wins-over-trigger. If the normal trigger
     // replay path materialised a live PENDING row before the cancel

internal-packages/run-engine/src/engine/index.ts

@@ -458,25 +458,45 @@ export class RunEngine {
    * Skips: queue insertion (no execution), waitpoint creation (the
    * mollifier gate refuses to buffer triggerAndWait children, so a
    * cancelled buffered run never has a waiting parent to unblock),
-   * concurrency reservation. Emits `runCancelled` so the existing
-   * TaskEvent handler writes the cancellation event row — the only side
-   * effect PG-side cancel has today per audit.
+   * concurrency reservation. Emits `runCancelled` by default — callers
+   * working on buffered-only runs (no primary trace event exists) can
+   * opt out via `emitRunCancelledEvent: false` to avoid the systematic
+   * "Failed to cancel run event" noise the handler would log when its
+   * `cancelRunEvent` call can't find a span.
    *
    * Idempotent: if a row with the same friendlyId already exists (double
    * drainer pop after requeue), Prisma's P2002 unique-constraint violation
    * is caught and the existing row is returned. The duplicate runCancelled
    * emission is skipped — the original drain's emit already wrote the
-   * TaskEvent.
+   * TaskEvent (when applicable).
    */
   async createCancelledRun(
     {
       snapshot,
       cancelledAt,
       cancelReason,
+      emitRunCancelledEvent = true,
     }: {
       snapshot: TriggerParams;
       cancelledAt: Date;
       cancelReason: string;
+      /**
+       * Whether to emit the `runCancelled` engine-bus event. Defaults to
+       * true.
+       *
+       * Set to `false` for buffered-only runs that never had a primary
+       * trace event written (the mollifier gate never called
+       * `repository.traceEvent` for them). The `runCancelled` handler in
+       * `runEngineHandlers.server.ts` calls `cancelRunEvent`, which
+       * looks up the run's primary span in the event store — for
+       * buffered-only runs that span doesn't exist, so the lookup fails,
+       * the handler's `tryCatch` swallows it, and a "[runCancelled]
+       * Failed to cancel run event" error is logged for every cancelled
+       * buffered run. Suppressing the emit avoids that systematic noise.
+       * The CANCELED PG row is still written; only the trace-event
+       * mirror is skipped.
+       */
+      emitRunCancelledEvent?: boolean;
     },
     tx?: PrismaClientOrTransaction,
   ): Promise<TaskRun> {
@@ -567,24 +587,26 @@ export class RunEngine {
           },
         });
 
-        this.eventBus.emit("runCancelled", {
-          time: cancelledAt,
-          run: {
-            id: taskRun.id,
-            status: taskRun.status,
-            friendlyId: taskRun.friendlyId,
-            spanId: taskRun.spanId,
-            taskEventStore: taskRun.taskEventStore,
-            createdAt: taskRun.createdAt,
-            completedAt: taskRun.completedAt,
-            error,
-            updatedAt: taskRun.updatedAt,
-            attemptNumber: taskRun.attemptNumber ?? 0,
-          },
-          organization: { id: snapshot.environment.organization.id },
-          project: { id: snapshot.environment.project.id },
-          environment: { id: snapshot.environment.id },
-        });
+        if (emitRunCancelledEvent) {
+          this.eventBus.emit("runCancelled", {
+            time: cancelledAt,
+            run: {
+              id: taskRun.id,
+              status: taskRun.status,
+              friendlyId: taskRun.friendlyId,
+              spanId: taskRun.spanId,
+              taskEventStore: taskRun.taskEventStore,
+              createdAt: taskRun.createdAt,
+              completedAt: taskRun.completedAt,
+              error,
+              updatedAt: taskRun.updatedAt,
+              attemptNumber: taskRun.attemptNumber ?? 0,
+            },
+            organization: { id: snapshot.environment.organization.id },
+            project: { id: snapshot.environment.project.id },
+            environment: { id: snapshot.environment.id },
+          });
+        }
 
         return taskRun;
       } catch (err) {
@@ -819,7 +841,16 @@ export class RunEngine {
               priorityMs,
               queueTimestamp: queueTimestamp ?? delayUntil ?? new Date(),
               ttl: resolvedTtl,
-              runTags: tags.length === 0 ? undefined : tags,
+              // Defensive: when the mollifier drainer replays a buffered
+              // snapshot whose payload was rewritten by a buffer-side Lua
+              // mutate (e.g. append_tags clears an empty list), cjson
+              // encodes an empty Lua table as `{}` rather than `[]`. JS
+              // parses that back as an empty object, and `{}.length` is
+              // undefined — the original `tags.length === 0` check would
+              // pass `{}` straight to Prisma's `String[]` column. Mirror
+              // the same Array.isArray guard that `createCancelledRun`
+              // uses for symmetry with the trigger replay path.
+              runTags: Array.isArray(tags) && tags.length > 0 ? tags : undefined,
               oneTimeUseToken,
               parentTaskRunId,
               rootTaskRunId,

internal-packages/run-engine/src/engine/tests/createCancelledRun.test.ts

@@ -144,6 +144,57 @@ describe("RunEngine.createCancelledRun", () => {
     },
   );
 
+  containerTest(
+    "emitRunCancelledEvent: false suppresses the bus emit but still writes the CANCELED PG row",
+    async ({ prisma, redisOptions }) => {
+      // The mollifier drainer passes `emitRunCancelledEvent: false` for
+      // buffered-only runs because the runCancelled handler's
+      // `cancelRunEvent` lookup fails for them (no primary trace event
+      // span exists — the mollifier gate never called
+      // `repository.traceEvent` for this run). Without the gate, every
+      // cancelled buffered run produces a `[runCancelled] Failed to
+      // cancel run event` error log. This pins the gate's contract: PG
+      // row still lands, bus emit suppressed.
+      const env = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const engine = new RunEngine({ prisma, ...baseEngineOptions(redisOptions) });
+      const captured: EventBusEventArgs<"runCancelled">[0][] = [];
+      engine.eventBus.on("runCancelled", (event) => {
+        captured.push(event);
+      });
+
+      try {
+        const friendlyId = freshRunId();
+        const result = await engine.createCancelledRun({
+          snapshot: {
+            friendlyId,
+            environment: env,
+            taskIdentifier: "test-task",
+            payload: "{}",
+            payloadType: "application/json",
+            context: {},
+            traceContext: {},
+            traceId: "0000000000000000eeee000000000000",
+            spanId: "ffff000000000000",
+            queue: "task/test-task",
+            isTest: false,
+            tags: [],
+          },
+          cancelledAt: new Date(),
+          cancelReason: "Test cancel (silent emit)",
+          emitRunCancelledEvent: false,
+        });
+
+        // PG row still lands.
+        expect(result.status).toBe("CANCELED");
+        expect(result.friendlyId).toBe(friendlyId);
+        // Bus emit suppressed.
+        expect(captured).toHaveLength(0);
+      } finally {
+        await engine.quit();
+      }
+    },
+  );
+
   containerTest(
     "idempotent on double-pop: second call returns existing row without re-emitting",
     async ({ prisma, redisOptions }) => {