feat: seq-based incremental resume + send idempotency (protocol P0, 2/2)

packages/protocol/src/schemas.ts

@@ -63,6 +63,12 @@ export const sessionAttachSchema = z.object({
   ...base,
   type: z.literal("session.attach"),
   sessionId: sessionIdField,
+  resume: z
+    .object({
+      key: z.string().min(1).max(LIMITS.ID_MAX),
+      sinceSeq: z.number().int().nonnegative(),
+    })
+    .optional(),
 });
 
 export const sessionDetachSchema = z.object({
@@ -79,6 +85,7 @@ export const sessionSendSchema = z.object({
   text: z.string().max(LIMITS.SEND_TEXT_MAX),
   attachments: z.array(attachmentSchema).max(LIMITS.ATTACHMENTS_MAX).optional(),
   priority: z.enum(["now", "next", "later"]).optional(),
+  clientMsgId: z.string().min(1).max(LIMITS.ID_MAX).optional(),
 });
 
 export const sessionInterruptSchema = z.object({

packages/protocol/src/types.ts

@@ -577,6 +577,15 @@ export interface SessionMessage {
   /** Extensible metadata — frontends ignore unknown keys */
   metadata?: Record<string, unknown>;
   timestamp: string;
+  /**
+   * Session sequence cursor (`replay.resume` capability): the session's
+   * monotonic mutation counter at the time this frame was produced. Clients
+   * track `max(seq)` per session and pass it back on `session.attach.resume`
+   * to receive an incremental tail instead of a full scrollback replay.
+   * Absent on daemons that predate resume, and on messages the daemon no
+   * longer holds in its replay buffer.
+   */
+  seq?: number;
 }
 
 /**
@@ -603,6 +612,8 @@ export interface SessionMessageDelta {
   /** Update tool state (state machine transition) */
   toolStateUpdate?: ToolState;
   timestamp: string;
+  /** Session sequence cursor — see `SessionMessage.seq`. */
+  seq?: number;
 }
 
 // =============================================================================
@@ -701,6 +712,16 @@ export interface SessionListMsg extends BaseClientMsg {
 export interface SessionAttachMsg extends BaseClientMsg {
   type: "session.attach";
   sessionId: string;
+  /**
+   * Incremental resume (`replay.resume` capability). `key` is the
+   * `resumeKey` from a previous replay on this session; `sinceSeq` is the
+   * highest `seq` the client has applied. When the key matches the daemon's
+   * current replay buffer, the daemon replays only entries mutated after
+   * `sinceSeq` (`mode: "incremental"`); on any mismatch (daemon restarted,
+   * buffer rebuilt, key unknown) it falls back to a full snapshot. Omit for
+   * the legacy full-replay behaviour.
+   */
+  resume?: { key: string; sinceSeq: number };
 }
 
 export interface SessionDetachMsg extends BaseClientMsg {
@@ -727,6 +748,15 @@ export interface SessionSendMsg extends BaseClientMsg {
    * Frontends that don't care pass nothing; FIFO stays the default.
    */
   priority?: "now" | "next" | "later";
+  /**
+   * Idempotency key (`send.idempotency` capability). Generate ONCE per user
+   * action (not per network attempt) and reuse it on retries: a send whose
+   * `clientMsgId` the daemon has already processed for this session is
+   * acknowledged without running a second turn. This is the guard against
+   * ambiguous delivery (socket drop between send and ack) turning one prompt
+   * into two billed turns. Omit to opt out (every send processes).
+   */
+  clientMsgId?: string;
 }
 
 export interface Attachment {
@@ -1302,10 +1332,32 @@ export interface ScrollbackReplayMsg {
   type: "scrollback.replay";
   sessionId: string;
   messages: SessionMessage[];
-  /** 0-based chunk index of a chunked replay. Absent = single-frame replay. */
+  /**
+   * 0-based CHUNK index of a chunked replay (#84). Absent = single-frame
+   * replay. NOTE: unrelated to the per-message session cursor
+   * `SessionMessage.seq` — this one only orders the frames of one replay.
+   */
   seq?: number;
   /** True on the last chunk of a chunked replay. Absent = single-frame replay. */
   final?: boolean;
+  /**
+   * Replay semantics (`replay.resume` capability):
+   *   - "snapshot" (or absent — the legacy shape): the authoritative full
+   *     scrollback; clients RESET their local buffer to it.
+   *   - "incremental": only entries mutated since the client's `sinceSeq`;
+   *     clients APPEND/UPSERT by messageId — never reset. Sent when a
+   *     `session.attach.resume` key matched.
+   */
+  mode?: "snapshot" | "incremental";
+  /**
+   * Identity of the daemon's replay buffer. Store it with `maxSeq` and pass
+   * both back on `session.attach.resume`. Changes whenever the buffer is
+   * rebuilt (e.g. daemon restart) — a mismatch means cursors are invalid and
+   * the daemon answers with a snapshot.
+   */
+  resumeKey?: string;
+  /** Highest session sequence included/known — the client's next cursor. */
+  maxSeq?: number;
 }
 
 /** Result of a session.search query. */

src/daemon/scrollback.ts

@@ -30,10 +30,15 @@ const DEFAULT_CONFIG: ScrollbackConfig = {
  * grow in place between push and finalize; eviction must subtract exactly
  * what was added, never the current (grown) serialized size — otherwise the
  * counter drifts negative and the byte cap stops evicting.
+ *
+ * `seq` is the buffer's monotonic mutation counter value at this entry's
+ * LAST mutation (push / update / touch). Incremental resume filters on it:
+ * any entry mutated after a client's cursor gets resent in merged form.
  */
 interface Entry {
   msg: DaemonMessage;
   size: number;
+  seq: number;
 }
 
 function messageIdOf(msg: DaemonMessage): string | undefined {
@@ -55,11 +60,24 @@ export class ScrollbackBuffer {
   #byId = new Map<string, Entry>();
   #bytes = 0;
   #config: ScrollbackConfig;
+  /**
+   * Monotonic mutation counter — the session sequence cursor domain for
+   * incremental resume (`replay.resume`). Bumped by every push / update /
+   * touch; NEVER reset while the buffer lives. Scoped to this buffer
+   * instance: cursors are only meaningful together with the session's
+   * `resumeKey`, which changes when the buffer is rebuilt.
+   */
+  #seq = 0;
 
   constructor(config: Partial<ScrollbackConfig> = {}) {
     this.#config = { ...DEFAULT_CONFIG, ...config };
   }
 
+  /** Highest sequence value assigned so far (0 = nothing ever buffered). */
+  get maxSeq(): number {
+    return this.#seq;
+  }
+
   /**
    * Push a message into the buffer. Evicts oldest entries if limits are exceeded.
    *
@@ -75,6 +93,15 @@ export class ScrollbackBuffer {
    * re-serializes every one of them purely for byte accounting.
    */
   push(msg: DaemonMessage, sizeHint?: number): void {
+    const seq = ++this.#seq;
+    // Stamp the session cursor onto the message itself so the object the
+    // session broadcasts (same reference) carries it on the wire. Skipped
+    // when the caller supplied a sizeHint (restore-from-transcript) — the
+    // hint reflects the unstamped line, and restored messages don't need a
+    // wire seq (the replay frame's maxSeq covers the client's cursor).
+    if (sizeHint === undefined && msg.type === "session.message") {
+      msg.seq = seq;
+    }
     const messageId = messageIdOf(msg);
     if (messageId !== undefined) {
       const existing = this.#byId.get(messageId);
@@ -83,17 +110,41 @@ export class ScrollbackBuffer {
         this.#bytes += size - existing.size;
         existing.msg = msg;
         existing.size = size;
+        existing.seq = seq;
         this.#evict();
         return;
       }
     }
-    const entry: Entry = { msg, size: sizeHint ?? serializedSizeOf(msg) };
+    const entry: Entry = { msg, size: sizeHint ?? serializedSizeOf(msg), seq };
     this.#entries.push(entry);
     if (messageId !== undefined) this.#byId.set(messageId, entry);
     this.#bytes += entry.size;
     this.#evict();
   }
 
+  /**
+   * Record a mutation of a buffered message WITHOUT re-accounting its bytes —
+   * the streaming path calls this once per delta, so it must stay O(1) with
+   * no re-serialization. Bumps the buffer counter, marks the entry as
+   * mutated-at-that-seq (so incremental resume resends the merged message),
+   * stamps the buffered message so later replays emit a self-consistent
+   * per-message seq (entry.seq === msg.seq), and returns the new seq for
+   * stamping the outgoing delta frame.
+   *
+   * The seq stamp is deliberately NOT byte-re-accounted: the drift is bounded
+   * (~15 bytes per entry, once) and re-serializing per delta would put a full
+   * JSON.stringify on the per-token hot path.
+   *
+   * Returns undefined when the message is unknown/evicted.
+   */
+  touch(messageId: string): number | undefined {
+    const entry = this.#byId.get(messageId);
+    if (!entry) return undefined;
+    entry.seq = ++this.#seq;
+    if (entry.msg.type === "session.message") entry.msg.seq = entry.seq;
+    return entry.seq;
+  }
+
   /** Evict oldest entries until within both limits. */
   #evict(): void {
     while (
@@ -131,12 +182,33 @@ export class ScrollbackBuffer {
    * the message. Returns [] for an empty buffer.
    */
   readChunked(maxBytes: number): DaemonMessage[][] {
+    return ScrollbackBuffer.#partition(this.#entries, maxBytes);
+  }
+
+  /**
+   * Incremental-resume read (`replay.resume`): every entry mutated after
+   * `sinceSeq` — new messages AND older messages that grew via deltas or
+   * tool-state transitions since the client's cursor — in buffer order,
+   * partitioned by the same byte budget as `readChunked`. Returns [] when
+   * the client is fully caught up.
+   */
+  readChunkedSince(sinceSeq: number, maxBytes: number): DaemonMessage[][] {
+    const stale = this.#entries.filter((e) => e.seq > sinceSeq);
+    return ScrollbackBuffer.#partition(stale, maxBytes);
+  }
+
+  /**
+   * Partition entries into ordered chunks (oldest→newest), each holding at
+   * most ~`maxBytes` of serialized payload, using the byte sizes already
+   * accounted per entry — no re-serialization. A single message larger than
+   * `maxBytes` occupies its own chunk (never split). Never emits an empty
+   * chunk; returns [] for no entries.
+   */
+  static #partition(entries: readonly Entry[], maxBytes: number): DaemonMessage[][] {
     const chunks: DaemonMessage[][] = [];
     let current: DaemonMessage[] = [];
     let currentBytes = 0;
-    for (const entry of this.#entries) {
-      // Start a new chunk when adding this entry would overflow the budget,
-      // but never emit an empty chunk (a lone oversized message stays put).
+    for (const entry of entries) {
       if (current.length > 0 && currentBytes + entry.size > maxBytes) {
         chunks.push(current);
         current = [];
@@ -149,17 +221,6 @@ export class ScrollbackBuffer {
     return chunks;
   }
 
-  /**
-   * Read messages after a given timestamp (for incremental catch-up).
-   */
-  readSince(timestamp: string): DaemonMessage[] {
-    return this.#entries
-      .map((e) => e.msg)
-      .filter(
-        (msg) => "timestamp" in msg && (msg as { timestamp: string }).timestamp > timestamp,
-      );
-  }
-
   /**
    * Update a message in the buffer by messageId. Used to apply tool state
    * transitions so scrollback replay shows final states, not intermediate.
@@ -168,6 +229,10 @@ export class ScrollbackBuffer {
     const entry = this.#byId.get(messageId);
     if (!entry) return;
     updater(entry.msg);
+    entry.seq = ++this.#seq;
+    // Keep the buffered message's wire seq consistent with the entry —
+    // stamped BEFORE re-accounting, so the bytes stay exact here.
+    if (entry.msg.type === "session.message") entry.msg.seq = entry.seq;
     const after = serializedSizeOf(entry.msg);
     this.#bytes += after - entry.size;
     entry.size = after;

src/daemon/server.ts

@@ -36,7 +36,11 @@ import type { IncomingMessage, ServerResponse } from "node:http";
  * capability-gated behaviour lands; clients feature-detect on it instead of
  * version-sniffing.
  */
-const SERVER_CAPABILITIES: string[] = [CAPABILITIES.CHUNKED_REPLAY];
+const SERVER_CAPABILITIES: string[] = [
+  CAPABILITIES.CHUNKED_REPLAY,
+  CAPABILITIES.SEQ_RESUME,
+  CAPABILITIES.SEND_IDEMPOTENCY,
+];
 
 /**
  * Per-connection state carried on `ws.data`. Defined once and cast against in

src/daemon/session-manager.ts

@@ -920,7 +920,7 @@ export class SessionManager {
       return { type: "response.error", requestId: msg.id, error: "Session not found", code: "not_found" };
     }
 
-    session.attach(client);
+    session.attach(client, msg.resume);
     return { type: "response.ok", requestId: msg.id, data: session.toInfo() };
   }
 
@@ -950,6 +950,15 @@ export class SessionManager {
       return { type: "response.error", requestId: msg.id, error: "Session not found", code: "not_found" };
     }
 
+    // Duplicate-send suppression (`send.idempotency`): a client that
+    // couldn't observe whether its send survived a dropped socket resends
+    // with the SAME clientMsgId — acknowledging instead of dispatching
+    // prevents one prompt from becoming two billed turns. Checked after
+    // scope + ownership so a rejected send never poisons the id.
+    if (msg.clientMsgId !== undefined && session.markClientMsgSeen(msg.clientMsgId)) {
+      return { type: "response.ok", requestId: msg.id, data: { duplicate: true } };
+    }
+
     // Fire and forget — output streams to attached clients. The user message
     // is persisted synchronously at the top of session.send() before any
     // fallible work, so a later throw can't lose it. Surface that throw as a