fix(daemon): persist the pending-output counter across empty incremental takes (STA-4297) (#14496)

* fix(daemon): persist the pending-output counter across empty incremental takes (STA-4297)

An empty incremental take advanced pendingOutputSeq without writing a log
batch, so the in-memory counter ran permanently ahead of the log. The next
warm reattach could not prove continuity and committed the live 1000-row
window over a deep durable checkpoint.

Advance the counter only for takes that get persisted: a snapshot take
(stamped into the checkpoint) or one carrying records/overflow. This matches
the layers below, which already treat an empty take as a no-op write.

* test(daemon): keep empty-take coverage outcome-based

src/main/daemon/daemon-restore-scrollback-depth.test.ts

@@ -486,6 +486,55 @@ describe('STA-4091 previously recoverable restore depth', () => {
       )
     })
 
+    // Assert durable depth, not the sequence that merely enables it.
+    it('preserves durable depth when an empty incremental take precedes a warm reattach', async () => {
+      const { id } = await adapter.spawn({
+        cols: 80,
+        rows: 24,
+        sessionId: 'empty-take-depth',
+        cwd: '/tmp'
+      })
+      lastSubprocess.emitData(numberedOutput(DESKTOP_TERMINAL_SCROLLBACK_ROWS_DEFAULT))
+      await adapter.getBufferSnapshot(id)
+      lastSubprocess.emitData(`${FRESH_AFTER_CHECKPOINT}\r\n`)
+
+      const oldInternals = adapter as unknown as { checkpointDirtySessions: () => Promise<void> }
+      await oldInternals.checkpointDirtySessions()
+      const beforeReattach = await new HistoryReader(historyDir).detectColdRestore(id, {
+        ignoreCleanEnd: true
+      })
+      expect(snapshotText(beforeReattach ?? {})).toContain(OLDEST_WRITTEN_LINE)
+
+      // The trigger: a dirty mark with no new PTY records (the mock swallows the write, nothing echoes back).
+      adapter.write(id, 'noop')
+      await oldInternals.checkpointDirtySessions()
+
+      simulateAdapterCrash(adapter)
+      adapter = new DaemonPtyAdapter({
+        socketPath: getDaemonSocketPath(dir),
+        tokenPath: join(dir, 'test.token'),
+        historyPath: historyDir
+      })
+
+      const reattach = await adapter.spawn({ cols: 80, rows: 24, sessionId: id, cwd: '/tmp' })
+      expect(reattach.snapshot).toContain(FRESH_AFTER_CHECKPOINT)
+      expect(reattach.snapshot).toContain(OLDEST_WRITTEN_LINE)
+
+      // First post-reattach compact runs the continuity proof; it must not flatten the deep checkpoint.
+      const newInternals = adapter as unknown as {
+        checkpointDirtySessions: () => Promise<void>
+      }
+      adapter.write(id, 'noop')
+      await newInternals.checkpointDirtySessions()
+
+      const restored = await new HistoryReader(historyDir).detectColdRestore(id, {
+        ignoreCleanEnd: true
+      })
+      expect(snapshotText(restored ?? {})).toContain(OLDEST_WRITTEN_LINE)
+      expect(snapshotText(restored ?? {})).toContain(PREVIOUSLY_RECOVERABLE_LINE)
+      expect(restored?.scrollbackLines).toBeGreaterThan(DAEMON_SESSION_SCROLLBACK_ROWS)
+    })
+
     it('falls back to the live window when durable history cannot be read', async () => {
       const { id } = await adapter.spawn({
         cols: 80,

src/main/daemon/session.ts

@@ -494,7 +494,10 @@ export class Session {
     this.pendingOutputRecords = []
     this.pendingOutputBytes = 0
     this.pendingOutputOverflowed = false
-    this.pendingOutputSeq += 1
+    // Empty incremental takes are not persisted; advancing them would create a false reattach gap.
+    if (includeSnapshot || records.length > 0 || overflowed) {
+      this.pendingOutputSeq += 1
+    }
     return {
       records: includeSnapshot
         ? releasedHeldBytes

src/main/daemon/types.ts

@@ -285,9 +285,10 @@ export type TakePendingOutputResult = {
   /** Drained pending queue. Absent on older daemons. includeSnapshot still
    *  keeps `records` as held-only so mixed-version adapters do not double-replay. */
   drainedRecords?: PendingOutputRecord[]
-  /** Monotonic per-session batch sequence. The history log stores it so the
+  /** Non-decreasing per-session batch sequence. The history log stores it so the
    *  cold-restore reader can detect a lost batch (gap) and discard the log
-   *  instead of replaying a stream with missing bytes. */
+   *  instead of replaying a stream with missing bytes. Snapshot, record, and
+   *  overflow takes advance it; empty incremental takes repeat the prior value. */
   seq: number
   /** True when the session's pending buffer exceeded its cap and records were
    *  dropped. The caller must fall back to a full snapshot checkpoint. */