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..40052b89 --- /dev/null +++ b/docs/superpowers/plans/2026-09-05-remote-transcript-retention.md @@ -0,0 +1,87 @@ +# Remote transcript retention + +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. + +## 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. + +## 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. + +## 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/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 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: {