From ea74fcc36d8d76f4204e6176b1708121051b6126 Mon Sep 17 00:00:00 2001 From: Julius Olsson Date: Sat, 5 Sep 2026 02:37:40 -0700 Subject: [PATCH 1/3] docs(remote): plan view-owned transcript retention --- .../2026-09-05-remote-transcript-retention.md | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-05-remote-transcript-retention.md diff --git a/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md b/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md new file mode 100644 index 00000000..7c8d176f --- /dev/null +++ b/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md @@ -0,0 +1,41 @@ +# Remote transcript retention + +Status: planned. Issue #805. Depends on #813 (fix/remote-output-backpressure, +61823921); separate PR against that branch. Main remains 5d641845. No edits to +A5 desktop rendering/worktree paths or the external operator toolkit. + +## Intended behavior and implementation + +1. Treat actual view subscriptions as transcript ownership. Unviewed sessions + retain lightweight status/file identity only; do not map or fold their JSONL + and semantic bodies. Release entries, indexes, UUID bookkeeping, mapper and + semantic state on last unsubscribe. Invalidate pending history responses and + backfill again when selected. Partial semantic turns wait for a fresh start. +2. Reuse the desktop pure live trim planner and marker helpers without editing + desktop code or using its global registries. Apply count/estimated-byte + targets to active live appends; preserve current/history semantic ownership, + cross-entry tool pairs, stable UUIDs and pagination anchors. Suspend trimming + while paging and briefly after older history is loaded. +3. Rebuild tool indexes from the retained window. Track trimmed UUID tombstones + per view so old live replays cannot reappear at the tail, but explicit older + pages can reload them. These small identity sets last for the viewed window; + all are released on unsubscribe. Correct replay dedupe takes precedence over + bounding UUID count during one continuously viewed session. +4. Carry byte offsets from history chunks into pagination and trim cursors. + Live records lacking offsets use provider markers. Preserve raw-record groups + at a trim boundary so multi-entry mapper output remains reloadable in order. +5. Add regressions for multi-session zero-view retention, detach/reselect and + pending responses, sustained count/byte trimming, replay dedupe, older-page + ordering/exact offsets, tool pair and semantic owner preservation. Adapt + transport tests to explicitly own a view where they assert live rendering. + +## Checks, evidence and limitations + +Run remote suites, relevant shared trim tests, typecheck, test contract, client +production build and diff check; full repository checks in CI. Compare retained +logical payload/cardinality against the audit's synthetic 4096 entries/32 MiB, +not production heap/latency claims. Safety constraints may pin an active window +above its target; never drop active ownership merely to hit a number. Explicit +older-history reading may exceed the live target during its grace period. +Synchronize issue/PR acceptance criteria and document these constraints. Review +checks/feedback, leave clean committed worktrees; do not merge. From 62bc322cf92e1f8b0c1563e960f2c9079ad44967 Mon Sep 17 00:00:00 2001 From: Julius Olsson Date: Sat, 5 Sep 2026 02:45:09 -0700 Subject: [PATCH 2/3] perf(remote): release unviewed transcripts and trim live windows Make view subscriptions own transcript bodies, mapper state and replay identities. Reuse the pure desktop trim policy while preserving tool pairs, semantic owners and exact history offsets; rebuild indexes and retain identity-only tombstones for safe replay and pagination. Refs #805 --- .../2026-09-05-remote-transcript-retention.md | 34 ++- .../WebSocketSessionFeed.integration.test.ts | 4 +- .../src/transcript/store.reconnect.test.ts | 3 + .../src/transcript/store.retention.test.ts | 271 ++++++++++++++++++ src/remote-client/src/transcript/store.ts | 150 ++++++++-- 5 files changed, 443 insertions(+), 19 deletions(-) create mode 100644 src/remote-client/src/transcript/store.retention.test.ts diff --git a/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md b/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md index 7c8d176f..67fe1ccb 100644 --- a/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md +++ b/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md @@ -1,6 +1,6 @@ # Remote transcript retention -Status: planned. Issue #805. Depends on #813 (fix/remote-output-backpressure, +Status: implemented and locally verified. Issue #805. Depends on #813 (fix/remote-output-backpressure, 61823921); separate PR against that branch. Main remains 5d641845. No edits to A5 desktop rendering/worktree paths or the external operator toolkit. @@ -39,3 +39,35 @@ above its target; never drop active ownership merely to hit a number. Explicit older-history reading may exceed the live target during its grace period. Synchronize issue/PR acceptance criteria and document these constraints. Review checks/feedback, leave clean committed worktrees; do not merge. + +## Implemented evidence and constraints + +Ten remote retention regressions use real Claude/Codex/OpenCode mappers and +semantic folds. The zero-view regression fails against the parent #813 store +with 4096 retained entries; after the fix all three synthetic sessions retain +zero entries, tool indexes and seen UUIDs. Local remote plus shared window tests +pass (137 tests / 14 files). This is logical payload/cardinality evidence, not a +browser heap or production throughput benchmark. + +A 2100-entry live burst trims to 1500 with matching indexes and totalEntries +unchanged. A 300-entry, 128 KiB-per-entry burst triggers the byte budget below +the count threshold and trims under the shared 24 MiB estimate target. Exact +history offsets survive trims and all-duplicate pages. Older pages reload +trimmed UUIDs in order, while live replay cannot append them at the tail. +Tool-result indexes rebuild chronologically before notifying subscribers so +historical duplicate tool ids cannot overwrite newer retained results. + +The pure shared planner retains paired tool entries and semantic owners. The +remote adapter additionally refuses cuts inside one raw provider record; it +never adjusts a planned cut in a way that could invalidate pair safety. +Identity-only tombstones remain for a continuously viewed session, and safety +constraints/history-reading grace can exceed the nominal count/byte targets. +All body and identity state is released on last unsubscribe. Live frames lack +byte offsets and use the provider marker fallback until history supplies an +exact cursor. No protocol migration or desktop implementation change is needed. + +Typecheck, test contract, client production build and diff check pass; the +existing client chunk-size/mixed-import warnings remain. Re-selection also +clears the authority of a file hint observed while unviewed, so a newer history +file can establish identity without waiting for another live append. New live +frames in the selected view still win over a stale history reply. diff --git a/src/remote-client/src/WebSocketSessionFeed.integration.test.ts b/src/remote-client/src/WebSocketSessionFeed.integration.test.ts index c8ffe773..2565813b 100644 --- a/src/remote-client/src/WebSocketSessionFeed.integration.test.ts +++ b/src/remote-client/src/WebSocketSessionFeed.integration.test.ts @@ -277,11 +277,11 @@ describe('WebSocketSessionFeed against a live RemoteServer', () => { ) as { input: { entries: Array> } } const durableQueuedPrompt = recordedBundle.input.entries[13]! await writeFile(transcript, disk.map(d => JSON.stringify(d)).join('\n') + '\n', 'utf8') - ;(manager.resolveTranscriptFile as ReturnType).mockResolvedValue(transcript) const f = makeFeed() const store = new TranscriptStore(f) await waitForOpen(f) + store.subscribe('s1', () => {}) manager.emit('started', { sessionId: 's1', kind: 'claude', projectDir: '/repo' }) // Live entry arrives FIRST (before backfill) — the desktop-order case. @@ -311,6 +311,8 @@ describe('WebSocketSessionFeed against a live RemoteServer', () => { ) // Backfill prepends the older records and skips the duplicate. + ;(manager.resolveTranscriptFile as ReturnType).mockResolvedValue(transcript) + await vi.waitFor(() => expect(store.getSnapshot('s1').loadingOlderHistory).toBe(false)) await store.loadInitialHistory('s1') await vi.waitFor(() => expect(store.getSnapshot('s1').entries.map(e => e.uuid)).toEqual([ diff --git a/src/remote-client/src/transcript/store.reconnect.test.ts b/src/remote-client/src/transcript/store.reconnect.test.ts index 7b5b5420..6042272f 100644 --- a/src/remote-client/src/transcript/store.reconnect.test.ts +++ b/src/remote-client/src/transcript/store.reconnect.test.ts @@ -23,6 +23,7 @@ function fixture() { } }, }) as unknown as WebSocketSessionFeed + getHistory.mockResolvedValue({ ok: false, error: 'No transcript yet' }) const store = new TranscriptStore(feed) return { store, getHistory, list, emit: (name: string, value: unknown) => { for (const cb of listeners.get(name) ?? []) cb(value) } } } @@ -62,6 +63,7 @@ describe('remote transcript reconnect recovery', () => { it('ignores an in-flight old history reply after disconnect, even for the same file', async () => { const f = fixture() + f.store.subscribe('s', () => {}) const old = deferred<{ ok: true; chunk: HistoryChunkResult }>() try { f.getHistory.mockReturnValueOnce(old.promise) @@ -78,6 +80,7 @@ describe('remote transcript reconnect recovery', () => { it('uses committed content after reconnect until a complete new semantic turn starts', () => { const f = fixture() try { + f.store.subscribe('s', () => {}) f.store.getSnapshot('s') f.emit('onConnectionState', 'closed') f.emit('onSessionSemanticEvent', { sessionId: 's', event: { type: 'block_started', source: 'proxy', turnId: 'lost-prefix', blockId: 'b', blockType: 'text' } }) diff --git a/src/remote-client/src/transcript/store.retention.test.ts b/src/remote-client/src/transcript/store.retention.test.ts new file mode 100644 index 00000000..fbabbb30 --- /dev/null +++ b/src/remote-client/src/transcript/store.retention.test.ts @@ -0,0 +1,271 @@ +import { describe, expect, it, vi } from 'vitest' +import type { WebSocketSessionFeed } from '../WebSocketSessionFeed' +import type { HistoryChunkResult } from '../wire' +import { TranscriptStore } from './store' +import { estimateLiveEntriesBytes, MAX_LIVE_ENTRY_BYTES, TRIM_TO_LIVE_ENTRY_BYTES } from '@renderer/session-runtime/liveEntryWindow' + +const FILE = '/synthetic/transcript.jsonl' +const EPOCH = Date.parse('2026-01-01T00:00:00Z') +const raw = (i: number, bytes = 0): Record => ({ + type: 'assistant', uuid: `u-${i}`, timestamp: new Date(EPOCH + i).toISOString(), + message: { role: 'assistant', content: [ + { type: 'text', text: bytes ? 'x'.repeat(bytes) : `synthetic ${i}` }, + { type: 'tool_use', id: `tool-${i}`, name: 'Read', input: { path: `/synthetic/${i}` } }, + ] }, +}) +const range = (start: number, end: number, bytes = 0) => Array.from({ length: end - start }, (_, i) => raw(start + i, bytes)) +const page = (entries: Array>, hasMore = false, offsets?: number[]): HistoryChunkResult => ({ entries, file: FILE, hasMore, offsets }) + +function fixture(kind = 'claude') { + const listeners = new Map void>>() + let now = EPOCH + const list = ['a', 'b', 'c'].map(sessionId => ({ sessionId, kind, alive: true, cwd: '/synthetic', lastActivityAt: 0 })) + const getHistory = vi.fn<(...args: unknown[]) => Promise<{ ok: true; chunk: HistoryChunkResult } | { ok: false; error: string }>>() + .mockResolvedValue({ ok: true, chunk: page([]) }) + const methods = { getHistory, getSessionList: () => list } + const feed = new Proxy(methods, { + get(target, key: string) { + if (key in target) return target[key as keyof typeof target] + return (cb: (value: unknown) => void) => { + let set = listeners.get(key) + if (!set) listeners.set(key, set = new Set()) + set.add(cb) + return () => set.delete(cb) + } + }, + }) as unknown as WebSocketSessionFeed + const store = new TranscriptStore(feed, () => now) + const emit = (name: string, value: unknown) => { for (const cb of listeners.get(name) ?? []) cb(value) } + return { + store, list, getHistory, emit, + advance: (ms: number) => { now += ms }, + live: (entries: Array>, sessionId = 'a') => emit('onSessionJsonlEntries', { sessionId, entries: entries.map(entry => ({ entry, file: FILE })) }), + semantic: (event: unknown, sessionId = 'a') => emit('onSessionSemanticEvent', { sessionId, event }), + view: async (sessionId = 'a') => { const unsub = store.subscribe(sessionId, () => {}); await store.loadInitialHistory(sessionId); return unsub }, + } +} +// Collection cardinalities establish retention, not browser heap usage. Inspect +// only identity sets; observable snapshots below check actual rendered bodies. +function bookkeeping(store: TranscriptStore, id = 'a') { + return (Reflect.get(store, 'sessions') as Map; trimmed: Set; liveMapper: unknown }>).get(id)! +} + +describe('view-owned remote transcript retention', () => { + it('retains no transcript bodies, indexes, identities or semantic state for several unviewed sessions', async () => { + const f = fixture() + try { + // Port of #805: 4096 unique tool/text entries, 32 MiB logical ASCII text + // per session. These are synthetic bodies, never private transcripts. + for (const { sessionId } of f.list) { + for (let start = 0; start < 4096; start += 128) f.live(range(start, start + 128, 8192), sessionId) + f.semantic({ type: 'turn_started', turnId: 'unviewed' }, sessionId) + f.semantic({ type: 'block_started', blockId: 'b', blockType: 'text', text: 'x'.repeat(8192) }, sessionId) + const t = f.store.getSnapshot(sessionId) + expect(t.entries).toHaveLength(0) + expect(t.toolUseIndex.size).toBe(0) + expect(t.toolResultIndex.size).toBe(0) + expect(t.semanticTurn).toBeNull() + expect(t.semanticHistory).toEqual([]) + expect(bookkeeping(f.store, sessionId).seen.size).toBe(0) + expect(bookkeeping(f.store, sessionId).liveMapper).toBeNull() + } + await f.store.loadInitialHistory('a') + expect(f.getHistory).not.toHaveBeenCalled() + } finally { f.store.dispose() } + }) + + it('releases the last view, invalidates pending history and backfills on reselect while preserving status', async () => { + const f = fixture() + try { + const leave = await f.view() + const leaveSecond = f.store.subscribe('a', () => {}) + f.live(range(0, 20)) + f.semantic({ type: 'turn_started', turnId: 'old' }) + f.emit('onSessionProcessState', { sessionId: 'a', active: true, status: 'Working' }) + leave() + expect(f.store.getSnapshot('a').entries).toHaveLength(20) + leaveSecond() + expect(f.store.getSnapshot('a').entries).toHaveLength(0) + expect(f.store.getSnapshot('a').workingStatus).toBe('Working') + expect(bookkeeping(f.store).seen.size).toBe(0) + expect(bookkeeping(f.store).trimmed.size).toBe(0) + expect(bookkeeping(f.store).liveMapper).toBeNull() + + let resolve!: (result: { ok: true; chunk: HistoryChunkResult }) => void + f.getHistory.mockReturnValueOnce(new Promise(r => { resolve = r })) + const pendingView = f.store.subscribe('a', () => {}) + const pending = f.store.loadInitialHistory('a') + pendingView() + resolve({ ok: true, chunk: page(range(0, 20)) }) + await pending + expect(f.store.getSnapshot('a').entries).toHaveLength(0) + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: { ...page(range(200, 210), true), file: '/synthetic/rolled-while-unviewed.jsonl' } }) + await f.view() + expect(f.store.getSnapshot('a').entries.map(e => e.uuid)).toEqual(range(200, 210).map(e => e.uuid)) + f.semantic({ type: 'block_started', turnId: 'old', blockId: 'suffix', blockType: 'text' }) + expect(f.store.getSnapshot('a').semanticTurn).toBeNull() + f.semantic({ type: 'turn_started', turnId: 'new' }) + expect(f.store.getSnapshot('a').semanticTurn?.turnId).toBe('new') + f.emit('onSessionList', []) + expect((Reflect.get(f.store, 'sessions') as Map).size).toBe(0) + } finally { f.store.dispose() } + }) + + it('trims sustained live entries and indexes, rejects old replay, and restores trimmed history in order', async () => { + const f = fixture() + try { + const leave = await f.view() + f.live(range(0, 2100)) + let t = f.store.getSnapshot('a') + expect(t.entries.map(e => e.uuid)).toEqual(range(600, 2100).map(e => e.uuid)) + expect(t.totalEntries).toBe(2100) + expect(t.toolUseIndex.size).toBe(1500) + expect(t.toolUseIndex.has('tool-0')).toBe(false) + const version = t.toolIndexVersion + f.live(range(0, 600)) + expect(f.store.getSnapshot('a').entries).toHaveLength(1500) + expect(f.store.getSnapshot('a').totalEntries).toBe(2100) + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: page(range(400, 600), true) }) + await f.store.loadOlderHistory('a') + expect(f.getHistory).toHaveBeenLastCalledWith('a', { beforeMarker: 'u-600', limit: 200 }) + t = f.store.getSnapshot('a') + expect(t.entries.map(e => e.uuid)).toEqual(range(400, 2100).map(e => e.uuid)) + expect(t.toolUseIndex.has('tool-400')).toBe(true) + expect(t.toolIndexVersion).toBeGreaterThan(version) + expect(bookkeeping(f.store).trimmed.has('u-400')).toBe(false) + // History the user just requested is protected from immediate eviction. + f.live(range(2100, 2600)) + expect(f.store.getSnapshot('a').entries).toHaveLength(2200) + f.advance(30_001) + f.live(range(2600, 2601)) + t = f.store.getSnapshot('a') + expect(t.entries.map(e => e.uuid)).toEqual(range(1101, 2601).map(e => e.uuid)) + expect(t.toolUseIndex.size).toBe(1500) + leave() + expect(bookkeeping(f.store).seen.size).toBe(0) + expect(bookkeeping(f.store).trimmed.size).toBe(0) + } finally { f.store.dispose() } + }) + + it('uses a byte budget even below the count trigger', async () => { + const f = fixture() + try { + await f.view() + f.live(range(0, 300, 128 * 1024)) + const t = f.store.getSnapshot('a') + expect(300 * 128 * 1024).toBeGreaterThan(MAX_LIVE_ENTRY_BYTES) + expect(t.entries.length).toBeLessThan(300) + expect(estimateLiveEntriesBytes(t.entries)).toBeLessThanOrEqual(TRIM_TO_LIVE_ENTRY_BYTES) + expect(t.toolUseIndex.size).toBe(t.entries.length) + expect(t.hasOlderHistory).toBe(true) + } finally { f.store.dispose() } + }) + + it('preserves cross-boundary tool pairs and the newest result when older history repeats a tool id', async () => { + const f = fixture() + const result = (i: number, text: string) => ({ type: 'user', uuid: `r-${i}`, message: { role: 'user', content: [{ type: 'tool_result', tool_use_id: 'tool-100', content: text }] } }) + try { + await f.view() + f.live([...range(0, 2100), result(2100, 'latest')]) + const t = f.store.getSnapshot('a') + expect(t.entries[0]?.uuid).toBe('u-100') + expect(t.toolUseIndex.has('tool-100')).toBe(true) + expect(t.toolResultIndex.get('tool-100')?.content).toBe('latest') + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: page([...range(0, 100), result(99, 'historical')]) }) + await f.store.loadOlderHistory('a') + expect(f.store.getSnapshot('a').toolResultIndex.get('tool-100')?.content).toBe('latest') + } finally { f.store.dispose() } + }) + + it('keeps committed owners while their semantic turn remains a paint input', async () => { + const f = fixture() + try { + await f.view() + f.semantic({ type: 'turn_started', turnId: 'owned' }) + const started = f.store.getSnapshot('a').semanticTurn!.startedAt + const rows = range(0, 2100) + for (let i = 100; i < rows.length; i++) rows[i].timestamp = new Date(started + i).toISOString() + f.live(rows) + expect(f.store.getSnapshot('a').entries[0]?.uuid).toBe('u-100') + expect(f.store.getSnapshot('a').semanticTurn?.turnId).toBe('owned') + f.semantic({ type: 'turn_completed', turnId: 'owned' }) + f.live([{ ...raw(2100), timestamp: new Date(started + 2100).toISOString() }]) + expect(f.store.getSnapshot('a').entries[0]?.uuid).toBe('u-100') + expect(f.store.getSnapshot('a').semanticHistory.some(t => t.turnId === 'owned')).toBe(true) + } finally { f.store.dispose() } + }) + + it('carries exact history offsets through a trim and older pagination', async () => { + const f = fixture() + try { + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: page(range(0, 2000), true, Array.from({ length: 2000 }, (_, i) => 100 + i * 80)) }) + await f.view() + f.live(range(2000, 2100)) + expect(f.store.getSnapshot('a').entries[0]?.uuid).toBe('u-600') + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: page(range(400, 600), true, Array.from({ length: 200 }, (_, i) => 100 + (i + 400) * 80)) }) + await f.store.loadOlderHistory('a') + expect(f.getHistory).toHaveBeenLastCalledWith('a', { beforeMarker: 'u-600', beforeOffset: 48100, limit: 200 }) + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: page(range(200, 400)) }) + await f.store.loadOlderHistory('a') + expect(f.getHistory).toHaveBeenLastCalledWith('a', { beforeMarker: 'u-400', beforeOffset: 32100, limit: 200 }) + expect(f.store.getSnapshot('a').entries.map(e => e.uuid)).toEqual(range(200, 2100).map(e => e.uuid)) + } finally { f.store.dispose() } + }) + it('advances an all-duplicate history page by its exact offset rather than repeating an ambiguous marker', async () => { + const f = fixture() + try { + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: page([raw(9)], true, [900]) }) + await f.view() + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: page([raw(9)], true, [500]) }) + await f.store.loadOlderHistory('a') + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: page(range(0, 9), false) }) + await f.store.loadOlderHistory('a') + expect(f.getHistory).toHaveBeenLastCalledWith('a', { beforeMarker: 'u-9', beforeOffset: 500, limit: 200 }) + expect(f.store.getSnapshot('a').entries.map(e => e.uuid)).toEqual(range(0, 10).map(e => e.uuid)) + expect(f.store.getSnapshot('a').hasOlderHistory).toBe(false) + } finally { f.store.dispose() } + }) + + it('preserves OpenCode raw-record fan-out, cursor and tool ownership across trim and reload', async () => { + const f = fixture('opencode') + const message = (i: number) => ({ + info: { role: 'assistant', id: `m-${i}`, time: { created: EPOCH + i, completed: EPOCH + i + 1 } }, + parts: [{ type: 'tool', callID: `call-${i}`, tool: 'Read', state: { status: 'completed', input: { path: '/synthetic' }, output: `result ${i}` } }], + }) + try { + await f.view() + f.live(Array.from({ length: 1051 }, (_, i) => message(i))) + const t = f.store.getSnapshot('a') + expect(t.entries).toHaveLength(1500) + expect(t.entries[0]?.uuid).toBe('m-301') + expect(t.entries[1]?.uuid).toBe('m-301:result:call-301') + expect(t.toolUseIndex.size).toBe(750) + expect(t.toolResultIndex.size).toBe(750) + expect(t.toolResultIndex.has('call-0')).toBe(false) + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: page([message(300)], true, [12345]) }) + await f.store.loadOlderHistory('a') + expect(f.getHistory).toHaveBeenLastCalledWith('a', { beforeMarker: 'm-301', limit: 200 }) + expect(f.store.getSnapshot('a').entries.slice(0, 4).map(e => e.uuid)).toEqual(['m-300', 'm-300:result:call-300', 'm-301', 'm-301:result:call-301']) + } finally { f.store.dispose() } + }) + + it('does not carry a Codex live mapper turn cursor through an unviewed interval', async () => { + const f = fixture('codex') + const codexMessage = (id: string) => ({ type: 'response_item', timestamp: new Date(EPOCH).toISOString(), payload: { type: 'message', id, role: 'assistant', content: [{ type: 'output_text', text: id }] } }) + try { + const leave = await f.view() + f.live([{ type: 'turn_context', payload: { turn_id: 'old-turn' } }, codexMessage('old')]) + const oldMapper = bookkeeping(f.store).liveMapper + expect(oldMapper).not.toBeNull() + leave() + f.live([{ type: 'turn_context', payload: { turn_id: 'unviewed-turn' } }]) + expect(bookkeeping(f.store).liveMapper).toBeNull() + await f.view() + f.live([codexMessage('new')]) + expect(bookkeeping(f.store).liveMapper).not.toBe(oldMapper) + expect((bookkeeping(f.store).liveMapper as { getTurnCursor(): string | null }).getTurnCursor()).toBeNull() + expect(f.store.getSnapshot('a').entries).toHaveLength(1) + } finally { f.store.dispose() } + }) + +}) diff --git a/src/remote-client/src/transcript/store.ts b/src/remote-client/src/transcript/store.ts index 35385fa5..dd1c6820 100644 --- a/src/remote-client/src/transcript/store.ts +++ b/src/remote-client/src/transcript/store.ts @@ -3,6 +3,7 @@ import { getRendererProviderCapabilities } from '@providers/registry.renderer.ca import { foldSemanticEvent } from '@renderer/session-runtime/semantic/foldEvent' import { reduceStreamPhase } from '@renderer/session-runtime/semantic/streamPhaseMachine' import type { StreamPhaseState } from '@renderer/session-runtime/semantic/streamPhaseMachine' +import { historyMarkerOf, stampHistoryMarker, planLiveEntryTrim, OLDER_PREPEND_TRIM_GRACE_MS } from '@renderer/session-runtime/liveEntryWindow' import { indexEntryIntoMaps } from '@renderer/session-runtime/entries' import { emptySemanticRuntime } from '@renderer/session-runtime/state' import type { SemanticLiveTurn, SemanticRuntimeState } from '@renderer/session-runtime/state' @@ -80,6 +81,11 @@ type SessionState = { transcript: SessionTranscript semantic: SemanticRuntimeState seen: Set + // Tombstones contain identities only, never entry/tool bodies. Live replay + // must not append trimmed old rows at the tail; explicit pagination may + // reload them. Both sets die when the last view releases this session. + trimmed: Set + olderPrependAt: number | null /** Session-lifetime mapper for LIVE entries only (codex cursor). Created * lazily once the kind is KNOWN from the session list — memoizing a * mapper built on a fallback guess would bake the wrong provider in for @@ -90,6 +96,7 @@ type SessionState = { * Disagreement between the two = the provider rolled the transcript. */ transcriptFile: string | null historyOldestMarker: string | null + historyOldestOffset: number | undefined historyLoaded: boolean historyLoading: boolean awaitingSemanticStart: boolean @@ -131,12 +138,18 @@ function emptyTranscript(): SessionTranscript { } } +// Entry-scoped cursor metadata dies with the entry. A single raw provider line +// may yield multiple feed entries: they are one pagination unit and must never +// be split by a trim. Offset is available on history replies, not live frames. +const entryCursors = new WeakMap() +const NO_GHOSTS: ReadonlyMap = new Map() + export class TranscriptStore { private readonly sessions = new Map() private readonly listeners = new Map void>>() private readonly unsubs: Array<() => void> = [] - constructor(private readonly feed: WebSocketSessionFeed) { + constructor(private readonly feed: WebSocketSessionFeed, private readonly now = Date.now) { this.unsubs.push( feed.onConnectionState(connection => { if (connection !== 'closed') return @@ -225,8 +238,26 @@ export class TranscriptStore { set = new Set() this.listeners.set(sessionId, set) } + const state = this.sessions.get(sessionId) + if (set.size === 0 && state) { + // File identity observed while unviewed is only a hint. A provider may + // have rolled without another forwarded entry; let the new backfill + // establish identity unless a live frame in THIS view wins the race. + state.transcriptFile = null + } set.add(cb) - return () => set?.delete(cb) + return () => { + set.delete(cb) + if (set.size > 0 || this.listeners.get(sessionId) !== set) return + this.listeners.delete(sessionId) + // Selecting another session/list screen is an ownership boundary, not + // merely a render pause. Keeping the last snapshot would retain the + // entire transcript through indexes, semantic folds and mapper state. + const file = this.state(sessionId).transcriptFile + this.resetTranscript(sessionId) + this.state(sessionId).transcriptFile = file + this.state(sessionId).awaitingSemanticStart = true + } } getSnapshot(sessionId: string): SessionTranscript { @@ -243,12 +274,13 @@ export class TranscriptStore { // --- backfill --- - /** Load the initial newest-N chunk once per session. Prepends behind any + /** Load the initial newest-N chunk once per viewed window. Prepends behind any * live entries that already arrived — the shared seen-set makes the * overlap safe, exactly like the desktop's initialHistory action. * Failure leaves the flags retryable; retries fire from the session-list * hook above and from live-entry arrival. */ async loadInitialHistory(sessionId: string): Promise { + if (!this.isViewed(sessionId)) return const state = this.state(sessionId) if (state.historyLoaded || state.historyLoading || state.transcript.historyError === REMOTE_HISTORY_TOO_LARGE) return state.historyLoading = true @@ -287,8 +319,12 @@ export class TranscriptStore { sessionId, result.chunk.entries as Array>, 'prepend', + result.chunk.offsets, ) - if (marker) state.historyOldestMarker = marker + if (marker) { + state.historyOldestMarker = marker.marker + state.historyOldestOffset = marker.offset + } this.mutate(sessionId, t => ({ ...t, loadingOlderHistory: false, @@ -299,9 +335,11 @@ export class TranscriptStore { hasOlderHistory: result.chunk.hasMore && state.historyOldestMarker !== null, totalEntries: result.chunk.totalEntries ?? t.entries.length, })) + this.trimLiveWindow(sessionId) } async loadOlderHistory(sessionId: string): Promise { + if (!this.isViewed(sessionId)) return const state = this.state(sessionId) if (state.transcript.loadingOlderHistory || !state.transcript.hasOlderHistory) return const beforeMarker = state.historyOldestMarker @@ -313,7 +351,11 @@ export class TranscriptStore { return } this.mutate(sessionId, t => ({ ...t, loadingOlderHistory: true })) - const result = await this.feed.getHistory(sessionId, { beforeMarker, limit: 200 }) + const result = await this.feed.getHistory(sessionId, { + beforeMarker, + ...(state.historyOldestOffset === undefined ? {} : { beforeOffset: state.historyOldestOffset }), + limit: 200, + }) if (this.sessions.get(sessionId) !== state) return if (!result.ok || this.chunkFileConflicts(state, result.chunk?.file)) { this.mutate(sessionId, t => ({ @@ -324,12 +366,17 @@ export class TranscriptStore { })) return } + state.olderPrependAt = this.now() const marker = this.ingestRawEntries( sessionId, result.chunk.entries as Array>, 'prepend', + result.chunk.offsets, ) - if (marker) state.historyOldestMarker = marker + if (marker) { + state.historyOldestMarker = marker.marker + state.historyOldestOffset = marker.offset + } this.mutate(sessionId, t => ({ ...t, loadingOlderHistory: false, @@ -350,13 +397,16 @@ export class TranscriptStore { transcript: emptyTranscript(), semantic: emptySemanticRuntime(), seen: new Set(), + trimmed: new Set(), + olderPrependAt: null, liveMapper: null, kind: null, transcriptFile: null, historyOldestMarker: null, + historyOldestOffset: undefined, historyLoaded: false, historyLoading: false, - awaitingSemanticStart: false, + awaitingSemanticStart: true, } this.sessions.set(sessionId, state) } @@ -429,6 +479,8 @@ export class TranscriptStore { this.state(sessionId).transcriptFile = file } + if (!this.isViewed(sessionId)) return + this.ingestRawEntries( sessionId, items.map(x => x.entry as Record), @@ -452,6 +504,7 @@ export class TranscriptStore { conditions: prev.transcript.conditions, workingStatus: prev.transcript.workingStatus, screenText: prev.transcript.screenText, + exited: prev.transcript.exited, }, // The fold state resets WITH the transcript (review finding — this used // to carry prev.semantic across the roll). ingestSemanticEvent folds @@ -470,20 +523,23 @@ export class TranscriptStore { // the normal path this reset runs before any new-turn semantic event // exists, and everything it wipes belongs to the OLD conversation — // exactly the intent. If jsonl-watcher latency ever inverted that - // ordering, the loss is bounded and self-healing: foldSemanticEvent - // re-opens a turn from later `turn_started`/`turn_delta` events, and - // the turn's content still commits through the jsonl entries — a brief + // ordering, the turn's content still commits through the jsonl entries. + // Semantic painting waits for a fresh turn_started (a suffix alone has + // no authority after a reset). This trades a temporary // live-streaming gap versus guaranteed cross-conversation contamination // the other way. semantic: emptySemanticRuntime(), seen: new Set(), + trimmed: new Set(), + olderPrependAt: null, liveMapper: null, kind: prev.kind, transcriptFile: null, historyOldestMarker: null, + historyOldestOffset: undefined, historyLoaded: false, historyLoading: false, - awaitingSemanticStart: false, + awaitingSemanticStart: true, }) const set = this.listeners.get(sessionId) if (set) for (const cb of [...set]) cb() @@ -496,31 +552,36 @@ export class TranscriptStore { sessionId: string, raws: Array>, mode: 'append' | 'prepend', - ): string | null { + offsets?: number[], + ): { marker: string; offset?: number } | null { if (raws.length === 0) return null const state = this.state(sessionId) const mapper = mode === 'append' ? this.liveMapperOf(sessionId) : this.chunkMapper(sessionId) const kept: Entry[] = [] - let firstKeptMarker: string | null = null + let firstKeptMarker: { marker: string; offset?: number } | null = null let toolIndexChanged = false - for (const raw of raws) { + for (const [rawIndex, raw] of raws.entries()) { const mapped = mapper.map(raw) + const cursor = { group: {}, offset: offsets?.[rawIndex] } if ( mode === 'prepend' && firstKeptMarker === null && mapped.entries.length > 0 && mapped.historyMarker ) { - firstKeptMarker = mapped.historyMarker + firstKeptMarker = { marker: mapped.historyMarker, offset: cursor.offset } } for (const entry of mapped.entries) { const uuid = typeof entry.uuid === 'string' ? entry.uuid : null if (uuid) { - if (state.seen.has(uuid)) continue + if (state.seen.has(uuid) && !(mode === 'prepend' && state.trimmed.has(uuid))) continue state.seen.add(uuid) + state.trimmed.delete(uuid) } + stampHistoryMarker(entry, mapped.historyMarker) + entryCursors.set(entry, cursor) kept.push(entry) if ( indexEntryIntoMaps( @@ -536,16 +597,71 @@ export class TranscriptStore { if (kept.length === 0 && !toolIndexChanged) return firstKeptMarker + const entries = mode === 'append' ? [...state.transcript.entries, ...kept] : [...kept, ...state.transcript.entries] + if (mode === 'prepend' && toolIndexChanged) { + // Older pages may repeat a tool id with an earlier body. Replaying the + // complete retained order keeps the newest block authoritative instead + // of letting a prepended historical result overwrite a live result. + // Finish this before notifying subscribers: a snapshot's version must + // never advertise indexes that still contain historical winners. + state.transcript.toolUseIndex.clear() + state.transcript.toolResultIndex.clear() + for (const entry of entries) { + indexEntryIntoMaps(entry, state.transcript.toolUseIndex, state.transcript.toolResultIndex) + } + } this.mutate(sessionId, t => ({ ...t, - entries: mode === 'append' ? [...t.entries, ...kept] : [...kept, ...t.entries], + entries, totalEntries: t.totalEntries + (mode === 'append' ? kept.length : 0), toolIndexVersion: toolIndexChanged ? t.toolIndexVersion + 1 : t.toolIndexVersion, })) + if (mode === 'append') this.trimLiveWindow(sessionId) return firstKeptMarker } + private isViewed(sessionId: string): boolean { + return (this.listeners.get(sessionId)?.size ?? 0) > 0 + } + + private trimLiveWindow(sessionId: string): void { + const state = this.state(sessionId) + const t = state.transcript + if (t.loadingOlderHistory) return + if (state.olderPrependAt !== null && this.now() - state.olderPrependAt < OLDER_PREPEND_TRIM_GRACE_MS) return + // Reuse only the desktop's pure safety policy. Remote has no ghosts and + // must not register ids in desktop-global tombstone/grace registries. + // Current/history semantic owners and cross-entry tool pairs can pin the + // window above its target; losing ownership to hit a cap repaints copies. + const plan = planLiveEntryTrim(t.entries, state.semantic, NO_GHOSTS) + if (!plan) return + const cut = plan.cut + // A raw line can map to several entries (OpenCode tool fan-out). A cursor + // addresses the whole line. Splitting it would make its trimmed children + // unreachable; moving the cut ourselves could invalidate tool-pair safety. + // Keep the window until a later burst permits a whole-record boundary. + if (entryCursors.get(t.entries[cut - 1])?.group === entryCursors.get(t.entries[cut])?.group) return + const marker = historyMarkerOf(t.entries[cut]) + if (!marker) return + const entries = t.entries.slice(cut) + for (const entry of t.entries.slice(0, cut)) state.trimmed.add(entry.uuid!) + state.historyOldestMarker = marker + state.historyOldestOffset = entryCursors.get(entries[0])?.offset + // Slicing entries alone leaves tool-result bodies retained by the indexes. + // Rebuild in chronological order so each id still resolves to its latest + // retained block. The planner guarantees a retained result keeps its use. + const toolUseIndex = new Map() + const toolResultIndex = new Map() + for (const entry of entries) indexEntryIntoMaps(entry, toolUseIndex, toolResultIndex) + this.mutate(sessionId, prev => ({ + ...prev, entries, toolUseIndex, toolResultIndex, + toolIndexVersion: prev.toolIndexVersion + 1, + hasOlderHistory: true, + })) + } + private ingestSemanticEvent(sessionId: string, event: unknown): void { + if (!this.isViewed(sessionId)) return const state = this.state(sessionId) const record = asRecord(event) if (!record) return From c4f175f0834c07ec4ba882ae639e4a362089fb8e Mon Sep 17 00:00:00 2001 From: Julius Olsson Date: Sat, 5 Sep 2026 17:20:33 -0700 Subject: [PATCH 3/3] fix(remote): resolve parser helpers from pinned source A clean checkout has no parser dist directory. The shared live-window planner introduced a runtime ghost-helper import, so mirror the desktop source alias rather than relying on a warmed workspace package build. Refs #805 --- .../2026-09-05-remote-transcript-retention.md | 14 ++++++++++++++ src/remote-client/vite.config.ts | 14 +++++++++++--- 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md b/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md index 67fe1ccb..40052b89 100644 --- a/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md +++ b/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md @@ -71,3 +71,17 @@ existing client chunk-size/mixed-import warnings remain. Re-selection also clears the authority of a file hint observed while unviewed, so a newer history file can establish identity without waiting for another live append. New live frames in the selected view still win over a stale history reply. + +## Clean-build review correction + +The first full CI run passed tests and coverage but failed the remote production +build: the shared planner introduces a runtime import of +`agent-transcript-parser/ghost`, whose package export points at unbuilt `dist/`. +A prebuilt parser in the original local dependency links had masked this. +The remote Vite config now aliases that pure leaf to pinned submodule source, +matching the desktop build convention without adding a ghost plane or changing +desktop files. The failure reproduced locally after pointing this worktree's +parser dependency at its own unbuilt checkout; the shared node_modules and +running app were left untouched. The same unbuilt checkout now passes the full `npm run test:package` gate. +The parser still has no dist directory. CI will rerun after rebasing onto +current main f7507980 (toolkit #812 and MCP repair #818). diff --git a/src/remote-client/vite.config.ts b/src/remote-client/vite.config.ts index e13e0283..ea0eb814 100644 --- a/src/remote-client/vite.config.ts +++ b/src/remote-client/vite.config.ts @@ -72,13 +72,21 @@ export default defineConfig({ find: 'workflow-mcp/state', replacement: resolve(src, '..', 'packages', 'workflow-mcp', 'src', 'state.ts'), }, + { + // The live-window planner imports the pure parser ghost helpers even + // though the phone has no ghost plane. Resolve pinned source just as + // the desktop build does: file: package exports point at dist/, which + // exists on a warmed developer checkout but not in a clean CI clone. + // Do not externalize it; the phone needs a self-contained browser bundle. + find: 'agent-transcript-parser/ghost', + replacement: resolve(src, '..', 'packages', 'agent-transcript-parser', 'src', 'ghost.ts'), + }, { find: '@renderer', replacement: resolve(src, 'renderer', 'src') }, { find: '@providers', replacement: resolve(src, 'providers') }, { find: '@shared', replacement: resolve(src, 'shared') }, { find: '@mcp', replacement: resolve(src, 'mcp') }, - // agent-transcript-parser is a file: workspace package; the desktop - // resolves it via node_modules symlink, which works here too — no - // alias needed. The headless packages never appear in renderer code. + // Other parser references are type-only. Keep runtime leaf imports + // source-aliased above; headless packages never enter the phone bundle. ], }, build: {