fix: streamed messages pushed into scrollback twice — commit finalize via upsert

src/daemon/scrollback.ts

@@ -24,8 +24,35 @@ const DEFAULT_CONFIG: ScrollbackConfig = {
   maxBytes: 20 * 1024 * 1024, // 20MB
 };
 
+/**
+ * Internal wrapper that records the byte size that was actually accounted
+ * into `#bytes` for this entry. Streamed messages are held by reference and
+ * 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.
+ */
+interface Entry {
+  msg: DaemonMessage;
+  size: number;
+}
+
+function messageIdOf(msg: DaemonMessage): string | undefined {
+  return msg.type === "session.message"
+    ? (msg as { messageId?: string }).messageId
+    : undefined;
+}
+
+/**
+ * Serialized size in real UTF-8 bytes. `String.length` counts UTF-16 code
+ * units, undercounting non-ASCII payloads against the byte cap.
+ */
+function serializedSizeOf(msg: DaemonMessage): number {
+  return Buffer.byteLength(JSON.stringify(msg), "utf8");
+}
+
 export class ScrollbackBuffer {
-  #entries: DaemonMessage[] = [];
+  #entries: Entry[] = [];
+  #byId = new Map<string, Entry>();
   #bytes = 0;
   #config: ScrollbackConfig;
 
@@ -35,10 +62,30 @@ export class ScrollbackBuffer {
 
   /**
    * Push a message into the buffer. Evicts oldest entries if limits are exceeded.
+   *
+   * Upserts by messageId: pushing a message whose messageId is already
+   * buffered re-accounts the existing entry in place (keeping its position)
+   * instead of appending a second entry. Duplicate entries for one messageId
+   * corrupt scrollback.replay — clients render the message twice and
+   * virtualizers keyed on messageId collide (the #50 bug class).
    */
   push(msg: DaemonMessage): void {
-    this.#entries.push(msg);
-    this.#bytes += JSON.stringify(msg).length;
+    const messageId = messageIdOf(msg);
+    if (messageId !== undefined) {
+      const existing = this.#byId.get(messageId);
+      if (existing) {
+        const size = serializedSizeOf(msg);
+        this.#bytes += size - existing.size;
+        existing.msg = msg;
+        existing.size = size;
+        this.#evict();
+        return;
+      }
+    }
+    const entry: Entry = { msg, size: serializedSizeOf(msg) };
+    this.#entries.push(entry);
+    if (messageId !== undefined) this.#byId.set(messageId, entry);
+    this.#bytes += entry.size;
     this.#evict();
   }
 
@@ -49,8 +96,11 @@ export class ScrollbackBuffer {
       this.#bytes > this.#config.maxBytes
     ) {
       const evicted = this.#entries.shift();
-      if (evicted) {
-        this.#bytes -= JSON.stringify(evicted).length;
+      if (!evicted) break;
+      this.#bytes -= evicted.size;
+      const id = messageIdOf(evicted.msg);
+      if (id !== undefined && this.#byId.get(id) === evicted) {
+        this.#byId.delete(id);
       }
     }
   }
@@ -60,37 +110,32 @@ export class ScrollbackBuffer {
    * Returns a snapshot — safe to iterate while new messages arrive.
    */
   read(): DaemonMessage[] {
-    return [...this.#entries];
+    return this.#entries.map((e) => e.msg);
   }
 
   /**
    * Read messages after a given timestamp (for incremental catch-up).
    */
   readSince(timestamp: string): DaemonMessage[] {
-    return this.#entries.filter(
-      (msg) => "timestamp" in msg && (msg as { timestamp: string }).timestamp > timestamp,
-    );
+    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.
    */
   updateMessage(messageId: string, updater: (msg: DaemonMessage) => void): void {
-    for (const entry of this.#entries) {
-      if (entry.type === "session.message" && (entry as { messageId?: string }).messageId === messageId) {
-        // Re-account bytes around the in-place mutation. Tool entries are
-        // pushed small (no output) then mutated to carry large output; without
-        // adjusting #bytes here, eviction later subtracts the grown size that
-        // was never added, drifting #bytes negative and defeating the byte cap.
-        const before = JSON.stringify(entry).length;
-        updater(entry);
-        const after = JSON.stringify(entry).length;
-        this.#bytes += after - before;
-        this.#evict();
-        return;
-      }
-    }
+    const entry = this.#byId.get(messageId);
+    if (!entry) return;
+    updater(entry.msg);
+    const after = serializedSizeOf(entry.msg);
+    this.#bytes += after - entry.size;
+    entry.size = after;
+    this.#evict();
   }
 
   /** Number of entries currently buffered. */
@@ -106,6 +151,7 @@ export class ScrollbackBuffer {
   /** Clear the buffer. */
   clear(): void {
     this.#entries = [];
+    this.#byId.clear();
     this.#bytes = 0;
   }
 }

src/daemon/session-manager.ts

@@ -157,10 +157,12 @@ export class SessionManager {
           onModels: (m) => this.#cacheModels(m),
         });
 
-        // Restore scrollback from transcript
+        // Restore scrollback from transcript, seeding the seq counter past
+        // the persisted tail so new appends continue the monotonic sequence.
         const entries = await this.#transcriptStore.loadTranscript(meta.sessionId);
         const messages = entries.map((e) => e.message);
-        session.restoreScrollback(messages);
+        const maxSeq = entries.reduce((max, e) => Math.max(max, e.seq), -1);
+        session.restoreScrollback(messages, maxSeq + 1);
 
         this.#sessions.set(session.id, session);
         // Resume is NOT a creation — don't burn a slot in the

src/daemon/session.ts

@@ -931,7 +931,14 @@ export class Session {
     await this.#transcriptStore.delete(this.id);
   }
 
-  restoreScrollback(messages: DaemonMessage[]): void {
+  restoreScrollback(messages: DaemonMessage[], nextSeq?: number): void {
+    // Seed the transcript sequence counter past the loaded log's tail.
+    // Without this, post-restart appends restart at seq 0 — harmless for
+    // loadTranscript (which orders by file position) but it makes seq
+    // unusable as a monotonic replay cursor.
+    if (nextSeq !== undefined && nextSeq > this.#seq) {
+      this.#seq = nextSeq;
+    }
     for (const msg of messages) {
       if (msg.type !== "session.message") continue;
       // Reconcile tool calls frozen in a non-terminal phase (streaming /
@@ -1797,7 +1804,12 @@ export class Session {
       case "text_delta": {
         if (!this.#activeAssistantMsg) {
           this.#activeAssistantMsg = this.#makeMessage("assistant", "", this.#agentIdentity, []);
-          this.#persistAndBuffer(this.#activeAssistantMsg);
+          // Scrollback only — no transcript row, no chunker event. The buffer
+          // holds the message by reference so clients attaching mid-stream see
+          // it grow; the durable row and the chunker event are emitted once,
+          // with final content, by #commitStreamed. Feeding the chunker an
+          // empty assistant message here would emit a promptless half-episode.
+          this.#scrollback.push(this.#activeAssistantMsg);
           this.#broadcastRaw(this.#activeAssistantMsg);
           if (this.#status === "tool_running") this.#setStatus("thinking");
         }
@@ -1823,7 +1835,7 @@ export class Session {
         if (this.#activeAssistantMsg) {
           this.#activeAssistantMsg.content = event.content;
           this.#activeAssistantMsg.parts = [{ kind: "text", text: event.content, markdown: true }];
-          this.#persistAndBuffer(this.#activeAssistantMsg);
+          this.#commitStreamed(this.#activeAssistantMsg);
           this.#broadcastRaw(this.#activeAssistantMsg);
           this.#activeAssistantMsg = null;
         } else if (event.content) {
@@ -1842,7 +1854,9 @@ export class Session {
           this.#finalizeActiveThinking();
           this.#activeThinkingMsg = this.#makeMessage("thinking", "", this.#agentIdentity, []);
           this.#activeThinkingIndex = event.blockIndex ?? null;
-          this.#persistAndBuffer(this.#activeThinkingMsg);
+          // Scrollback only — see the text_delta note; committed by
+          // #finalizeActiveThinking → #commitStreamed.
+          this.#scrollback.push(this.#activeThinkingMsg);
           this.#broadcastRaw(this.#activeThinkingMsg);
         }
         if (event.content) {
@@ -2093,7 +2107,7 @@ export class Session {
     if (!m.content || m.content.length === 0) {
       m.content = "(no output)";
     }
-    this.#persistAndBuffer(m);
+    this.#commitStreamed(m);
     this.#broadcastRaw(m);
   }
 
@@ -2119,7 +2133,8 @@ export class Session {
 
     const msg = this.#makeMessage("assistant", "", this.#agentIdentity, []);
     this.#activeAssistantMsg = msg;
-    this.#persistAndBuffer(msg);
+    // Scrollback only — committed with final content below.
+    this.#scrollback.push(msg);
     this.#broadcastRaw(msg);
 
     for (let pos = 0; pos < content.length; pos += charsPerStep) {
@@ -2138,24 +2153,9 @@ export class Session {
     }
 
     if (this.#activeAssistantMsg !== msg) return; // interrupted on last frame
-    const finalParts: ContentPart[] = [{ kind: "text", text: content, markdown: true }];
-    // Reset to the placeholder size so updateMessage measures the correct
-    // before/after byte delta — the buffer holds msg by reference, so
-    // mutations here are visible to the accounting logic inside updateMessage.
-    msg.content = "";
-    msg.parts = [];
-    // Do NOT call #persistAndBuffer again — it would push a second scrollback entry
-    // for the same messageId, causing duplicate messages on scrollback.replay.
-    // The updater sets final content/parts inside the buffer's size-accounting pass.
-    this.#scrollback.updateMessage(msg.messageId, (entry) => {
-      const sm = entry as SessionMessage;
-      sm.content = content;
-      sm.parts = finalParts;
-    });
-    this.#transcriptStore.append(this.id, msg, this.#seq++).catch((e) => {
-      console.error(`[codeoid/session ${this.id}] transcript append failed: ${e instanceof Error ? e.message : String(e)}`);
-    });
-    this.#chunker?.onMessage(msg);
+    msg.content = content;
+    msg.parts = [{ kind: "text", text: content, markdown: true }];
+    this.#commitStreamed(msg);
     this.#broadcastRaw(msg);
     this.#activeAssistantMsg = null;
   }
@@ -2175,7 +2175,7 @@ export class Session {
     if (!m.content || m.content.length === 0) {
       m.content = "(reasoning elided)";
     }
-    this.#persistAndBuffer(m);
+    this.#commitStreamed(m);
     this.#broadcastRaw(m);
   }
 
@@ -2260,6 +2260,28 @@ export class Session {
     };
   }
 
+  /**
+   * Commit the final content of a streamed message. The message was pushed
+   * into scrollback (by reference) at stream start and grew in place via
+   * deltas; ScrollbackBuffer.push upserts by messageId, so this re-accounts
+   * the existing entry — or re-adds it if it was evicted mid-stream — without
+   * ever creating a duplicate. A second entry per messageId corrupts
+   * scrollback.replay: clients render the message twice and virtualizers
+   * keyed on messageId collide (the #50 bug class). The durable transcript
+   * row and the memory-chunker event are emitted here exactly once, with
+   * final content, so plain turns produce one user+assistant episode instead
+   * of two half-episodes.
+   */
+  #commitStreamed(msg: SessionMessage): void {
+    this.#scrollback.push(msg);
+    this.#transcriptStore.append(this.id, msg, this.#seq++).catch((e) => {
+      console.error(
+        `[codeoid/session ${this.id}] transcript append failed: ${e instanceof Error ? e.message : String(e)}`,
+      );
+    });
+    this.#chunker?.onMessage(msg);
+  }
+
   /** Persist to transcript + scrollback buffer + memory chunker */
   #persistAndBuffer(msg: SessionMessage): void {
     this.#scrollback.push(msg);