J3: HttpSseTransport event IDs, replay dedupe, 410 gap -> resume

src/peer.ts

@@ -138,6 +138,13 @@ export interface PeerOptions {
    * owns it. Default: true.
    */
   validateCompletions?: boolean;
+  /**
+   * Called when the session cannot be continued after an event gap (the
+   * transport saw 410 and the automatic wmp.session.resume failed or had no
+   * resumption token). The application creates a new session
+   * (Transport Section 4.3.4).
+   */
+  onSessionLost?: (error: Error) => void;
   /** Called for every refused completion, with the violation (never a member value). */
   onRefusedCompletion?: (flowId: string, error: CompletionResultError) => void;
 }
@@ -160,6 +167,7 @@ export class Peer implements PeerContext {
   private authorize?: PeerOptions["authorize"];
   private validateCompletions: boolean;
   private onRefusedCompletion?: PeerOptions["onRefusedCompletion"];
+  private onSessionLost?: PeerOptions["onSessionLost"];
 
   constructor(transport: Transport, opts?: PeerOptions) {
     this.transport = transport;
@@ -169,11 +177,13 @@ export class Peer implements PeerContext {
     this.authorize = opts?.authorize;
     this.validateCompletions = opts?.validateCompletions ?? true;
     this.onRefusedCompletion = opts?.onRefusedCompletion;
+    this.onSessionLost = opts?.onSessionLost;
 
     // Wire up transport events.
     this.transport.on("message", (msg: Message) => this.dispatch(msg));
     this.transport.on("error", (err: Error) => this.onTransportError(err));
     this.transport.on("close", () => this.onTransportClose());
+    this.transport.on("gap", () => this.onTransportGap());
   }
 
   /** Register a profile. Call before starting to receive messages. */
@@ -261,6 +271,7 @@ export class Peer implements PeerContext {
       );
     }
     this.sessionId = result.wmp.session_id;
+    this.terminalSeen.clear(); // a new session: flow state of the old one does not carry over
     this.bindSessionToTransport(result.wmp.session_id);
     this.negotiatedVersion = result.wmp.version ?? VERSION;
     if (result.resumption_token) {
@@ -280,7 +291,9 @@ export class Peer implements PeerContext {
       wmp: { version: this.negotiatedVersion ?? VERSION } as Metadata,
       session_id: this.sessionId,
       resumption_token: this.resumptionToken,
-      last_received_id: opts?.lastReceivedId,
+      // In the HTTP binding this is the last event ID (Transport Section 4.5).
+      last_received_id:
+        opts?.lastReceivedId ?? (this.transport as { lastEventId?: string }).lastEventId,
     };
     const result = await this.call<SessionResumeResult>(
       Method.SessionResume,
@@ -324,6 +337,7 @@ export class Peer implements PeerContext {
       reason,
     });
     this.sessionId = undefined;
+    this.terminalSeen.clear();
   }
 
   // -------------------------------------------------------------------------
@@ -505,6 +519,11 @@ export class Peer implements PeerContext {
     method: string,
     params: unknown,
   ): Promise<unknown> {
+    if ((method === Method.FlowComplete || method === Method.FlowError) && isFlowIdEnvelope(params)) {
+      if (this.terminalSeen.has(params.flow_id)) return undefined; // a repeat from a resume
+      this.markTerminal(params.flow_id);
+    }
+
     // A completion that breaks the result rules is refused before the generic
     // validator runs: a notification the validator rejected would be dropped
     // silently and the receiving handler would never learn that the flow failed.
@@ -546,6 +565,7 @@ export class Peer implements PeerContext {
         break;
 
       case Method.SessionClose:
+        this.terminalSeen.clear();
         if (this.handler.onSessionClose) {
           this.handler.onSessionClose(p);
         }
@@ -673,6 +693,7 @@ export class Peer implements PeerContext {
 
   private recordFlowType(flowId: unknown, flowType: unknown, params: unknown): void {
     if (typeof flowId !== "string" || typeof flowType !== "string") return;
+    this.terminalSeen.delete(flowId); // a new flow under a reused ID
     const q = flowType === "match" && typeof params === "object" && params !== null ? (params as { dcql_query?: unknown }).dcql_query : undefined;
     this.recordedFlows.delete(flowId); // re-insert as the newest
     this.recordedFlows.set(flowId, { type: flowType, queryIds: q === undefined ? undefined : dcqlQueryIds(q) });
@@ -716,6 +737,21 @@ export class Peer implements PeerContext {
     else this.handler.onFlowError?.(synthesized);
   }
 
+  /**
+   * Flows whose terminal notification (complete / error) this peer already
+   * processed. A resume re-sends terminal notifications with fresh event IDs, so
+   * transport-level dedupe cannot recognise them; "a client that already
+   * processed the terminal notification of a flow ignores the repeat"
+   * (Transport 4.3.4). Per session: cleared when the session changes or ends,
+   * and for a flow ID when a new flow starts under it. Bounded.
+   */
+  private terminalSeen = new Set<string>();
+
+  private markTerminal(flowId: string): void {
+    this.terminalSeen.add(flowId);
+    if (this.terminalSeen.size > 4096) this.terminalSeen.delete(this.terminalSeen.values().next().value as string);
+  }
+
   // Track flow_id → flow_type mapping for routing to the right FlowHandler.
   private activeFlows = new Map<string, string>();
 
@@ -752,6 +788,22 @@ export class Peer implements PeerContext {
     // Could emit events here for reconnection logic.
   }
 
+  /**
+   * The transport lost events (410): resume the session, which re-sends the
+   * latest progress of active flows and the terminal notification of ended
+   * flows on the already open stream. Without a resumption token, or if the
+   * resume fails, the session is lost.
+   */
+  private onTransportGap(): void {
+    if (!this.sessionId || !this.resumptionToken) {
+      this.onSessionLost?.(new Error("event gap and no resumption token to resume with"));
+      return;
+    }
+    this.resumeSession().catch((err: unknown) => {
+      this.onSessionLost?.(err instanceof Error ? err : new Error(String(err)));
+    });
+  }
+
   private onTransportClose(): void {
     if (!this.closed) {
       // Transport closed unexpectedly — reject all pending.
@@ -768,6 +820,10 @@ export class Peer implements PeerContext {
 // Error conversion
 // ---------------------------------------------------------------------------
 
+function isFlowIdEnvelope(params: unknown): params is { flow_id: string } {
+  return typeof params === "object" && params !== null && typeof (params as { flow_id?: unknown }).flow_id === "string";
+}
+
 function isFlowCompleteEnvelope(params: unknown): params is FlowCompleteParams {
   return typeof params === "object" && params !== null && typeof (params as { flow_id?: unknown }).flow_id === "string";
 }

src/transport.ts

@@ -17,6 +17,12 @@ export interface TransportEvents {
   error: (err: Error) => void;
   close: () => void;
   open: () => void;
+  /**
+   * The server answered a reconnect with 410 (Transport Section 4.3.4): events
+   * were lost. The transport has already discarded its cursor and opened a new
+   * stream with none; the application (Peer) now resumes the session.
+   */
+  gap: () => void;
 }
 
 export type TransportEventName = keyof TransportEvents;
@@ -160,8 +166,13 @@ export class WebSocketTransport extends Emitter implements Transport {
 export interface HttpSseTransportOptions {
   /** Authorization header value (e.g., "Bearer <token>"). */
   authorization?: string;
-  /** Last-Event-ID for SSE replay. */
+  /**
+   * Initial SSE cursor (the ID of the last event already processed). It is
+   * sent on the first connectSSE() and replaces itself as events arrive.
+   */
   lastEventId?: string;
+  /** How many processed event IDs to remember for replay dedupe. Default: 4096. */
+  dedupeWindow?: number;
   /** Custom fetch implementation (default: global fetch). */
   fetch?: typeof globalThis.fetch;
   /** Custom EventSource constructor (for polyfills that support headers). */
@@ -183,6 +194,10 @@ export class HttpSseTransport extends Emitter implements Transport {
   private fetchFn: typeof globalThis.fetch;
   private EventSourceCtor: typeof EventSource;
   private closed = false;
+  private cursor?: string;
+  private seenIds = new Set<string>();
+  private dedupeWindow: number;
+  private pendingGap = false;
 
   constructor(
     rpcUrl: string,
@@ -195,6 +210,17 @@ export class HttpSseTransport extends Emitter implements Transport {
     this.authorization = opts?.authorization;
     this.fetchFn = opts?.fetch ?? globalThis.fetch.bind(globalThis);
     this.EventSourceCtor = opts?.EventSource ?? globalThis.EventSource;
+    this.cursor = opts?.lastEventId || undefined;
+    this.dedupeWindow = opts?.dedupeWindow ?? 4096;
+  }
+
+  /**
+   * The ID of the last event this transport delivered (the SSE cursor), or
+   * undefined when none was received or the cursor was discarded after a gap.
+   * Event IDs are opaque: store and send them back, never parse or compare.
+   */
+  get lastEventId(): string | undefined {
+    return this.cursor;
   }
 
   /**
@@ -211,75 +237,147 @@ export class HttpSseTransport extends Emitter implements Transport {
    * - `session_id` query parameter on SSE connections
    */
   setSessionId(sessionId: string): void {
+    if (this.sessionId !== undefined && this.sessionId !== sessionId) {
+      // Event IDs are unique only within a session: another session's cursor
+      // would draw a 410 and its seen IDs could swallow this session's events.
+      this.cursor = undefined;
+      this.seenIds.clear();
+      this.pendingGap = false;
+    }
     this.sessionId = sessionId;
   }
 
   /**
    * Connect the SSE stream. Call this after constructing the transport.
    * Separated from the constructor so callers can register listeners first.
+   * Resumes from the transport's own cursor unless `lastEventId` is given.
    */
   connectSSE(lastEventId?: string): void {
-    this.eventSource?.close();
-    this.eventSource = null;
-
-    this.setupEventSource(lastEventId);
+    if (lastEventId !== undefined) this.cursor = lastEventId || undefined;
+    this.openStream(this.cursor);
   }
 
   /**
-   * Reconnect the SSE stream with the current authorization/session state.
-   * Useful after calling setAuthorization.
+   * Reconnect the SSE stream with the current authorization/session state
+   * and the recorded cursor. Useful after calling setAuthorization.
    */
   reconnectSSE(lastEventId?: string): void {
     this.connectSSE(lastEventId);
   }
 
-  private setupEventSource(lastEventId?: string): void {
-    // Build the SSE URL with session_id (required for tenant routing)
-    // and optional lastEventId for event replay.
-    let url = this.eventsUrl;
-    if (this.sessionId) {
-      const sep = url.includes("?") ? "&" : "?";
-      url += `${sep}session_id=${encodeURIComponent(this.sessionId)}`;
-    }
-    if (lastEventId) {
-      const sep = url.includes("?") ? "&" : "?";
-      url += `${sep}lastEventId=${encodeURIComponent(lastEventId)}`;
+  private streamHeaders(cursor?: string): Record<string, string> {
+    const h: Record<string, string> = {};
+    if (this.authorization) h["Authorization"] = this.authorization;
+    if (cursor) h["Last-Event-ID"] = cursor;
+    return h;
+  }
+
+  private streamUrl(cursor?: string): string {
+    // session_id (required for tenant routing) and the cursor. The cursor goes
+    // as last_event_id (Transport Section 4.3.3) and, for servers that predate
+    // that name, as lastEventId. Only the query string is touched: a relative
+    // URL ("events", "/x", "//host/x") must resolve exactly as it was supplied.
+    const hashAt = this.eventsUrl.indexOf("#");
+    const hash = hashAt < 0 ? "" : this.eventsUrl.slice(hashAt);
+    const noHash = hashAt < 0 ? this.eventsUrl : this.eventsUrl.slice(0, hashAt);
+    const qAt = noHash.indexOf("?");
+    const base = qAt < 0 ? noHash : noHash.slice(0, qAt);
+    const params = new URLSearchParams(qAt < 0 ? "" : noHash.slice(qAt + 1));
+    if (this.sessionId) params.set("session_id", this.sessionId);
+    if (cursor) {
+      params.set("last_event_id", cursor);
+      params.set("lastEventId", cursor);
     }
+    const query = params.toString();
+    return base + (query ? `?${query}` : "") + hash;
+  }
 
-    this.eventSource = new this.EventSourceCtor(url);
+  private openStream(cursor: string | undefined): void {
+    this.eventSource?.close();
+    this.eventSource = null;
 
-    this.eventSource.onopen = () => this.emit("open");
-    this.eventSource.onerror = () => {
-      if (!this.closed) {
-        this.emit("error", new Error("SSE connection error"));
+    const headers = this.streamHeaders(cursor);
+    // Polyfills that can set request headers read `headers` from the init
+    // object; a native EventSource ignores it.
+    const init = Object.keys(headers).length > 0 ? ({ headers } as EventSourceInit) : undefined;
+    const es = new this.EventSourceCtor(this.streamUrl(cursor), init);
+    this.eventSource = es;
+
+    es.onopen = () => {
+      this.emit("open");
+      if (this.pendingGap) {
+        this.pendingGap = false;
+        this.emit("gap");
       }
     };
-
-    // WMP events use the "wmp" event type per the spec.
-    this.eventSource.addEventListener("wmp", (ev) => {
-      try {
-        const msg = decodeMessage((ev as MessageEvent).data, { maxSize: 4 * 1024 * 1024 });
-        this.emit("message", msg);
-      } catch (err) {
-        this.emit(
-          "error",
-          err instanceof Error ? err : new Error(String(err)),
-        );
+    es.onerror = () => {
+      if (this.closed || this.eventSource !== es) return;
+      // A closed stream with a cursor may be the 410 of a gap; the EventSource
+      // API does not expose the status, so ask.
+      if (es.readyState === 2 /* CLOSED */ && this.cursor) {
+        void this.checkGap(es, this.cursor);
+        return;
       }
-    });
+      this.emit("error", new Error("SSE connection error"));
+    };
 
+    // WMP events use the "wmp" event type per the spec.
+    es.addEventListener("wmp", (ev) => this.onSseEvent(ev as MessageEvent));
     // Also handle default "message" events for compatibility.
-    this.eventSource.onmessage = (ev) => {
-      try {
-        const msg = decodeMessage(ev.data, { maxSize: 4 * 1024 * 1024 });
-        this.emit("message", msg);
-      } catch (err) {
-        this.emit(
-          "error",
-          err instanceof Error ? err : new Error(String(err)),
-        );
-      }
-    };
+    es.onmessage = (ev) => this.onSseEvent(ev);
+  }
+
+  /**
+   * Handles one SSE event: an event whose ID was already processed (a replay
+   * after a reconnect) is dropped; otherwise it is delivered and then its ID
+   * becomes the cursor (Transport Section 4.3.2).
+   */
+  private onSseEvent(ev: MessageEvent): void {
+    const id = typeof ev.lastEventId === "string" && ev.lastEventId !== "" ? ev.lastEventId : undefined;
+    if (id !== undefined && this.seenIds.has(id)) return;
+    try {
+      this.emit("message", decodeMessage(ev.data, { maxSize: 4 * 1024 * 1024 }));
+    } catch (err) {
+      this.emit("error", err instanceof Error ? err : new Error(String(err)));
+    } finally {
+      if (id !== undefined) this.recordEventId(id);
+    }
+  }
+
+  private recordEventId(id: string): void {
+    this.cursor = id;
+    this.seenIds.add(id);
+    if (this.seenIds.size > this.dedupeWindow) {
+      // Sets iterate in insertion order: drop the oldest.
+      this.seenIds.delete(this.seenIds.values().next().value as string);
+    }
+  }
+
+  /** Probes the stream URL for a 410 after the EventSource gave up with `cursor`. */
+  private async checkGap(failed: EventSource, cursor: string): Promise<void> {
+    let status: number | undefined;
+    try {
+      const abort = new AbortController();
+      const resp = await this.fetchFn(this.streamUrl(cursor), {
+        headers: { ...this.streamHeaders(cursor), Accept: "text/event-stream" },
+        signal: abort.signal,
+      });
+      status = resp.status;
+      abort.abort(); // only the status matters; do not hold the stream open
+    } catch {
+      status = undefined;
+    }
+    if (this.closed || this.eventSource !== failed) return;
+    if (status === 410) {
+      // Discard the cursor (never reconnect with it again) and open a stream
+      // with none; "gap" is emitted once it is open, so the resume's re-sent
+      // notifications have a connection to arrive on.
+      this.cursor = undefined;
+      this.pendingGap = true;
+      this.openStream(undefined);
+      return;
+    }
+    this.emit("error", new Error("SSE connection error"));
   }
 
   async send(msg: Message): Promise<void> {