fix(ADA-79): WS keepalive + disconnect-banner debounce + friendly diagnostic

apps/server/src/ws.ts

@@ -114,6 +114,7 @@ import * as PairingGrantStore from "./auth/PairingGrantStore.ts";
 import * as SessionStore from "./auth/SessionStore.ts";
 import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http.ts";
 import * as RelayClient from "@t3tools/shared/relayClient";
+import { makeServerConfigHeartbeatStream, shouldSendServerConfigHeartbeat } from "./wsKeepalive.ts";
 const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError);
 const isOrchestrationScheduledTaskMutationError = Schema.is(
   OrchestrationScheduledTaskMutationError,
@@ -1533,7 +1534,7 @@ const makeWsRpcLayer = (
             ),
             { "rpc.aggregate": "preview" },
           ),
-        [WS_METHODS.subscribeServerConfig]: (_input) =>
+        [WS_METHODS.subscribeServerConfig]: (input) =>
           observeRpcStreamEffect(
             WS_METHODS.subscribeServerConfig,
             Effect.gen(function* () {
@@ -1568,10 +1569,16 @@ const makeWsRpcLayer = (
                 .refresh()
                 .pipe(Effect.ignoreCause({ log: true }), Effect.forkScoped);
 
-              const liveUpdates = Stream.merge(
+              const configUpdates = Stream.merge(
                 keybindingsUpdates,
                 Stream.merge(providerStatuses, settingsUpdates),
               );
+              // Only send keepalive heartbeats to clients that declared support
+              // for the `heartbeat` union variant; older/version-skewed clients
+              // would otherwise fail to decode it and lose the subscription.
+              const liveUpdates = shouldSendServerConfigHeartbeat(input)
+                ? Stream.merge(makeServerConfigHeartbeatStream(), configUpdates)
+                : configUpdates;
 
               return Stream.concat(
                 Stream.make({

apps/server/src/wsKeepalive.test.ts

@@ -0,0 +1,43 @@
+import { describe, expect, it } from "@effect/vitest";
+import * as Duration from "effect/Duration";
+import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
+import * as Stream from "effect/Stream";
+import * as TestClock from "effect/testing/TestClock";
+
+import {
+  makeServerConfigHeartbeatStream,
+  shouldSendServerConfigHeartbeat,
+  WS_KEEPALIVE_INTERVAL_MS,
+} from "./wsKeepalive.ts";
+
+describe("websocket keepalive", () => {
+  it("only sends heartbeats to clients that opted in", () => {
+    expect(shouldSendServerConfigHeartbeat({ supportsHeartbeat: true })).toBe(true);
+    // Older/version-skewed clients omit the flag and must not receive the new
+    // `heartbeat` union variant they cannot decode.
+    expect(shouldSendServerConfigHeartbeat({ supportsHeartbeat: false })).toBe(false);
+    expect(shouldSendServerConfigHeartbeat({})).toBe(false);
+    expect(shouldSendServerConfigHeartbeat({ supportsHeartbeat: undefined })).toBe(false);
+  });
+
+  it.effect("emits application frames well before Cloudflare's idle websocket timeout", () =>
+    Effect.gen(function* () {
+      expect(WS_KEEPALIVE_INTERVAL_MS).toBeLessThan(100_000);
+
+      const eventsFiber = yield* makeServerConfigHeartbeatStream().pipe(
+        Stream.take(2),
+        Stream.runCollect,
+        Effect.forkChild,
+      );
+
+      yield* TestClock.adjust(Duration.millis(WS_KEEPALIVE_INTERVAL_MS));
+      yield* TestClock.adjust(Duration.millis(WS_KEEPALIVE_INTERVAL_MS));
+
+      expect(Array.from(yield* Fiber.join(eventsFiber))).toEqual([
+        { version: 1, type: "heartbeat" },
+        { version: 1, type: "heartbeat" },
+      ]);
+    }).pipe(Effect.provide(TestClock.layer())),
+  );
+});

apps/server/src/wsKeepalive.ts

@@ -0,0 +1,32 @@
+import type { ServerConfigStreamEvent } from "@t3tools/contracts";
+import * as Duration from "effect/Duration";
+import * as Stream from "effect/Stream";
+
+export const WS_KEEPALIVE_INTERVAL_MS = 30_000;
+
+/**
+ * Whether to emit websocket keepalive `heartbeat` frames on this
+ * `subscribeServerConfig` subscription. Heartbeats are a new
+ * `ServerConfigStreamEvent` union variant, so they must only be sent to clients
+ * that declared support — otherwise an older/version-skewed client's schema
+ * decoder fails on the unknown variant and tears down the config subscription.
+ */
+export function shouldSendServerConfigHeartbeat(input: {
+  readonly supportsHeartbeat?: boolean | undefined;
+}): boolean {
+  return input.supportsHeartbeat === true;
+}
+
+export function makeServerConfigHeartbeatEvent(): ServerConfigStreamEvent {
+  return {
+    version: 1,
+    type: "heartbeat",
+  };
+}
+
+export function makeServerConfigHeartbeatStream(): Stream.Stream<ServerConfigStreamEvent> {
+  return Stream.tick(Duration.millis(WS_KEEPALIVE_INTERVAL_MS)).pipe(
+    Stream.drop(1),
+    Stream.map(() => makeServerConfigHeartbeatEvent()),
+  );
+}

apps/web/src/components/ChatView.logic.test.ts

@@ -3,8 +3,10 @@ import { describe, expect, it } from "vite-plus/test";
 
 import type { Thread } from "../types";
 import {
+  ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS,
   MAX_HIDDEN_MOUNTED_PREVIEW_THREADS,
   MAX_HIDDEN_MOUNTED_TERMINAL_THREADS,
+  buildThreadErrorDismissKey,
   buildExpiredTerminalContextToastCopy,
   buildThreadTurnInterruptInput,
   createLocalDispatchSnapshot,
@@ -15,7 +17,9 @@ import {
   hasServerAcknowledgedLocalDispatch,
   reconcileMountedTerminalThreadIds,
   reconcileRetainedMountedThreadIds,
+  resolveVisibleServerThreadError,
   resolveSendEnvMode,
+  shouldShowEnvironmentUnavailableBanner,
   shouldWriteThreadErrorToCurrentServerThread,
   threadHasEstablishedProviderBinding,
 } from "./ChatView.logic";
@@ -209,6 +213,98 @@ describe("buildThreadTurnInterruptInput", () => {
   });
 });
 
+describe("shouldShowEnvironmentUnavailableBanner", () => {
+  it("suppresses short reconnect blips", () => {
+    expect(
+      shouldShowEnvironmentUnavailableBanner({
+        connectionPhase: "reconnecting",
+        unavailableSinceMs: 1_000,
+        nowMs: 1_000 + ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS - 1,
+      }),
+    ).toBe(false);
+  });
+
+  it("shows the banner once the connection has stayed unavailable", () => {
+    expect(
+      shouldShowEnvironmentUnavailableBanner({
+        connectionPhase: "reconnecting",
+        unavailableSinceMs: 1_000,
+        nowMs: 1_000 + ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS,
+      }),
+    ).toBe(true);
+  });
+
+  it("clears immediately when the connection is back", () => {
+    expect(
+      shouldShowEnvironmentUnavailableBanner({
+        connectionPhase: "connected",
+        unavailableSinceMs: 1_000,
+        nowMs: 1_000 + ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS,
+      }),
+    ).toBe(false);
+  });
+});
+
+describe("resolveVisibleServerThreadError", () => {
+  it("uses a dismissed key to hide a persisted server error for that turn", () => {
+    const turnId = TurnId.make("turn-dismissed");
+    const dismissKey = buildThreadErrorDismissKey({
+      threadKey: "environment-local:thread-1",
+      turnId,
+      error: "The turn was interrupted. Send your message again to retry.",
+    });
+
+    expect(dismissKey).not.toBeNull();
+    expect(
+      resolveVisibleServerThreadError({
+        localError: null,
+        sessionError: "The turn was interrupted. Send your message again to retry.",
+        dismissedSessionErrorKeys: { [dismissKey!]: true },
+        threadKey: "environment-local:thread-1",
+        turnId,
+      }),
+    ).toBeNull();
+  });
+
+  it("allows the same friendly interruption message to appear on a later turn", () => {
+    const dismissedTurnId = TurnId.make("turn-dismissed");
+    const laterTurnId = TurnId.make("turn-later");
+    const dismissKey = buildThreadErrorDismissKey({
+      threadKey: "environment-local:thread-1",
+      turnId: dismissedTurnId,
+      error: "The turn was interrupted. Send your message again to retry.",
+    });
+
+    expect(
+      resolveVisibleServerThreadError({
+        localError: null,
+        sessionError: "The turn was interrupted. Send your message again to retry.",
+        dismissedSessionErrorKeys: { [dismissKey!]: true },
+        threadKey: "environment-local:thread-1",
+        turnId: laterTurnId,
+      }),
+    ).toBe("The turn was interrupted. Send your message again to retry.");
+  });
+
+  it("keeps local errors visible even when a persisted session error was dismissed", () => {
+    const dismissKey = buildThreadErrorDismissKey({
+      threadKey: "environment-local:thread-1",
+      turnId: null,
+      error: "Persisted error",
+    });
+
+    expect(
+      resolveVisibleServerThreadError({
+        localError: "Select a base branch before sending.",
+        sessionError: "Persisted error",
+        dismissedSessionErrorKeys: { [dismissKey!]: true },
+        threadKey: "environment-local:thread-1",
+        turnId: null,
+      }),
+    ).toBe("Select a base branch before sending.");
+  });
+});
+
 describe("deriveComposerSendState", () => {
   it("treats expired terminal pills as non-sendable content", () => {
     const state = deriveComposerSendState({

apps/web/src/components/ChatView.logic.ts

@@ -9,6 +9,7 @@ import {
   type ThreadId,
   type TurnId,
 } from "@t3tools/contracts";
+import type { EnvironmentConnectionPresentation } from "@t3tools/client-runtime/connection";
 import { type ChatMessage, type SessionPhase, type Thread } from "../types";
 import { type ComposerImageAttachment, type DraftThreadState } from "../composerDraftStore";
 import * as Schema from "effect/Schema";
@@ -24,6 +25,7 @@ import type { DraftThreadEnvMode } from "../composerDraftStore";
 export const LAST_INVOKED_SCRIPT_BY_PROJECT_KEY = "t3code:last-invoked-script-by-project";
 export const MAX_HIDDEN_MOUNTED_TERMINAL_THREADS = 10;
 export const MAX_HIDDEN_MOUNTED_PREVIEW_THREADS = 3;
+export const ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS = 2_000;
 
 export const LastInvokedScriptByProjectSchema = Schema.Record(ProjectId, Schema.String);
 
@@ -74,6 +76,56 @@ export function shouldWriteThreadErrorToCurrentServerThread(input: {
   );
 }
 
+export function shouldShowEnvironmentUnavailableBanner(input: {
+  connectionPhase: EnvironmentConnectionPresentation["phase"];
+  unavailableSinceMs: number | null;
+  nowMs: number;
+  debounceMs?: number;
+}): boolean {
+  if (input.connectionPhase === "connected" || input.unavailableSinceMs === null) {
+    return false;
+  }
+
+  const debounceMs = input.debounceMs ?? ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS;
+  return input.nowMs - input.unavailableSinceMs >= debounceMs;
+}
+
+export function buildThreadErrorDismissKey(input: {
+  threadKey: string;
+  turnId: TurnId | null | undefined;
+  error: string | null | undefined;
+}): string | null {
+  const error = input.error?.trim();
+  if (!error) {
+    return null;
+  }
+
+  return `${input.threadKey}:${input.turnId ?? "session"}:${error}`;
+}
+
+export function resolveVisibleServerThreadError(input: {
+  localError: string | null;
+  sessionError: string | null;
+  dismissedSessionErrorKeys: Readonly<Record<string, true | undefined>>;
+  threadKey: string;
+  turnId: TurnId | null | undefined;
+}): string | null {
+  if (input.localError !== null) {
+    return input.localError;
+  }
+
+  const dismissKey = buildThreadErrorDismissKey({
+    threadKey: input.threadKey,
+    turnId: input.turnId,
+    error: input.sessionError,
+  });
+  if (dismissKey !== null && input.dismissedSessionErrorKeys[dismissKey]) {
+    return null;
+  }
+
+  return input.sessionError;
+}
+
 export function buildThreadTurnInterruptInput(thread: Pick<Thread, "id" | "session">): {
   threadId: ThreadId;
   turnId?: TurnId;