Sync ordered assistant output changes

auth.py

@@ -114,10 +114,10 @@ def _oauth_token_expired(info, *, now=None) -> bool:
         logger.warning("Unexpected OAuth expiry value; requiring a fresh login.")
         return True
     if expires_at.tzinfo is None:
-        expires_at = expires_at.replace(tzinfo=timezone.utc)
+        expires_at = expires_at.astimezone(timezone.utc)
     current = now or datetime.now(timezone.utc)
     if current.tzinfo is None:
-        current = current.replace(tzinfo=timezone.utc)
+        current = current.astimezone(timezone.utc)
     return expires_at <= current + _OAUTH_EXPIRY_SKEW
 
 

main.js

@@ -17,6 +17,7 @@
  */
 
 import { S2sWsRealtimeClient } from "./ws/s2s-ws-client.js";
+import { ToolCallBatcher } from "./tool-call-batcher.js";
 import { $, truncateError, DEBUG } from "./ui/dom.js";
 import { ChatView } from "./ui/chat.js";
 import { Account } from "./ui/account.js";
@@ -767,14 +768,14 @@ function flashPreview() {
 }
 
 // ── Tool executor ─────────────────────────────────────────────────────────
-// Runs the function the model called, returns the result, and asks for a
-// response so the model speaks it. Errors come back as the tool output too, so
-// the model can recover gracefully instead of the turn stalling.
+// Runs the function the model called and returns the result. The connection's
+// ToolCallBatcher sends every result from the originating response together,
+// then requests one follow-up. Errors come back as tool output too, so the
+// model can recover gracefully instead of the turn stalling.
 
 /**
- * Run the function the model called, return its result to the backend, and ask
- * for a follow-up response. We also hand the result back to the caller so it
- * can be shown in the conversation once the tool has actually run.
+ * Run the function the model called. The caller batches the returned result
+ * with the other calls from the same response before sending it to the model.
  * @param {string} name @param {string} argsJson @param {string} callId
  * @returns {Promise<{ output: string, image?: string }>}
  */
@@ -792,37 +793,23 @@ async function runTool(name, argsJson, callId) {
     if (name === "web_search") {
       const query = typeof args.query === "string" ? args.query : "";
       result.output = await execWebSearch(query);
-      // Return the result and let the bare response.create (below) trigger the
-      // spoken answer.
-      client.sendToolOutput(callId, result.output);
     } else if (name === "camera_snapshot") {
       const dataUrl = captureSnapshot();
       if (dataUrl) {
         if (DEBUG) console.debug(`[tool] camera_snapshot captured frame (${dataUrl.length} chars), sending image + output`);
         result = { output: "Snapshot captured from the webcam and attached as an image.", image: dataUrl };
-        // Return the tool output; the frame itself rides along with the
-        // response.create below (sent right before it), so the model sees the
-        // snapshot in the very response it's about to speak.
-        client.sendToolOutput(callId, result.output);
         flashPreview();
       } else {
         console.warn("[tool] camera_snapshot: no frame — camera off or not ready");
         result.output = "The camera is not available right now.";
-        client.sendToolOutput(callId, result.output);
       }
     } else {
       result.output = `Unknown tool: ${name}`;
-      client.sendToolOutput(callId, result.output);
     }
   } catch (err) {
     const msg = err instanceof Error ? err.message : String(err);
     result.output = `Tool failed: ${msg}`;
-    client.sendToolOutput(callId, result.output);
   }
-  if (DEBUG) console.debug(`[tool] requesting model response after ${name}`);
-  // Camera: the captured frame rides with the response.create (sent just before
-  // it) so it's in context for the reply. Other tools: a bare create.
-  client.requestResponse(result.image ? { image: result.image } : undefined);
   return result;
 }
 
