feat(sdk): harden daemon session client

packages/sdk-typescript/src/daemon/DaemonSessionClient.ts

@@ -27,7 +27,9 @@ export interface DaemonSessionClientOptions {
   state?: DaemonSessionState;
   /**
    * Seed replay state for callers that persisted the last seen SSE event id.
-   * When omitted, the first event subscription starts live.
+   * When omitted, the first event subscription starts live. Values must be
+   * finite, non-negative integers because the daemon uses these ids as
+   * `Last-Event-ID` resume cursors.
    */
   lastEventId?: number;
 }
@@ -62,7 +64,7 @@ export class DaemonSessionClient {
     this.client = opts.client;
     this.session = { ...opts.session };
     this.state = { ...(opts.state ?? {}) };
-    this.lastSeenEventId = opts.lastEventId;
+    this.lastSeenEventId = validateLastEventId(opts.lastEventId);
   }
 
   /**
@@ -76,7 +78,10 @@ export class DaemonSessionClient {
     // `modelServiceId` switch failures are reported on SSE, not the
     // create/attach HTTP response. Seed the first subscription from the
     // daemon replay ring so create-then-subscribe clients observe attach-time
-    // `model_switch_failed` / `model_switched` events.
+    // `model_switch_failed` / `model_switched` events. The daemon treats
+    // Last-Event-ID: 0 as "replay from the beginning of the bounded ring";
+    // if older events have already been evicted, clients receive the retained
+    // suffix and continue live from there.
     const lastEventId = req.modelServiceId ? 0 : undefined;
     return new DaemonSessionClient({ client, session, lastEventId });
   }
@@ -140,7 +145,7 @@ export class DaemonSessionClient {
   }
 
   setLastEventId(lastEventId: number | undefined): void {
-    this.lastSeenEventId = lastEventId;
+    this.lastSeenEventId = validateLastEventId(lastEventId);
   }
 
   async prompt(
@@ -167,21 +172,77 @@ export class DaemonSessionClient {
 
   events(
     opts: DaemonSessionSubscribeOptions = {},
-  ): AsyncGenerator<DaemonEvent> {
-    return this.subscribeEvents(opts);
+  ): AsyncGenerator<DaemonEvent, void, unknown> {
+    return this.openEventSubscription(opts);
   }
 
-  async *subscribeEvents(
+  /**
+   * @deprecated Use {@link events} instead. Both methods are equivalent.
+   */
+  subscribeEvents(
     opts: DaemonSessionSubscribeOptions = {},
-  ): AsyncGenerator<DaemonEvent> {
-    if (this.subscriptionActive) {
-      throw new Error(
-        'Another event subscription is already active on this session. ' +
-          'Reuse the existing AsyncGenerator or create a separate DaemonSessionClient.',
-      );
-    }
+  ): AsyncGenerator<DaemonEvent, void, unknown> {
+    return this.openEventSubscription(opts);
+  }
+
+  private openEventSubscription(
+    opts: DaemonSessionSubscribeOptions,
+  ): AsyncGenerator<DaemonEvent, void, unknown> {
+    const requestedLastEventId = validateLastEventId(opts.lastEventId);
+    let started = false;
+    let released = false;
+    const release = () => {
+      if (released) return;
+      released = true;
+      this.subscriptionActive = false;
+    };
+    const acquire = () => {
+      if (started) return;
+      if (this.subscriptionActive) {
+        throw new Error(
+          'Another event subscription is already active on this session. ' +
+            'Reuse the existing AsyncGenerator or create a separate DaemonSessionClient.',
+        );
+      }
+      this.subscriptionActive = true;
+      started = true;
+    };
+    const iterator = this.iterateEvents(
+      { ...opts, lastEventId: requestedLastEventId },
+      release,
+    );
 
-    this.subscriptionActive = true;
+    return {
+      next: async (value?: unknown) => {
+        if (!released) {
+          acquire();
+        }
+        return await iterator.next(value);
+      },
+      return: async () => {
+        try {
+          return await iterator.return(undefined);
+        } finally {
+          release();
+        }
+      },
+      throw: async (error?: unknown) => {
+        try {
+          return await iterator.throw(error);
+        } finally {
+          release();
+        }
+      },
+      [Symbol.asyncIterator]() {
+        return this;
+      },
+    };
+  }
+
+  private async *iterateEvents(
+    opts: DaemonSessionSubscribeOptions,
+    release: () => void,
+  ): AsyncGenerator<DaemonEvent, void, unknown> {
     try {
       const { resume = true, ...subscribeOpts } = opts;
       const lastEventId =
@@ -193,11 +254,36 @@ export class DaemonSessionClient {
         lastEventId,
       })) {
         yield event;
-        // Terminal/synthetic frames may not carry an SSE id.
-        if (event.id !== undefined) this.lastSeenEventId = event.id;
+        // Cursor updates happen after the consumer resumes iteration. That
+        // avoids acknowledging an event before the adapter has processed it,
+        // but means `lastEventId` intentionally lags while the handler for the
+        // just-yielded event is still running.
+        // The cursor is a replay watermark, so it only moves forward even if a
+        // replayed or synthetic frame arrives with an older id.
+        if (event.id !== undefined) {
+          this.lastSeenEventId = Math.max(
+            this.lastSeenEventId ?? 0,
+            validateLastEventId(event.id),
+          );
+        }
       }
     } finally {
-      this.subscriptionActive = false;
+      release();
     }
   }
 }
+
+function validateLastEventId(lastEventId: number): number;
+function validateLastEventId(lastEventId: undefined): undefined;
+function validateLastEventId(
+  lastEventId: number | undefined,
+): number | undefined;
+function validateLastEventId(
+  lastEventId: number | undefined,
+): number | undefined {
+  if (lastEventId === undefined) return undefined;
+  if (!Number.isInteger(lastEventId) || lastEventId < 0) {
+    throw new TypeError('lastEventId must be a finite non-negative integer');
+  }
+  return lastEventId;
+}