Merge remote-tracking branch 'origin/fix/watch-mode-keepalive-tri-13065' into feat/query-safety-tri-11165

Author: kathiekiwi

.changeset/watch-mode-keepalive.md

@@ -3,3 +3,5 @@
 ---
 
 Watch-mode chat subscriptions now stay connected across quiet periods.
+
+Read-only chat subscriptions no longer stop a turn when they disconnect.

apps/webapp/test/__snapshots__/reportRenderParity.test.ts.snap

@@ -0,0 +1,64 @@
+/**
+ * The email gate is the one place the installation's mail config decides whether a watch may
+ * subscribe. Both variables are required: with either one missing the channel would be created
+ * but never deliver, so the refusal has to come off the env, not off the caller.
+ */
+
+import { beforeEach, describe, expect, test, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+  env: {} as { ALERT_FROM_EMAIL?: string; ALERT_EMAIL_TRANSPORT?: string },
+  canAccessDashboardAgent: vi.fn(async () => true),
+}));
+
+vi.mock("~/env.server", () => ({ env: mocks.env }));
+vi.mock("~/v3/canAccessDashboardAgent.server", () => ({
+  canAccessDashboardAgent: mocks.canAccessDashboardAgent,
+}));
+vi.mock("~/db.server", () => ({ prisma: {}, $replica: {}, sqlDatabaseSchema: undefined }));
+vi.mock("~/v3/alertsWorker.server", () => ({ alertsWorker: { enqueue: vi.fn() } }));
+vi.mock("~/v3/services/alerts/createAlertChannel.server", () => ({
+  CreateAlertChannelService: class {},
+}));
+vi.mock("~/services/logger.server", () => ({
+  logger: { debug: vi.fn(), error: vi.fn(), warn: vi.fn(), info: vi.fn() },
+}));
+
+import { canUseDashboardAgentEmailAlerts } from "~/services/dashboardAgentWatchAlerts.server";
+
+const PARAMS = {
+  userId: "usr_1",
+  organizationId: "org_1",
+  organizationSlug: "acme",
+  orgFeatureFlags: null,
+  projectId: "proj_1",
+};
+
+describe("canUseDashboardAgentEmailAlerts", () => {
+  beforeEach(() => {
+    mocks.env.ALERT_FROM_EMAIL = "alerts@example.com";
+    mocks.env.ALERT_EMAIL_TRANSPORT = "smtp";
+  });
+
+  test("refuses when only the from address is missing", async () => {
+    mocks.env.ALERT_FROM_EMAIL = undefined;
+
+    await expect(canUseDashboardAgentEmailAlerts(PARAMS)).resolves.toEqual({
+      allowed: false,
+      reason: "email_alerts_not_configured",
+    });
+  });
+
+  test("refuses when only the transport is missing", async () => {
+    mocks.env.ALERT_EMAIL_TRANSPORT = undefined;
+
+    await expect(canUseDashboardAgentEmailAlerts(PARAMS)).resolves.toEqual({
+      allowed: false,
+      reason: "email_alerts_not_configured",
+    });
+  });
+
+  test("allows when both are configured and the base gate passes", async () => {
+    await expect(canUseDashboardAgentEmailAlerts(PARAMS)).resolves.toEqual({ allowed: true });
+  });
+});

apps/webapp/test/dashboardAgentWatchAlertGate.test.ts

@@ -1334,6 +1334,94 @@ describe("TriggerChatTransport", () => {
     });
   });
 