@@ -1337,6 +1324,18 @@ async function doStart(audioContext = null) {
   client = c;
   c.setMuted(micMuted || userAudioReplaying);
 
+  /** @param {{ callId: string; output: string; image?: string }[]} results */
+  const sendToolBatch = (results) => {
+    if (client !== c) return;
+    for (const result of results) c.sendToolOutput(result.callId, result.output);
+    for (const result of results) {
+      if (result.image) c.sendUserImage(result.image);
+    }
+    if (DEBUG) console.debug(`[tool] requesting one follow-up after ${results.length} tool result(s)`);
+    c.requestResponse();
+  };
+  const toolBatches = new ToolCallBatcher(sendToolBatch);
+
   c.addEventListener("queue", (e) => {
     const { position, queueId } = /** @type {CustomEvent<{ position: number; queueId: string }>} */ (e).detail;
     if (queueId) queuedTicketId = queueId;
@@ -1381,16 +1380,24 @@ async function doStart(audioContext = null) {
   c.addEventListener("response-finished", (e) => {
     const detail = /** @type {CustomEvent<{ responseId: string; status: string; audible?: boolean; transcript?: string }>} */ (e).detail;
     chat.onResponseFinished(detail);
+    const flush = toolBatches.finish(detail.responseId, detail.status);
+    if (flush) void flush.catch((err) => onFatalError(err));
   });
 
   c.addEventListener("toolcall", (e) => {
-    const { name, arguments: args, callId } = /** @type {CustomEvent<{ name: string; arguments: string; callId: string }>} */ (e).detail;
+    const { name, arguments: args, callId, responseId } = /** @type {CustomEvent<{ name: string; arguments: string; callId: string; responseId: string }>} */ (e).detail;
+    if (!responseId) {
+      console.warn(`[tool] call ${callId || "<unknown>"} has no response_id; ignoring uncorrelated tool call`);
+      return;
+    }
     chat.onToolCall(name);
     // Execute the tool, then push it to the conversation once the result is in,
     // so the toggle shows both the call input and its output together.
-    void runTool(name, args, callId).then(({ output, image }) => {
-      chat.onToolResult(name, args, output, image);
+    const execution = runTool(name, args, callId).then(({ output, image }) => {
+      if (client === c) chat.onToolResult(name, args, output, image);
+      return { callId, output, ...(image ? { image } : {}) };
     });
+    toolBatches.add(responseId, execution);
   });
   c.addEventListener("error", (e) => {
     const detail = /** @type {CustomEvent<{ error: unknown }>} */ (e).detail;

server.py

@@ -321,6 +321,10 @@ async def session(request: Request):
     if lb.status_code == 401:
         body = _safe_json(lb)
         reason = body.get("reason") or "login_required"
+        if reason == "token_invalid":
+            session = getattr(request, "scope", {}).get("session")
+            if isinstance(session, dict):
+                session.pop("oauth_info", None)
         logger.info("Session authentication rejected: %s", reason)
         return _login_required_response(reason, set_cookie)
 

tool-call-batcher.js

@@ -0,0 +1,59 @@
+// @ts-check
+
+/**
+ * @typedef {Object} ToolExecutionResult
+ * @property {string} callId
+ * @property {string} output
+ * @property {string} [image]
+ */
+
+/** Collect every tool execution from one response before requesting its follow-up. */
+export class ToolCallBatcher {
+  /** @param {(results: ToolExecutionResult[]) => void | Promise<void>} onReady */
+  constructor(onReady) {
+    this._onReady = onReady;
+    /** @type {Map<string, { executions: Promise<ToolExecutionResult>[]; flush: Promise<void> | null }>} */
+    this._batches = new Map();
+  }
+
+  /**
+   * Register a tool execution in the order its call appeared in the response.
+   * @param {string} responseId
+   * @param {Promise<ToolExecutionResult>} execution
+   */
+  add(responseId, execution) {
+    let batch = this._batches.get(responseId);
+    if (!batch) {
+      batch = { executions: [], flush: null };
+      this._batches.set(responseId, batch);
+    }
+    batch.executions.push(execution);
+  }
+
+  /**
+   * Finish the originating response. Completed responses flush once all tools
+   * settle; unsuccessful responses discard calls the backend rolled back.
+   * @param {string} responseId
+   * @param {string} status
+   * @returns {Promise<void> | null}
+   */
+  finish(responseId, status) {
+    const batch = this._batches.get(responseId);
+    if (!batch) return null;
+    if (status !== "completed") {
+      this._batches.delete(responseId);
+      // Executions cannot be cancelled, but a discarded rejection should not
+      // become unhandled after the response is gone.
+      for (const execution of batch.executions) void execution.catch(() => {});
+      return null;
+    }
+    if (batch.flush) return batch.flush;
+
+    batch.flush = Promise.all(batch.executions)
+      .then((results) => this._onReady(results))
+      .finally(() => {
+        if (this._batches.get(responseId) === batch) this._batches.delete(responseId);
+      });
+    return batch.flush;
+  }
+}

ws/s2s-ws-client.js

@@ -177,12 +177,12 @@ export class S2sWsRealtimeClient extends EventTarget {
      * UI can tell a barge-in cut (keep it) from a never-heard speculative
      * response (drop it). */
     this._audibleResponses = new Set();
-    /** @type {Map<string, string>} The CURRENT assistant transcript segment per
-     * response, accumulated from streamed deltas (reset on each segment's done). */
+    /** @type {Map<string, string>} The in-progress assistant transcript per
+     * response, accumulated from streamed deltas until the terminal done. */
     this._asstTranscriptByResp = new Map();
-    /** @type {Map<string, string>} Completed assistant transcript segments per
-     * response, space-joined. A single response can emit several
-     * `*.transcript.done` events; we concatenate them until response.done. */
+    /** @type {Map<string, string>} Terminal assistant transcript per response.
+     * Realtime emits one `*.transcript.done`; appending remains for compatibility
+     * with legacy endpoints that emitted a done event for every text segment. */
     this._asstFullByResp = new Map();
     this._muted = false;
     // ── Response lock ────────────────────────────────────────────────────
@@ -222,8 +222,8 @@ export class S2sWsRealtimeClient extends EventTarget {
     this.dispatchEvent(new CustomEvent("status", { detail: { status } }));
   }
 
-  /** Full assistant transcript so far for a response: the completed segments
-   *  plus the in-progress one, all space-joined.
+  /** Full assistant transcript so far for a response: a terminal transcript,
+   *  when present, plus any in-progress deltas from a legacy segmented stream.
    *  @param {string} rid @returns {string} */
   _asstDisplay(rid) {
     const full = this._asstFullByResp.get(rid) || "";
@@ -775,9 +775,9 @@ export class S2sWsRealtimeClient extends EventTarget {
         const audible = responseId ? this._audibleResponses.has(responseId) : false;
         this._audibleResponses.delete(responseId);
         // Pull whatever transcript the response carries, falling back to the
-        // segments we concatenated from the `*.transcript.done` events (plus any
-        // in-progress delta). For an interrupted reply the response payload may
-        // be empty, so this is the last chance to capture the text.
+        // terminal `*.transcript.done` value (plus any in-progress delta from a
+        // legacy segmented stream). For an interrupted reply the response
+        // payload may be empty, so this is the last chance to capture the text.
         const transcript =
           extractResponseTranscript(event.response) ||
           this._asstDisplay(responseId) ||
@@ -798,9 +798,10 @@ export class S2sWsRealtimeClient extends EventTarget {
         const name = typeof event.name === "string" ? event.name : "";
         const args = typeof event.arguments === "string" ? event.arguments : "{}";
         const callId = typeof event.call_id === "string" ? event.call_id : "";
+        const responseId = typeof event.response_id === "string" ? event.response_id : "";
         if (name) {
           this.dispatchEvent(new CustomEvent("toolcall", {
-            detail: { name, arguments: args, callId },
+            detail: { name, arguments: args, callId, responseId },
           }));
         } else {
           // A nameless call can't be executed, so no function_call_output is
@@ -844,6 +845,10 @@ export class S2sWsRealtimeClient extends EventTarget {
               },
             }),
           );
+        } else if (this._status === "processing" && !this._responsePending()) {
+          // Empty STT results intentionally do not create a response, so there
+          // will be no response.done event to return the UI to listening.
+          this._setStatus("connected");
         }
         break;
       }
@@ -859,7 +864,7 @@ export class S2sWsRealtimeClient extends EventTarget {
         const delta = typeof event.delta === "string" ? event.delta : "";
         if (delta) {
           this._asstTranscriptByResp.set(rid, (this._asstTranscriptByResp.get(rid) || "") + delta);
-          // Show completed segments + the segment streaming in right now.
+          // Show the complete transcript accumulated from deltas so far.
           this.dispatchEvent(
             new CustomEvent("transcript", {
               detail: { role: "assistant", text: this._asstDisplay(rid), partial: true, responseId: rid },
@@ -872,13 +877,14 @@ export class S2sWsRealtimeClient extends EventTarget {
       case "response.audio_transcript.done":
       case "response.output_audio_transcript.done": {
         const rid = typeof event.response_id === "string" ? event.response_id : "";
-        // This is ONE completed segment. A response can emit several; concatenate
-        // them, space-separated, until response.done clears the accumulator.
+        // This terminalizes the assistant transcript for the response. Realtime
+        // emits it once after all deltas; retain space-joining only for legacy
+        // endpoints that emitted a done event for every text segment.
         const segment =
           (typeof event.transcript === "string" && event.transcript) ||
           this._asstTranscriptByResp.get(rid) ||
           "";
-        this._asstTranscriptByResp.delete(rid); // segment finished; next one starts fresh
+        this._asstTranscriptByResp.delete(rid);
         if (segment) {
           const prev = this._asstFullByResp.get(rid) || "";
           this._asstFullByResp.set(rid, prev ? `${prev} ${segment}` : segment);
@@ -1083,6 +1089,11 @@ export class S2sWsRealtimeClient extends EventTarget {
     return this._openResponses > 0 || this._createInFlight;
   }
 
+  /** True while a response is active, awaiting confirmation, or queued. */
+  _responsePending() {
+    return this._responseActive() || this._createQueue.length > 0;
+  }
+
   /** Send a response.create immediately and arm the in-flight guard. Any image
    *  on the payload is added as user content right before the create.
    *  @param {{ image?: string }} [opts] */