fix(core,sdk): drain cursor, scan timeout, seqNum NaN guard

Three small follow-ups from the code-review pass.

`StandardSessionStreamManager.on()`'s buffer-drain now advances the
committed-consume cursor (`lastDispatchedSeqNums`) to the highest
buffered seq before deleting the buffer. Drained records are removed
from the buffer and no longer reachable via `once()`, so they're
effectively consumed from the manager's perspective — without the
advance, a worker using `messagesInput.on()` (pendingMessages mode)
would persist a `.in` cursor that lags behind records the handler
already processed and the next boot would re-deliver them.

`findLatestSessionInCursor`'s `timeoutInSeconds` raised from 1 to 5.
S2 trim is eventually-consistent over 10-60s; a worker booting just
after a trim could still see pre-trim records, and 1s isn't always
enough headroom to drain them and read the latest turn-complete header
before the SSE long-poll closes. Without the cursor seed, `.in` opens
from seq=0 and the slim-wire merge catches the dupes — correct but
the optimization is silently lost.

`subscribeToSessionStream`'s control-event dispatch replaces
`Number.parseInt(part.id, 10) || 0` with an explicit `Number.isFinite`
guard that drops the event when the id is non-numeric, rather than
firing `onControl` with a misleading `seqNum: 0`. Doesn't matter in
practice (S2 always emits numeric ids), but `findLatestSessionInCursor`
and the dashboard rely on the seqNum being meaningful for resume.

Author: ericallam

packages/core/src/v3/apiClient/index.ts

@@ -1346,10 +1346,21 @@ export class ApiClient {
           // Trigger control record — route to onControl, never enqueue.
           const subtype = controlSubtype(part.headers);
           if (subtype) {
+            // `part.id` is the S2 seq_num in decimal string form.
+            // `parseInt` returns NaN if S2 ever surfaces a non-numeric
+            // id (shouldn't happen, but `|| 0` would mask it as a real
+            // seq 0). Drop the malformed event rather than fire
+            // `onControl` with a misleading cursor — callers like
+            // `findLatestSessionInCursor` and the dashboard rely on the
+            // seqNum being meaningful for resume.
+            const parsedSeqNum = Number.parseInt(part.id, 10);
+            if (!Number.isFinite(parsedSeqNum)) {
+              return;
+            }
             onControl?.({
               subtype,
               headers: part.headers ?? [],
-              seqNum: Number.parseInt(part.id, 10) || 0,
+              seqNum: parsedSeqNum,
               timestamp: part.timestamp,
             });
             return;

packages/core/src/v3/sessionStreams/manager.ts

@@ -118,6 +118,20 @@ export class StandardSessionStreamManager implements SessionStreamManager {
       for (const data of buffered) {
         this.#invokeHandler(handler, data);
       }
+      // Advance the committed-consume cursor to the highest seq drained
+      // into the new handler. `on()`-drain removes the records from the
+      // buffer, so they're no longer available to a future `once()` —
+      // from the manager's perspective they've been consumed. Without
+      // this, a worker that uses `messagesInput.on()` for user-message
+      // delivery (pendingMessages mode) would persist a `.in` cursor
+      // that lags behind the records the handler already processed, and
+      // the next boot would re-deliver them.
+      const seqList = this.bufferSeqNums.get(key);
+      if (seqList) {
+        for (const s of seqList) {
+          if (s !== undefined) this.#advanceLastDispatched(key, s);
+        }
+      }
       this.buffer.delete(key);
       // Keep `bufferSeqNums` in lock-step with `buffer` — without this,
       // the parallel array desyncs and the next `#dispatch` that buffers

packages/trigger-sdk/src/v3/ai.ts

@@ -178,7 +178,13 @@ async function findLatestSessionInCursor(
   const apiClient = apiClientManager.clientOrThrow();
   let latestCursor: number | undefined;
   const stream = await apiClient.subscribeToSessionStream<unknown>(chatId, "out", {
-    timeoutInSeconds: 1,
+    // 5s rather than 1s: S2 trim is eventually-consistent (10-60s
+    // window), so a worker booting just after a trim could still see
+    // pre-trim records and need a bit longer to drain them all before
+    // the SSE long-poll closes. Without enough headroom the scan would
+    // fall back to `undefined`, the `.in` cursor wouldn't be seeded,
+    // and the next subscribe would replay messages already processed.
+    timeoutInSeconds: 5,
     onControl: (event) => {
       if (event.subtype !== TRIGGER_CONTROL_SUBTYPE.TURN_COMPLETE) return;
       const raw = headerValue(event.headers, SESSION_IN_EVENT_ID_HEADER);