fix(mcp): inbox relay messages say the run's state and why it stopped

apps/server/src/mcp/McpEventRelay.ts

@@ -12,6 +12,7 @@ import {
   ThreadId,
   RunId,
   type McpThreadNotify,
+  type OrchestrationV2DomainEvent,
   mcpClientScopes,
 } from "@t3tools/contracts";
 import * as Clock from "effect/Clock";
@@ -38,6 +39,30 @@ const DELIVERED_RETENTION_MS = 86400_000;
 const denied = () => new McpEventsError({ code: -32001, reason: "access_denied" });
 const storage = () => new McpEventsError({ code: -32603, reason: "storage" });
 const encode = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown));
+
+/**
+ * The relayed event's state, and why a cancelled or interrupted run stopped, as webhook event
+ * data carries them; `server_shutdown` / `server_restart` mark a restart, not a real cancel.
+ */
+export function relayEventState(event: OrchestrationV2DomainEvent): {
+  readonly state?: string;
+  readonly stopReason?: string;
+} {
+  switch (event.type) {
+    case "run.updated": {
+      const { status, stop } = event.payload;
+      return (status === "cancelled" || status === "interrupted") && stop !== undefined
+        ? { state: status, stopReason: stop.reason }
+        : { state: status };
+    }
+    case "runtime-request.updated":
+      return { state: event.payload.status };
+    case "provider-session.detached":
+      return { state: "detached" };
+    default:
+      return {};
+  }
+}
 const Modes = { runtimeMode: RuntimeMode, interactionMode: ProviderInteractionMode };
 const Pin = Schema.Struct({ inboxThreadId: ThreadId, ...Modes });
 const Binding = Schema.Struct({
@@ -252,6 +277,9 @@ const make = Effect.gen(function* () {
         .pipe(Stream.take(256), Stream.runCollect, Effect.mapError(storage));
       let through = Math.max(binding.scanCursor, floor);
       let finished = false;
+      // A failover run continues the bound run's work, so the relay follows it (and its own
+      // failover) and relays that run's result, as the webhook path reports it.
+      let boundRunId = binding.runId;
       for (const stored of batch) {
         through = stored.sequence;
         if (!("event" in stored) || stored.event.threadId !== binding.threadId) continue;
@@ -261,10 +289,23 @@ const make = Effect.gen(function* () {
           event.payload.userMessageId.startsWith(RELAY_MESSAGE_ID_PREFIX)
         )
           continue;
+        if (
+          (event.type === "run.created" || event.type === "run.updated") &&
+          event.payload.failoverOfRunId === boundRunId
+        ) {
+          boundRunId = event.payload.id;
+        }
         const runId = event.type === "run.updated" ? event.payload.id : event.runId;
         const threadLevelRelease =
           event.type === "provider-session.detached" && runId === undefined;
-        if (!threadLevelRelease && runId !== binding.runId) continue;
+        if (!threadLevelRelease && runId !== boundRunId) continue;
+        // A run failing over is not a blocker: its failover run reports the outcome. A skip is.
+        if (
+          event.type === "run.updated" &&
+          event.payload.failover !== undefined &&
+          event.payload.failover.state !== "skipped"
+        )
+          continue;
         const outcome =
           event.type === "run.updated" &&
           ["completed", "failed", "interrupted", "cancelled", "rolled_back"].includes(
@@ -282,14 +323,14 @@ const make = Effect.gen(function* () {
         if (name === undefined || !binding.events.includes(name)) continue;
         const records = yield* threads
           .getThreadRecords(binding.threadId, ["messages"], {
-            messageRunIds: [RunId.make(binding.runId)],
+            messageRunIds: [RunId.make(boundRunId)],
           })
           .pipe(Effect.mapError(storage));
         const text =
           records.messages
-            .filter((message) => message.runId === binding.runId && message.role === "assistant")
+            .filter((message) => message.runId === boundRunId && message.role === "assistant")
             .at(-1)?.text ?? "";
-        const sourceKey = outcome ? `run:${binding.runId}:outcome` : event.id;
+        const sourceKey = outcome ? `run:${boundRunId}:outcome` : event.id;
         yield* enqueue({
           ...binding,
           sourceKey,
@@ -302,7 +343,8 @@ const make = Effect.gen(function* () {
             threadId: binding.threadId,
             threadTitle: excerpt(records.thread.title),
             runId: runId ?? null,
-            ...(threadLevelRelease ? { boundRunId: binding.runId } : {}),
+            ...relayEventState(event),
+            ...(threadLevelRelease ? { boundRunId } : {}),
             ledgerKey: binding.ledgerKey === null ? null : excerpt(binding.ledgerKey),
             occurredAt: DateTime.formatIso(event.occurredAt),
             resultExcerpt: excerpt(text),
@@ -323,7 +365,7 @@ const make = Effect.gen(function* () {
           break;
         }
       }
-      yield* sql`UPDATE mcp_event_relay_bindings SET scan_cursor = ${through}, expires_at = CASE WHEN ${finished ? 1 : 0} = 1 THEN 0 ELSE expires_at END WHERE id = ${binding.id}`.pipe(
+      yield* sql`UPDATE mcp_event_relay_bindings SET scan_cursor = ${through}, run_id = ${boundRunId}, expires_at = CASE WHEN ${finished ? 1 : 0} = 1 THEN 0 ELSE expires_at END WHERE id = ${binding.id}`.pipe(
         Effect.mapError(storage),
       );
       behind ||= !finished && through < latest;

apps/server/src/mcp/McpEvents.test.ts

@@ -399,7 +399,9 @@ it.effect(
           threadTitle: expect.any(String),
           resultExcerpt: "é".repeat(512),
           pullRequestUrls: ["https://github.com/acme/app/pull/1"],
+          state: "failed",
         });
+        expect(body).not.toHaveProperty("stopReason");
         yield* f.call(rebuilt.setInbox({ threadId: null }));
         expect(yield* f.call(rebuilt.prepareNotify({}))).toMatchObject({
           reason: "no_pinned_inbox",
@@ -412,6 +414,89 @@ it.effect(
     ).pipe(Effect.provide(NodeCrypto.layer)),
 );
 
+it.effect("relays a failover run's result, not a blocker for the run that failed over to it", () =>
+  Effect.scoped(
+    Effect.gen(function* () {
+      const f = yield* fixture;
+      yield* f.grantWrite(f.principal);
+      const threadId = ThreadId.make("failover-source");
+      const inbox = ThreadId.make("failover-inbox");
+      const runId = RunId.make("failing-run");
+      const fallbackRunId = RunId.make("fallback-run");
+      yield* f.call(f.service.setInbox({ threadId: inbox }));
+      const notification = yield* f.call(
+        ThreadNotifications.prepare({}).pipe(Effect.provideService(McpEvents.McpEvents, f.service)),
+      );
+      f.projections.set(threadId, idleThreadProjection(liveThreadShell(threadId)));
+      const now = DateTime.makeUnsafe(1);
+      const instanceId = ProviderInstanceId.make("replay");
+      const run = {
+        id: runId,
+        threadId,
+        ordinal: 1,
+        providerInstanceId: instanceId,
+        modelSelection: { instanceId, model: "fixture" },
+        providerThreadId: null,
+        userMessageId: MessageId.make("failover-message"),
+        rootNodeId: null,
+        activeAttemptId: null,
+        status: "failed" as const,
+        requestedAt: now,
+        startedAt: now,
+        completedAt: now,
+        checkpointId: null,
+        contextHandoffId: null,
+      };
+      const failover = {
+        reason: "usage_limit" as const,
+        modelSelection: { instanceId: ProviderInstanceId.make("claude"), model: "opus" },
+      };
+      yield* f.store.appendAgentEvents({
+        events: [
+          {
+            id: EventId.make("failing-pending"),
+            threadId,
+            occurredAt: now,
+            type: "run.updated",
+            payload: { ...run, failover: { ...failover, state: "pending" } },
+          },
+          {
+            id: EventId.make("failing-dispatched"),
+            threadId,
+            occurredAt: now,
+            type: "run.updated",
+            payload: {
+              ...run,
+              failover: { ...failover, state: "dispatched", runId: fallbackRunId },
+            },
+          },
+          {
+            id: EventId.make("fallback-completed"),
+            threadId,
+            occurredAt: now,
+            type: "run.updated",
+            payload: {
+              ...run,
+              id: fallbackRunId,
+              ordinal: 2,
+              status: "completed",
+              failoverOfRunId: runId,
+            },
+          },
+        ],
+      });
+      yield* f.call(ThreadNotifications.finish(notification, threadId, runId, "ledger/failover"));
+      yield* f.service.drain();
+      expect(f.inboxSends).toHaveLength(1);
+      expect(json(f.inboxSends[0]!.text.slice("[T3 event relay]\n".length))).toMatchObject({
+        eventName: "thread.run.result",
+        runId: fallbackRunId,
+        state: "completed",
+      });
+    }),
+  ).pipe(Effect.provide(NodeCrypto.layer)),
+);
+
 it.effect("recovers the source result when an idempotent retry follows a lost binding", () =>
   Effect.scoped(
     Effect.gen(function* () {
@@ -1107,7 +1192,7 @@ it("says why a cancelled run stopped, so a restart cancel does not read as a fai
   };
   const observe = (
     payload: Omit<typeof run, "status"> & {
-      readonly status: "cancelled" | "rolled_back";
+      readonly status: "cancelled" | "interrupted" | "rolled_back";
       readonly stop?: { reason: string; actor: { kind: string } };
     },
   ) =>
@@ -1131,6 +1216,34 @@ it("says why a cancelled run stopped, so a restart cancel does not read as a fai
     observe({ ...run, stop: { reason: "server_shutdown", actor: { kind: "server" } } }),
   ).toMatchObject({ state: "cancelled", stopReason: "server_shutdown" });
   expect(observe(run)).not.toHaveProperty("stopReason");
+  // The inbox relay says the same, so a restart cancel reads as one on that route too.
+  const relayed = (payload: Parameters<typeof observe>[0]) =>
+    Relay.relayEventState({
+      id: EventId.make("restart-event"),
+      threadId,
+      runId: run.id,
+      occurredAt: DateTime.makeUnsafe(0),
+      type: "run.updated",
+      payload,
+    });
+  expect(
+    relayed({ ...run, stop: { reason: "server_shutdown", actor: { kind: "server" } } }),
+  ).toEqual({ state: "cancelled", stopReason: "server_shutdown" });
+  expect(relayed(run)).toEqual({ state: "cancelled" });
+  expect(
+    relayed({
+      ...run,
+      status: "interrupted",
+      stop: { reason: "provider_interrupted", actor: { kind: "server" } },
+    }),
+  ).toEqual({ state: "interrupted", stopReason: "provider_interrupted" });
+  expect(
+    relayed({
+      ...run,
+      status: "rolled_back",
+      stop: { reason: "server_shutdown", actor: { kind: "server" } },
+    }),
+  ).toEqual({ state: "rolled_back" });
   expect(
     observe({
       ...run,