feat(webapp): session.out wait=0 + X-Session-Settled on settled tail

/realtime/v1/sessions/:session/:io=out now peeks the tail record in S2
at connection time. When the tail chunk is trigger:turn-complete, the
agent has finished a turn and is either idle-waiting on .in or has
exited — either way no more chunks will arrive without further user
action. In that case the downstream S2 read switches to wait=0 so the
SSE drains and closes in ~1s instead of long-polling for 60s, and the
response carries X-Session-Settled: true so the client can tell the
close is terminal rather than a normal 60s cycle.

Mid-turn tails (streaming UIMessageChunks in flight) fall through to
the existing wait=60 long-poll. Crashed-mid-turn is indistinguishable
from live-streaming at this point and gets the same 60s retry loop as
today — that's a separate hardening, not in scope here.

The peek uses GET /records?tail_offset=1&count=1&wait=0 (single-digit
ms on S2), then unwraps the agent-side envelope written by
StreamsWriterV2: record.body parses to {data: <chunk>, id}, where
<chunk> is the raw UIMessageChunk object. No double-parse on data.

404 / 416 from the peek (stream never written / empty stream) short-
circuit to settled=false so first-connect on a freshly-created session
keeps the long-poll semantics the agent's first chunks depend on.

Verified end-to-end against an idle chat-agent-smoke session: caught-
up reconnect (Last-Event-ID = tail) closes in 1.08s with the header;
behind reconnect (Last-Event-ID < tail) drains remaining records then
closes in 0.94s with the header; empty-stream reconnect keeps the 60s
long-poll behavior unchanged.

Author: ericallam

.server-changes/session-out-settled-signal.md

@@ -0,0 +1,8 @@
+---
+area: webapp
+type: improvement
+---
+
+`/realtime/v1/sessions/:session/out` now peeks the tail record in S2 at connection time. If the last chunk is `trigger:turn-complete` (agent finished a turn and is either idle-waiting on `.in` or has exited), the downstream S2 read uses `wait=0` so the SSE drains and closes immediately instead of holding the connection open for 60s. The response also carries `X-Session-Settled: true` so the client can tell the close is terminal rather than a normal long-poll cycle.
+
+Lets `TriggerChatTransport.reconnectToStream` return quickly on page reloads of settled chats without requiring callers to persist an `isStreaming` flag — the server decides from the stream's own tail. Mid-turn tails still take the 60s long-poll path unchanged.

apps/webapp/app/services/realtime/s2realtimeStreams.server.ts

@@ -293,6 +293,18 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
   /**
    * Serve SSE from a `Session`-primitive channel addressed by
    * `(friendlyId, io)`.
+   *
+   * For `io=out`, peek the tail record first. If it's
+   * `trigger:turn-complete`, the agent has finished a turn and is
+   * either idle-waiting on `.in` or has exited — either way, no more
+   * chunks will arrive without further user action. We switch the
+   * downstream S2 read to `wait=0` (drain whatever's left, close fast)
+   * and set `X-Session-Settled: true` so the client knows this SSE
+   * close is terminal instead of the normal 60s long-poll cycle.
+   *
+   * Mid-turn tail (streaming UIMessageChunk) falls through to the
+   * long-poll path; a crashed-mid-turn stream is indistinguishable
+   * here and behaves like today (client sees wait=60 close, retries).
    */
   async streamResponseFromSessionStream(
     request: Request,
@@ -301,7 +313,98 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     signal: AbortSignal,
     options?: StreamResponseOptions
   ): Promise<Response> {
-    return this.#streamResponseByName(this.toSessionStreamName(friendlyId, io), signal, options);
+    const s2Stream = this.toSessionStreamName(friendlyId, io);
+
+    let waitSeconds = options?.timeoutInSeconds ?? this.s2WaitSeconds;
+    let settled = false;
+
+    if (io === "out") {
+      const lastChunk = await this.#peekLastChunkBody(s2Stream);
+      if (
+        lastChunk != null &&
+        typeof lastChunk === "object" &&
+        (lastChunk as { type?: unknown }).type === "trigger:turn-complete"
+      ) {
+        settled = true;
+        waitSeconds = 0;
+      }
+    }
+
+    const s2Response = await this.#streamResponseByName(s2Stream, signal, {
+      ...options,
+      timeoutInSeconds: waitSeconds,
+    });
+
+    if (!settled) return s2Response;
+
+    const headers = new Headers(s2Response.headers);
+    headers.set("X-Session-Settled", "true");
+    return new Response(s2Response.body, {
+      status: s2Response.status,
+      statusText: s2Response.statusText,
+      headers,
+    });
+  }
+
+  async #peekLastChunkBody(s2Stream: string): Promise<unknown | null> {
+    const qs = new URLSearchParams();
+    // `tail_offset=1` reads one record before the next seq — i.e. the
+    // most recently appended record. `count=1` caps it to just that
+    // record. `wait=0` returns immediately with no long-poll.
+    qs.set("tail_offset", "1");
+    qs.set("count", "1");
+    qs.set("wait", "0");
+
+    let res: Response;
+    try {
+      res = await fetch(
+        `${this.baseUrl}/streams/${encodeURIComponent(s2Stream)}/records?${qs}`,
+        {
+          method: "GET",
+          headers: {
+            Authorization: `Bearer ${this.token}`,
+            Accept: "application/json",
+            "S2-Format": "raw",
+            "S2-Basin": this.basin,
+          },
+        }
+      );
+    } catch (err) {
+      this.logger.warn("S2 peek last record: fetch failed", { err, stream: s2Stream });
+      return null;
+    }
+
+    if (!res.ok) {
+      // 404: stream has never been written to. 416: range not
+      // satisfiable (empty stream). Both mean "nothing to peek."
+      if (res.status === 404 || res.status === 416) return null;
+      const text = await res.text().catch(() => "");
+      this.logger.warn("S2 peek last record failed", {
+        status: res.status,
+        statusText: res.statusText,
+        text,
+        stream: s2Stream,
+      });
+      return null;
+    }
+
+    try {
+      const json = (await res.json()) as {
+        records?: Array<{ body: string; seq_num: number; timestamp: number }>;
+      };
+      const record = json.records?.[0];
+      if (!record) return null;
+      // The record body is a JSON string `{data: <chunk>, id: partId}`
+      // where `<chunk>` is the raw UIMessageChunk object (see
+      // `StreamsWriterV2` — the agent-side writer serializes the chunk
+      // object directly, not double-encoded). Unwrap the envelope and
+      // return `data` as-is.
+      const envelope = JSON.parse(record.body) as { data: unknown; id: string };
+      return envelope.data;
+    } catch (err) {
+      this.logger.warn("S2 peek last record: parse failed", { err, stream: s2Stream });
+      return null;
+    }
   }
 
   async #streamResponseByName(