+  describe("reconnectToStream stop-on-abort ownership (TRI-13070)", () => {
+    // A quiet stream: EOF, no records, never settled — the subscription
+    // stays alive (watch mode) so an abort mid-flight exercises the stop path.
+    function quietWatchTransport(): {
+      transport: TriggerChatTransport;
+      appends: () => number;
+    } {
+      let appendCount = 0;
+      global.fetch = vi.fn().mockImplementation(async (url: string | URL) => {
+        const urlStr = typeof url === "string" ? url : url.toString();
+        if (isSessionStreamAppendUrl(urlStr)) {
+          appendCount++;
+          return defaultAppendResponse();
+        }
+        if (isSessionOutSubscribeUrl(urlStr)) return defaultSseResponse([]);
+        throw new Error(`Unexpected URL: ${urlStr}`);
+      });
+      const transport = new TriggerChatTransport({
+        task: "my-chat-task",
+        accessToken: () => "pat",
+        watch: true,
+        sessions: { "chat-own": { publicAccessToken: "p", isStreaming: true } },
+      });
+      return { transport, appends: () => appendCount };
+    }
+
+    it("passive subscriber aborting writes no stop chunk to .in", async () => {
+      vi.useFakeTimers();
+      try {
+        const { transport, appends } = quietWatchTransport();
+        const abort = new AbortController();
+        const stream = await transport.reconnectToStream({
+          chatId: "chat-own",
+          abortSignal: abort.signal,
+        });
+        const drained = drainChunks(stream!);
+        await vi.advanceTimersByTimeAsync(1_000);
+        abort.abort();
+        await drained;
+        await vi.advanceTimersByTimeAsync(1_000);
+        expect(appends()).toBe(0);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+
+    it("owning subscriber with stopOnAbort:true sends a stop chunk on abort", async () => {
+      vi.useFakeTimers();
+      try {
+        const { transport, appends } = quietWatchTransport();
+        const abort = new AbortController();
+        const stream = await transport.reconnectToStream({
+          chatId: "chat-own",
+          abortSignal: abort.signal,
+          stopOnAbort: true,
+        });
+        const drained = drainChunks(stream!);
+        await vi.advanceTimersByTimeAsync(1_000);
+        abort.abort();
+        await drained;
+        await vi.advanceTimersByTimeAsync(1_000);
+        expect(appends()).toBe(1);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+
+    it("abortSignal presence alone (stopOnAbort unset) sends no stop", async () => {
+      vi.useFakeTimers();
+      try {
+        const { transport, appends } = quietWatchTransport();
+        const abort = new AbortController();
+        const stream = await transport.reconnectToStream({
+          chatId: "chat-own",
+          abortSignal: abort.signal,
+        });
+        const drained = drainChunks(stream!);
+        await vi.advanceTimersByTimeAsync(1_000);
+        abort.abort();
+        await drained;
+        await vi.advanceTimersByTimeAsync(1_000);
+        expect(appends()).toBe(0);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+  });
+
   describe("multi-tab coordination", () => {
     it("isReadOnly defaults to false when multiTab is disabled", () => {
       const transport = new TriggerChatTransport({

packages/trigger-sdk/src/v3/chat.test.ts

@@ -873,7 +873,11 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     state.isStreaming = true;
     this.notifySessionChange(chatId, state);
 
-    return this.subscribeToSessionStream(state, abortSignal, chatId, { sinceInSeq: inSeq });
+    // Owning turn: aborting this live send stops the turn the user drives.
+    return this.subscribeToSessionStream(state, abortSignal, chatId, {
+      sinceInSeq: inSeq,
+      sendStopOnAbort: true,
+    });
   };
 
   /**
@@ -1146,6 +1150,13 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     options: {
       chatId: string;
       abortSignal?: AbortSignal | undefined;
+      /**
+       * Whether aborting this subscription sends `{kind:"stop"}` on `.in`.
+       * A subscription ending is not session ownership — a passive/watch
+       * reader unmounting must never stop a turn it doesn't drive. Only
+       * pass `true` from a caller that owns the live turn. @default false
+       */
+      stopOnAbort?: boolean;
     } & ChatRequestOptions
   ): Promise<ReadableStream<UIMessageChunk> | null> => {
     const state = this.sessions.get(options.chatId);
@@ -1163,7 +1174,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
 
     return this.subscribeToSessionStream(state, abortSignal, options.chatId, {
       resumed: true,
-      sendStopOnAbort: !!options.abortSignal,
+      sendStopOnAbort: options.stopOnAbort ?? false,
       // Reconnect-on-reload opts into the server's settled-peek shortcut
       // so the SSE doesn't hang for 60s when no turn is in flight. Active
       // send-a-message paths must keep wait=60 to avoid racing the
@@ -1266,7 +1277,11 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     state.isStreaming = true;
     this.notifySessionChange(chatId, state);
 
-    return this.subscribeToSessionStream(state, undefined, chatId, { sinceInSeq: inSeq });
+    // Owning action: aborting this send stops the turn the user drives.
+    return this.subscribeToSessionStream(state, undefined, chatId, {
+      sinceInSeq: inSeq,
+      sendStopOnAbort: true,
+    });
   };
 
   // -------------------------------------------------------------------------