From 7d6bfef7b655a761434395a62781ce525b93a6c7 Mon Sep 17 00:00:00 2001 From: Narry <3047793+NarryG@users.noreply.github.com> Date: Mon, 19 Oct 2026 03:16:33 -0400 Subject: [PATCH] fix(omp): reconcile uncertain sends from owned native history --- docs/site/content/docs/agents/native-chat.mdx | 4 + src/main/acp/acp-launch-specs.ts | 2 + src/main/acp/acp-recovery-history.test.ts | 22 +- src/main/acp/acp-recovery-history.ts | 12 +- src/main/acp/acp-session-reopen-failure.ts | 37 ++ .../acp-structured-launch-resolution.test.ts | 81 ++-- .../acp/acp-structured-launch-resolution.ts | 34 +- .../journal-submission-reconciler.ts | 11 +- src/main/omp/omp-acp-stored-messages.test.ts | 394 ++++++++++++++++++ src/main/omp/omp-acp-stored-messages.ts | 185 ++++++++ src/main/omp/omp-session-cwd-lookup.test.ts | 52 +++ src/main/omp/omp-session-cwd-lookup.ts | 42 +- .../opencode-acp-stored-messages.test.ts | 2 + 13 files changed, 811 insertions(+), 67 deletions(-) create mode 100644 src/main/omp/omp-acp-stored-messages.test.ts create mode 100644 src/main/omp/omp-acp-stored-messages.ts diff --git a/docs/site/content/docs/agents/native-chat.mdx b/docs/site/content/docs/agents/native-chat.mdx index 1a812697e0ad..082d517f4a42 100644 --- a/docs/site/content/docs/agents/native-chat.mdx +++ b/docs/site/content/docs/agents/native-chat.mdx @@ -29,6 +29,10 @@ When Claude shows an **AskUserQuestion** (or similar structured permission/quest Chat UI ships on desktop for supported local and remote (paired server) agent sessions. The [mobile companion](/docs/mobile) reuses chat-style transcript patterns for the same paired sessions. +Orca uses saved native session history on the execution host to reconcile uncertain OMP sends. Only a uniquely owned session in the launch account and workspace can confirm delivery. Missing, unreadable, ambiguous or incomplete history — including a missing journal — leaves delivery uncertain, and Orca never automatically replays the message. A send without a saved turn also prevents Orca from silently treating a session that cannot reopen as unused. + +This recovery uses the existing native ACP execution support: local folder and repository workspaces, or the local runtime of a paired remote server. It does not add native ACP execution inside a desktop-managed WSL workspace or read SSH-host history from the desktop filesystem. + Transcript fidelity, streaming, and terminal parity are still under active tuning. Prefer the raw TUI when you need every OSC/status detail. diff --git a/src/main/acp/acp-launch-specs.ts b/src/main/acp/acp-launch-specs.ts index fb39656239ff..ee84d159aac3 100644 --- a/src/main/acp/acp-launch-specs.ts +++ b/src/main/acp/acp-launch-specs.ts @@ -11,6 +11,7 @@ import { directoryAccountBinding, type AcpAccountBinding } from './acp-account-b import { openCodeAcpAccountBinding } from '../opencode/opencode-structured-account-home' import { scrubOpenCodeAcpEnvironment } from '../opencode/opencode-acp-environment' import { openCodeStoredUserMessagesReader } from '../opencode/opencode-acp-stored-messages' +import { ompStoredUserMessagesReader } from '../omp/omp-acp-stored-messages' import type { AcpStoredUserMessagesReader } from './acp-recovery-history' import { isStableCliVersionFrom, isStableCliVersionOnLine } from '../agent-cli-version-probe' import type { TuiAgent } from '../../shared/tui-agent' @@ -278,6 +279,7 @@ const OMP_LAUNCH_SPEC: AcpLaunchSpec = { installDirectories: () => [], // Stable releases from 17.0.5, the release verified to serve `omp acp`. supportsVersion: (version) => isStableCliVersionFrom(version, '17.0.5'), + readStoredUserMessages: ompStoredUserMessagesReader(), // `omp models --json` lists every model without the `enabledModels` filter a chat applies, so it // could offer models a chat refuses; its chats' own listings fill the catalog instead. modelDiscovery: { diff --git a/src/main/acp/acp-recovery-history.test.ts b/src/main/acp/acp-recovery-history.test.ts index 97a4b230c1b5..ec2c4cac19ad 100644 --- a/src/main/acp/acp-recovery-history.test.ts +++ b/src/main/acp/acp-recovery-history.test.ts @@ -96,7 +96,7 @@ describe('ACP restart recovery from the agent store', () => { stored('msg_2', 'run the tests', 205) ]) expect(readStoredUserMessages).toHaveBeenCalledWith( - expect.objectContaining({ env: launch.env, providerSessionId: 'ses_1' }) + expect.objectContaining({ env: launch.env, cwd: launch.cwd, providerSessionId: 'ses_1' }) ) expect(window).toMatchObject({ boundaryConsistent: false, turnInFlight: true }) expect( @@ -184,6 +184,26 @@ describe('ACP restart recovery from the agent store', () => { expect(readStoredUserMessages).not.toHaveBeenCalled() }) + it('keeps a missing or newer journal uncertain without consulting provider history', async () => { + const load = journal(submission('held', 'run tests', 'unknown', 200)) + for (const unreadable of [null, { ...load, newer: { sequence: 3 } }]) { + const { window, readStoredUserMessages } = await windowFor(unreadable, [ + stored('delivered', 'run tests', 205) + ]) + expect(window).toBeNull() + expect(readStoredUserMessages).not.toHaveBeenCalled() + } + }) + + it('never treats an empty provider history as non-delivery', async () => { + const load = journal(submission('held', 'run tests', 'unknown', 200)) + const { window } = await windowFor(load, []) + expect(window).toMatchObject({ boundaryConsistent: false, turnInFlight: true, items: [] }) + expect( + reconcileSubmissions({ submissions: [...load.state.submissions.values()], history: window! }) + ).toMatchObject([{ clientMessageId: 'held', outcome: 'unknown' }]) + }) + it('reads nothing for an agent whose store Orca does not read', async () => { await expect( readAcpRecoveryHistory( diff --git a/src/main/acp/acp-recovery-history.ts b/src/main/acp/acp-recovery-history.ts index 07ee22aa410d..395920e11510 100644 --- a/src/main/acp/acp-recovery-history.ts +++ b/src/main/acp/acp-recovery-history.ts @@ -33,6 +33,8 @@ export type AcpStoredUserMessage = { /** Reads the user messages of one provider session from the agent's own store; null when it cannot. */ export type AcpStoredUserMessagesReader = (input: { env: Record + /** The workspace on the execution host that owns this launch and its store. */ + cwd: string providerSessionId: string signal: AbortSignal }) => Promise @@ -90,7 +92,14 @@ function recoveryWindow( return stored.flatMap(({ message, fingerprint }, at) => claimed.has(at) || message.createdAt < from ? [] - : [{ providerItemId: message.id, clientMessageId: null, payloadFingerprint: fingerprint }] + : [ + { + providerItemId: message.id, + clientMessageId: null, + payloadFingerprint: fingerprint, + createdAt: message.createdAt + } + ] ) } @@ -120,6 +129,7 @@ export async function readAcpRecoveryHistory( const messages = await waitForPromiseWithSignal( readStoredUserMessages({ env: launch.env, + cwd: launch.cwd, providerSessionId: launch.resume.sessionId, signal }), diff --git a/src/main/acp/acp-session-reopen-failure.ts b/src/main/acp/acp-session-reopen-failure.ts index 90c7442d24b6..6d0ab9d422ce 100644 --- a/src/main/acp/acp-session-reopen-failure.ts +++ b/src/main/acp/acp-session-reopen-failure.ts @@ -10,6 +10,10 @@ import type { AgentSessionProviderHandleLink } from '../../shared/agent-session- import type { ProviderTimelineEvent } from '../native-chat/agent-session-timeline/provider-timeline-event' import { AcpAuthRequiredError, AcpRpcError } from './acp-errors' import type { AcpStructuredLaunch } from './acp-structured-launch-resolution' +import type { JournalLoad } from '../native-chat/agent-session-journal/journal-open' +import { readAgentJournalTurn } from '../../shared/agent-session-turn-record' +import { isQueuedAgentJournalSubmission } from '../../shared/agent-session-queued-submission' +import { isProviderTimelineTurnInNamespace } from '../native-chat/agent-session-timeline/provider-timeline-identity' /** ACP's "resource not found": the agent holds no session under the id this chat proved. */ const ACP_RESOURCE_NOT_FOUND = -32002 @@ -63,3 +67,36 @@ export function acpSessionNotRestoredRow( } ] } + +/** Only a complete journal with no potentially delivered send can prove a session unused. */ +export function acpSessionHasNoExchange( + readJournal: (sessionId: string) => JournalLoad | null, + sessionId: string, + nativeId: string +): boolean { + let load: JournalLoad | null + try { + load = readJournal(sessionId) + } catch { + return false + } + if (!load) { + return false + } + if (load.damage || load.newer) { + return false + } + // A send can reach the provider before its turn row becomes durable. + for (const submission of load.state.submissions.values()) { + if (submission.dispatchState !== 'rejected' && !isQueuedAgentJournalSubmission(submission)) { + return false + } + } + for (const item of load.state.items.values()) { + const turn = readAgentJournalTurn(item.body) + if (turn && isProviderTimelineTurnInNamespace(turn.turnId, nativeId)) { + return false + } + } + return true +} diff --git a/src/main/acp/acp-structured-launch-resolution.test.ts b/src/main/acp/acp-structured-launch-resolution.test.ts index ace94e1f1558..7f2e5a720292 100644 --- a/src/main/acp/acp-structured-launch-resolution.test.ts +++ b/src/main/acp/acp-structured-launch-resolution.test.ts @@ -21,7 +21,8 @@ import { createProviderSpawnSpec } from '../provider-process/provider-process-su import { ACP_LAUNCH_SPECS, acpLaunchSpecFor } from './acp-launch-specs' import { agentChatPermissionModes } from '../../shared/agent-chat-permission-mode' import { STRUCTURED_CHILD_ENV_TO_DELETE } from '../runtime/structured-session-child-env' -import { acpSessionNotRestoredItem } from './acp-session-reopen-failure' +import { acpReopenTakeover, acpSessionNotRestoredItem } from './acp-session-reopen-failure' +import { AcpRpcError } from './acp-errors' const GROK = acpLaunchSpecFor('grok')! const identity = { @@ -61,40 +62,33 @@ function journalWithTurns(...providerSessions: string[]): JournalLoad { return { state, newer: null, damage: null } } -function resolver(record: AgentSessionRecord, readJournal: () => JournalLoad | null = () => null) { - const searched: (string | null | undefined)[] = [] +function resolver( + record: AgentSessionRecord, + readJournal: () => JournalLoad | null = () => null, + environment: Record = { PATH: '/usr/bin', HOME: '/home/user' } +) { return { - searched, resolve: createAcpStructuredLaunchResolver(GROK, { store: { getRecord: () => record, pinLaunchDirectory: vi.fn() }, readJournal, resolveWorkspacePath: async () => '/repo/worktree', - resolveEnvironment: async () => ({ PATH: '/usr/bin', HOME: '/home/user' }), + resolveEnvironment: async () => environment, resolveLaunchEnv: () => ({ GROK_EXTRA: '1' }), - resolveCommand: (command, options) => { - searched.push(options?.pathEnv) - return `/resolved/${command}` - } + resolveCommand: (command) => `/resolved/${command}` }) } } describe('ACP launch resolution', () => { - it('pins the record account home and searches the agent install dir', async () => { - const { resolve, searched } = resolver(grokRecord()) - const launch = await resolve({ identity }) - expect(launch).toMatchObject({ - command: '/resolved/grok', - args: ['agent', 'stdio'], - cwd: '/repo/worktree', - fullAccess: false, - resume: null - }) - expect(launch.env).toMatchObject({ - GROK_HOME: '/home/user/.grok-work', - GROK_EXTRA: '1' - }) - expect(searched[0]).toContain('/home/user/.grok-work/bin') + it('keeps the sealed account home when the ambient environment names another account', async () => { + const accountHome = { variable: 'GROK_HOME', path: '/accounts/work' } + const launch = await resolver(grokRecord({ accountHome }), undefined, { + PATH: '/usr/bin', + HOME: '/accounts/personal', + GROK_HOME: '/accounts/personal/.grok' + }).resolve({ identity }) + + expect(launch.env.GROK_HOME).toBe(accountHome.path) }) it('creates a new ACP session after clear instead of probing the old session', async () => { @@ -132,9 +126,9 @@ describe('ACP launch resolution', () => { observedAt: 1, handle: { transport: 'acp', agent: 'grok', nativeId: 'acp-1' } }) - const created = await resolver(grokRecord({ providerHandleChain: [link('created')] })).resolve({ - identity - }) + const created = await resolver(grokRecord({ providerHandleChain: [link('created')] }), () => + journalWithTurns() + ).resolve({ identity }) const handle: AgentSessionProviderHandle = { transport: 'acp', agent: 'grok', @@ -172,11 +166,42 @@ describe('ACP launch resolution', () => { }).resolve({ identity }) // Read only when asked: a reopen that works never replays the journal. expect(reads).toEqual([]) - expect(launch.resume?.mayBeUnsaved()).toBe(true) + expect(launch.resume?.mayBeUnsaved()).toBe(false) expect(await unsaved(() => journalWithTurns())).toBe(true) expect(await unsaved(() => journalWithTurns('acp-other'))).toBe(true) expect(await unsaved(() => journalWithTurns('acp-other', 'acp-1'))).toBe(false) + for (const dispatchState of ['pending', 'unknown', 'accepted'] as const) { + const load = journalWithTurns() + load.state.submissions.set('held', { + clientMessageId: 'held', + fence: 1, + payloadFingerprint: 'uncertain', + dispatchState, + providerItemId: null, + reason: null, + submittedAt: 100, + resolvedAt: dispatchState === 'accepted' ? 110 : null + }) + const resume = (await resolver(created, () => load).resolve({ identity })).resume! + expect(resume.mayBeUnsaved()).toBe(false) + expect( + acpReopenTakeover(new AcpRpcError(-32002, 'Session missing'), resume, { + over: false, + now: 200, + warn: vi.fn() + }) + ).toEqual({ replaces: { key: resume.key, reason: 'restore-failed', replacedAt: 200 } }) + const held = load.state.submissions.get('held')! + load.state.submissions.set('held', { ...held, dispatchState: 'rejected' }) + expect(resume.mayBeUnsaved()).toBe(true) + load.state.submissions.set('held', { + ...held, + dispatchState: 'pending', + handoverRecorded: true + }) + expect(resume.mayBeUnsaved()).toBe(true) + } // A journal that does not read whole, or at all, proves nothing. expect( await unsaved(() => ({ diff --git a/src/main/acp/acp-structured-launch-resolution.ts b/src/main/acp/acp-structured-launch-resolution.ts index d0ddf202dcd3..7b434ae26fe9 100644 --- a/src/main/acp/acp-structured-launch-resolution.ts +++ b/src/main/acp/acp-structured-launch-resolution.ts @@ -26,10 +26,8 @@ import { resolveInstalledRuntimeRelease } from '../node-runtime/managed-runtime-release' import { hasExplicitTuiLaunchCommand } from '../../shared/tui-agent-launch-command-override' -import { readAgentJournalTurn } from '../../shared/agent-session-turn-record' import type { JournalLoad } from '../native-chat/agent-session-journal/journal-open' import { - isProviderTimelineTurnInNamespace, providerTimelineKeyPart, spelledProviderTimelineItemKey } from '../native-chat/agent-session-timeline/provider-timeline-identity' @@ -41,7 +39,7 @@ import { withNativeChatVisualsEnv } from '../native-chat/native-chat-visuals-delivery' import type { AcpLaunchSpec } from './acp-launch-specs' -import { acpSessionNotRestoredItem } from './acp-session-reopen-failure' +import { acpSessionHasNoExchange, acpSessionNotRestoredItem } from './acp-session-reopen-failure' import { loadAcpVisualsSkill, type AcpLaunchVisualsDeps } from './acp-structured-launch-visuals' export type AcpStructuredLaunch = { @@ -270,7 +268,7 @@ export function acpStructuredLaunchPart( key: agentSessionProviderHandleKey(head.handle), mayBeUnsaved: () => head.origin === 'created' && - nothingExchangedOn(deps.readJournal, identity.sessionId, head.handle.nativeId), + acpSessionHasNoExchange(deps.readJournal, identity.sessionId, head.handle.nativeId), unannouncedLosses: () => unannouncedLosses(deps.readJournal, identity.sessionId, record.providerHandleChain) } @@ -288,34 +286,6 @@ export function acpStructuredLaunchPart( } } -/** Proof the chat holds no turn of provider session `nativeId`; a journal that does not read whole, - * or does not read at all, proves nothing. */ -function nothingExchangedOn( - readJournal: AcpStructuredLaunchPartDeps['readJournal'], - sessionId: string, - nativeId: string -): boolean { - let load: JournalLoad | null - try { - load = readJournal(sessionId) - } catch { - return false - } - if (!load) { - return true - } - if (load.damage || load.newer) { - return false - } - for (const item of load.state.items.values()) { - const turn = readAgentJournalTurn(item.body) - if (turn && isProviderTimelineTurnInNamespace(turn.turnId, nativeId)) { - return false - } - } - return true -} - /** Derived on every start, never stored: a row an attach failure dropped is written by the next * start that succeeds. A journal that does not read whole proves nothing, so none is written. */ function unannouncedLosses( diff --git a/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts b/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts index 91522f76efef..0c8e5e7e3d71 100644 --- a/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts +++ b/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts @@ -30,6 +30,8 @@ export type ProviderHistoryItem = { /** Fingerprint of the submitted payload, when the caller can compute one from * provider content. Used only to break an otherwise unique tie. */ payloadFingerprint: string | null + /** Content-only evidence cannot prove a send that happened after this time. */ + createdAt?: number /** Absent when the provider keeps no identity of its own for the item: a submission it matches * keeps Orca's (`orca:`), which is the identity its row already has. */ identity?: AgentJournalItemIdentity @@ -117,8 +119,15 @@ export function reconcileSubmissions(input: { submissionsByFingerprint.set(submission.payloadFingerprint, sameFingerprint) } for (const [fingerprint, fingerprintSubmissions] of submissionsByFingerprint) { + const firstPossibleSend = fingerprintSubmissions.reduce( + (earliest, submission) => Math.min(earliest, submission.submittedAt), + Number.POSITIVE_INFINITY + ) const candidates = input.history.items.filter( - (item) => !claimed.has(item.providerItemId) && item.payloadFingerprint === fingerprint + (item) => + !claimed.has(item.providerItemId) && + item.payloadFingerprint === fingerprint && + (item.createdAt === undefined || item.createdAt >= firstPossibleSend) ) if (fingerprintSubmissions.length === 1 && candidates.length === 1) { const [submission] = fingerprintSubmissions diff --git a/src/main/omp/omp-acp-stored-messages.test.ts b/src/main/omp/omp-acp-stored-messages.test.ts new file mode 100644 index 000000000000..75f848300dcf --- /dev/null +++ b/src/main/omp/omp-acp-stored-messages.test.ts @@ -0,0 +1,394 @@ +import { mkdir, mkdtemp, realpath, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import type { AgentJournalSubmission } from '../../shared/agent-session-journal-types' +import { agentSessionSendBodyFingerprint } from '../../shared/structured-agent-session-send-mutation' +import { acpLaunchSpecFor } from '../acp/acp-launch-specs' +import { readAcpRecoveryHistory } from '../acp/acp-recovery-history' +import type { AcpStructuredLaunch } from '../acp/acp-structured-launch-resolution' +import { createJournalReducerState } from '../native-chat/agent-session-journal/journal-reducer' +import { reconcileSubmissions } from '../native-chat/agent-session-journal/journal-submission-reconciler' +import { ompStoredUserMessagesReader } from './omp-acp-stored-messages' + +const PROVIDER = '0198c2a4-7d1e-7c3a-9a4b-1f2e3d4c5b6a' +const FILE = `2026-10-12T00-00-00-000Z_${PROVIDER}.jsonl` +const SESSION = 'session-alpha-1' +const identity = { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'omp', + providerHandle: null +} +let root: string +let cwd: string +let home: string +let agentRoot: string +let env: Record + +beforeEach(async () => { + root = await realpath(await mkdtemp(join(tmpdir(), 'orca-omp-recovery-'))) + home = join(root, 'home') + cwd = join(home, 'project') + agentRoot = join(root, 'account') + await mkdir(cwd, { recursive: true }) + env = { + HOME: home, + USERPROFILE: home, + PI_CODING_AGENT_DIR: agentRoot, + TMPDIR: join(root, 'temp'), + TEMP: join(root, 'temp') + } +}) + +afterEach(async () => { + await rm(root, { recursive: true, force: true }) +}) + +function turn(id: string, parentId: string | null, text: string, timestamp = 205) { + return { + type: 'message', + id, + parentId, + timestamp: new Date(timestamp).toISOString(), + message: { role: 'user', content: [{ type: 'text', text }], timestamp } + } +} + +async function save( + records: unknown[], + options: { bucket?: string; dir?: string; file?: string; suffix?: string } = {} +) { + const dir = join(options.dir ?? join(agentRoot, 'sessions'), options.bucket ?? '-project') + await mkdir(dir, { recursive: true }) + const file = join(dir, options.file ?? FILE) + const header = { type: 'session', version: 3, id: PROVIDER, cwd } + await writeFile( + file, + [header, ...records].map((record) => JSON.stringify(record)).join('\n') + + (options.suffix ?? '\n') + ) + return file +} + +function submission( + id: string, + text: string, + submittedAt: number, + accepted = false +): AgentJournalSubmission { + return { + clientMessageId: id, + fence: 1, + payloadFingerprint: agentSessionSendBodyFingerprint(SESSION, { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text }] + }), + dispatchState: accepted ? 'accepted' : 'unknown', + providerItemId: null, + reason: null, + submittedAt, + resolvedAt: accepted ? submittedAt + 1 : null + } +} + +async function recover(text = 'run tests', priorSubmissions: AgentJournalSubmission[] = []) { + const state = createJournalReducerState(SESSION, 'epoch-1') + for (const entry of [...priorSubmissions, submission('held', text, 200)]) { + state.submissions.set(entry.clientMessageId, entry) + } + const launch: AcpStructuredLaunch = { + spec: acpLaunchSpecFor('omp')!, + command: 'omp', + args: ['acp'], + cwd, + env, + envToDelete: [], + fullAccess: false, + resume: { + sessionId: PROVIDER, + key: `acp:omp:${PROVIDER}`, + mayBeUnsaved: () => false, + unannouncedLosses: () => [] + } + } + const window = await readAcpRecoveryHistory( + { + spec: { readStoredUserMessages: ompStoredUserMessagesReader() }, + readLaunch: async () => launch, + readJournal: () => ({ state, damage: null, newer: null }) + }, + identity + ) + return { + window, + outcomes: window + ? reconcileSubmissions({ submissions: [...state.submissions.values()], history: window }) + : null + } +} + +describe('OMP restart delivery evidence', () => { + it('confirms a saved user message, even across compaction, using its provider entry ID', async () => { + await save([ + turn('delivered', null, 'run tests'), + { + type: 'message', + id: 'kept', + parentId: 'delivered', + timestamp: new Date(208).toISOString(), + message: { role: 'assistant', content: 'OK' } + }, + { + type: 'compaction', + id: 'compact', + parentId: 'kept', + firstKeptEntryId: 'kept', + summary: 'run tests', + tokensBefore: 100, + timestamp: new Date(210).toISOString() + } + ]) + const { window, outcomes } = await recover() + expect(window).toMatchObject({ boundaryConsistent: false, turnInFlight: true }) + expect(outcomes).toMatchObject([ + { clientMessageId: 'held', outcome: 'accepted', providerItemId: 'delivered' } + ]) + }) + + it.each(['unrelated request', 'run tests'])( + 'never reuses pre-send text when an earlier %s is still uncertain', + async (priorText) => { + await save([turn('before-held-send', null, 'run tests', 150)]) + const { outcomes } = await recover('run tests', [submission('older', priorText, 100)]) + expect(outcomes?.find((outcome) => outcome.clientMessageId === 'held')).toMatchObject({ + outcome: 'unknown' + }) + } + ) + + it('does not confirm an abandoned branch or text appearing only in a summary', async () => { + await save([ + turn('root', null, 'hello', 105), + turn('abandoned', 'root', 'run tests'), + { + type: 'branch_summary', + id: 'branch', + parentId: 'root', + fromId: 'abandoned', + summary: 'run tests', + timestamp: new Date(220).toISOString() + } + ]) + expect((await recover()).outcomes).toMatchObject([{ outcome: 'unknown' }]) + }) + + it('claims an accepted duplicate before offering the second stored copy', async () => { + const accepted = [submission('first', 'run tests', 100, true)] + await save([turn('first-copy', null, 'run tests', 205)]) + expect((await recover('run tests', accepted)).outcomes).toMatchObject([ + { clientMessageId: 'held', outcome: 'unknown' } + ]) + await save([ + turn('first-copy', null, 'run tests', 205), + turn('second-copy', 'first-copy', 'run tests', 210) + ]) + expect((await recover('run tests', accepted)).outcomes).toMatchObject([ + { clientMessageId: 'held', outcome: 'accepted', providerItemId: 'second-copy' } + ]) + }) + + it('claims duplicate copies by their send times rather than active-branch traversal order', async () => { + await save([ + turn('later-copy', null, 'run tests', 210), + turn('earlier-copy', 'later-copy', 'run tests', 205) + ]) + const { outcomes } = await recover('run tests', [submission('first', 'run tests', 100, true)]) + expect(outcomes).toMatchObject([ + { clientMessageId: 'held', outcome: 'accepted', providerItemId: 'later-copy' } + ]) + }) + + it('respects the captured active leaf of late usage rather than reviving an abandoned turn', async () => { + await save([ + turn('root', null, 'hello', 105), + turn('abandoned', 'root', 'run tests'), + { type: 'custom', id: 'branch', parentId: 'root', customType: 'branch-marker' }, + { type: 'model_usage', id: 'usage', parentId: 'abandoned', activeLeafId: 'branch' } + ]) + expect((await recover()).outcomes).toMatchObject([{ outcome: 'unknown' }]) + }) + + it('does not guess an active branch when a late usage record names a missing leaf', async () => { + await save([ + turn('delivered', null, 'run tests'), + { type: 'model_usage', id: 'usage', parentId: 'delivered', activeLeafId: 'missing' } + ]) + expect((await recover()).window).toBeNull() + }) + + it('honors an explicitly empty active branch instead of offering a saved abandoned turn', async () => { + await save([ + turn('abandoned', null, 'run tests'), + { type: 'model_usage', id: 'usage', parentId: 'abandoned', activeLeafId: null } + ]) + expect((await recover()).outcomes).toMatchObject([{ outcome: 'unknown' }]) + }) + + it('preserves whitespace and uses the earlier message time, not a delayed append time', async () => { + const padded = turn('padded', null, ' run tests ', 205) + await save([padded]) + expect((await recover()).outcomes).toMatchObject([{ outcome: 'unknown' }]) + expect((await recover(' run tests ')).outcomes).toMatchObject([{ outcome: 'accepted' }]) + const old = turn('old', null, 'run tests', 100) + old.timestamp = new Date(205).toISOString() + await save([old]) + expect((await recover()).outcomes).toMatchObject([{ outcome: 'unknown' }]) + }) + + it.each(['partial', 'malformed', 'duplicate-id', 'missing-parent', 'wrong-session'])( + 'leaves incomplete or ambiguous %s evidence in doubt', + async (kind) => { + const records: unknown[] = [turn('delivered', null, 'run tests')] + if (kind === 'duplicate-id') { + records.push(turn('delivered', null, 'run tests')) + } + if (kind === 'missing-parent') { + records.push(turn('broken', 'missing', 'run tests')) + } + if (kind === 'wrong-session') { + records.unshift({ type: 'session', version: 3, id: 'other', cwd }) + } + await save(records, { + suffix: kind === 'partial' ? '' : kind === 'malformed' ? '\n{broken\n' : '\n' + }) + expect((await recover()).window).toBeNull() + } + ) + + it('does not use unreadable files, another cwd, or an ambiguous second matching file', async () => { + expect((await recover()).window).toBeNull() + await save([turn('foreign', null, 'run tests')], { bucket: '-other-project' }) + expect((await recover()).window).toBeNull() + const path = await save([turn('delivered', null, 'run tests')]) + await save([turn('duplicate', null, 'run tests')], { file: `other_${PROVIDER}.jsonl` }) + expect((await recover()).window).toBeNull() + await rm(path) + await rm(join(agentRoot, 'sessions', '-project', `other_${PROVIDER}.jsonl`)) + await mkdir(path) + expect((await recover()).window).toBeNull() + }) + + it('uses the pinned account root and profile rather than another account with the same session ID', async () => { + await save([turn('wrong-account', null, 'run tests')], { + dir: join(root, 'other-account', 'sessions') + }) + expect((await recover()).window).toBeNull() + await save([turn('account', null, 'run tests')]) + expect((await recover()).outcomes).toMatchObject([ + { providerItemId: 'account', outcome: 'accepted' } + ]) + env.OMP_PROFILE = 'review' + expect((await recover()).window).toBeNull() + await save([turn('profile', null, 'run tests')], { + dir: join(home, '.omp', 'profiles', 'review', 'agent', 'sessions') + }) + expect((await recover()).outcomes).toMatchObject([ + { providerItemId: 'profile', outcome: 'accepted' } + ]) + }) + + it('honors a custom sessions root and rejects a colliding cwd bucket with a different header', async () => { + await save([turn('ignored-account', null, 'run tests')]) + const custom = join(root, 'custom', 'sessions') + env.OMP_CODING_AGENT_DIR = custom + expect((await recover()).window).toBeNull() + const path = await save([turn('custom', null, 'run tests')], { dir: custom }) + await mkdir(join(custom, '-project', FILE.slice(0, -'.jsonl'.length))) + expect((await recover()).outcomes).toMatchObject([ + { outcome: 'accepted', providerItemId: 'custom' } + ]) + await writeFile( + path, + `${[ + { type: 'session', version: 3, id: PROVIDER, cwd: join(home, 'other') }, + turn('foreign-cwd', null, 'run tests') + ] + .map((record) => JSON.stringify(record)) + .join('\n')}\n` + ) + expect((await recover()).window).toBeNull() + }) + + it('refuses truncated text and mixed image evidence instead of confirming only the text', async () => { + const mixed = { + ...turn('mixed', null, 'run tests'), + message: { + role: 'user', + content: [ + { type: 'text', text: 'run tests' }, + { type: 'image', data: { blob: 'ref' } } + ] + } + } + await save([mixed]) + expect((await recover()).outcomes).toMatchObject([{ outcome: 'unknown' }]) + const text = 'run tests\n\n[Session persistence truncated large content]' + await save([turn('truncated', null, text)]) + expect((await recover(text)).outcomes).toMatchObject([{ outcome: 'unknown' }]) + }) + + it('does not use a file beyond the bounded recovery size', async () => { + await save([ + turn('delivered', null, 'run tests'), + { type: 'custom', id: 'large', parentId: 'delivered', padding: 'x'.repeat(16 * 1024 * 1024) } + ]) + expect((await recover()).window).toBeNull() + }) + + it('does not use a record stream beyond the bounded recovery length', async () => { + const records: unknown[] = [turn('delivered', null, 'run tests')] + for (let index = 1; index < 100_000; index++) { + records.push({ type: 'custom', id: `entry-${index}`, parentId: 'delivered' }) + } + await save(records) + expect((await recover()).window).toBeNull() + }) + + it('does not repair malformed UTF-8 into evidence for a replacement-character message', async () => { + const path = await save([]) + const content = [ + { type: 'session', version: 3, id: PROVIDER, cwd }, + turn('invalid', null, 'run PLACEHOLDERtests') + ] + .map((record) => JSON.stringify(record)) + .join('\n') + await writeFile(path, Buffer.from(`${content.replace('PLACEHOLDER', '\xff')}\n`, 'latin1')) + expect((await recover('run \ufffdtests')).window).toBeNull() + }) + + it('reads a title-slot-prefixed session and honors cancellation without claiming delivery', async () => { + const path = await save([turn('delivered', null, 'run tests')]) + await writeFile( + path, + `${[ + { type: 'title', v: 1, title: 'Saved', pad: '' }, + { type: 'session', version: 3, id: PROVIDER, cwd }, + turn('delivered', null, 'run tests') + ] + .map((record) => JSON.stringify(record)) + .join('\n')}\n` + ) + expect((await recover()).outcomes).toMatchObject([{ outcome: 'accepted' }]) + expect( + await ompStoredUserMessagesReader()({ + env, + cwd, + providerSessionId: PROVIDER, + signal: AbortSignal.abort() + }) + ).toBeNull() + }) +}) diff --git a/src/main/omp/omp-acp-stored-messages.ts b/src/main/omp/omp-acp-stored-messages.ts new file mode 100644 index 000000000000..136841216d2b --- /dev/null +++ b/src/main/omp/omp-acp-stored-messages.ts @@ -0,0 +1,185 @@ +import { homedir } from 'node:os' +import { realpath } from 'node:fs/promises' +import { isAbsolute, resolve } from 'node:path' +import type { AcpStoredUserMessage, AcpStoredUserMessagesReader } from '../acp/acp-recovery-history' +import { resolveOmpSessionsDir } from '../ai-vault/omp-session-root' +import { asRecord, parseJsonObject } from '../ai-vault/session-scanner-values' +import { + openLocalRegularFile, + readLocalFileBounded +} from '../ipc/filesystem/local-regular-file-read' +import { findOmpSessionByCwd } from './omp-session-cwd-lookup' + +const MAX_RECOVERY_BYTES = 16 * 1024 * 1024 +const MAX_RECOVERY_RECORDS = 100_000 +const TRUNCATION_NOTICE = '\n\n[Session persistence truncated large content]' + +type Entry = { + parentId: string | null + message: AcpStoredUserMessage | null +} + +/** Presence evidence, not model context: compaction never erases a saved ancestor's delivery. */ +function storedMessages( + content: string, + sessionId: string +): { cwd: string; messages: AcpStoredUserMessage[] } | null { + if (!content.endsWith('\n')) { + return null + } + const entries = new Map() + let headerCwd: string | null = null + let leaf: string | null = null + let lineCount = 0 + let start = 0 + while (start < content.length) { + const end = content.indexOf('\n', start) + const line = content.slice(start, end) + start = end + 1 + if (!line.trim()) { + continue + } + if (++lineCount > MAX_RECOVERY_RECORDS) { + return null + } + const record = parseJsonObject(line) + if (!record) { + return null + } + if (headerCwd === null) { + if (lineCount === 1 && record.type === 'title') { + continue + } + if ( + record.type !== 'session' || + record.id !== sessionId || + (record.version !== 2 && record.version !== 3) || + typeof record.cwd !== 'string' || + !isAbsolute(record.cwd) + ) { + return null + } + headerCwd = record.cwd + continue + } + const { id, parentId } = record + if ( + typeof id !== 'string' || + !id || + entries.has(id) || + !(parentId === null || (typeof parentId === 'string' && entries.has(parentId))) + ) { + return null + } + let message: AcpStoredUserMessage | null = null + const body = record.type === 'message' ? asRecord(record.message) : null + if (body?.role === 'user') { + const blocks: AcpStoredUserMessage['blocks'] = [] + if (typeof body.content === 'string' && !body.content.endsWith(TRUNCATION_NOTICE)) { + blocks.push({ type: 'text', text: body.content }) + } else if (Array.isArray(body.content)) { + for (const value of body.content) { + const block = asRecord(value) + // Never turn an image/blob/truncated payload into evidence for a text-only send. + if ( + block?.type !== 'text' || + typeof block.text !== 'string' || + block.text.endsWith(TRUNCATION_NOTICE) + ) { + blocks.length = 0 + break + } + blocks.push({ type: 'text', text: block.text }) + } + } + const envelopeTime = + typeof record.timestamp === 'string' ? Date.parse(record.timestamp) : Number.NaN + const createdAt = + typeof body.timestamp === 'number' ? Math.min(body.timestamp, envelopeTime) : envelopeTime + if (blocks.length && Number.isFinite(createdAt)) { + message = { id, blocks, createdAt } + } + } + entries.set(id, { parentId, message }) + // OMP v18.11.2 sessionEntryLeafId: late model_usage restores its captured branch. + if (record.type === 'model_usage' && record.activeLeafId !== undefined) { + if (!(record.activeLeafId === null || typeof record.activeLeafId === 'string')) { + return null + } + leaf = record.activeLeafId + } else { + leaf = id + } + } + if (headerCwd === null) { + return null + } + if (leaf !== null && !entries.has(leaf)) { + return null + } + const messages: AcpStoredUserMessage[] = [] + while (leaf !== null) { + const entry = entries.get(leaf) + if (!entry) { + return null + } + if (entry.message) { + messages.push(entry.message) + } + leaf = entry.parentId + } + return { cwd: headerCwd, messages } +} + +/** Runs only on the execution host, against the sealed launch environment and known cwd. */ +export function ompStoredUserMessagesReader(): AcpStoredUserMessagesReader { + return async ({ env, cwd, providerSessionId, signal }) => { + try { + signal.throwIfAborted() + const homeDir = (process.platform === 'win32' ? env.USERPROFILE : env.HOME) || homedir() + const sessionsDir = resolveOmpSessionsDir({ env, homeDir }) + if (!sessionsDir || !isAbsolute(sessionsDir) || !isAbsolute(cwd)) { + return null + } + const file = await findOmpSessionByCwd({ + sessionsDir, + sessionId: providerSessionId, + cwd, + homeDir, + env, + requireUnique: true, + signal + }) + if (!file) { + return null + } + signal.throwIfAborted() + const { handle, stats } = await openLocalRegularFile(file) + try { + if (stats.size > MAX_RECOVERY_BYTES) { + return null + } + const content = await readLocalFileBounded(handle, MAX_RECOVERY_BYTES, stats.size) + signal.throwIfAborted() + const text = new TextDecoder('utf-8', { fatal: true }).decode(content) + const parsed = storedMessages(text, providerSessionId) + if (!parsed) { + return null + } + const headerCwd = parsed.cwd + const [savedCwd, launchCwd] = await Promise.all([ + realpath(headerCwd).catch(() => resolve(headerCwd)), + realpath(cwd).catch(() => resolve(cwd)) + ]) + if (savedCwd !== launchCwd) { + return null + } + return parsed.messages + } finally { + await handle.close() + } + } catch { + return null + } + } +} diff --git a/src/main/omp/omp-session-cwd-lookup.test.ts b/src/main/omp/omp-session-cwd-lookup.test.ts index 691275f09fcb..5914f4fd3832 100644 --- a/src/main/omp/omp-session-cwd-lookup.test.ts +++ b/src/main/omp/omp-session-cwd-lookup.test.ts @@ -13,6 +13,7 @@ vi.mock('node:fs/promises', async (importOriginal) => { }) import { OMP_TRANSCRIPT_IMPORT } from './omp-transcript-import' +import { findOmpSessionByCwd } from './omp-session-cwd-lookup' const ID = '0198c2a4-7d1e-7c3a-9a4b-1f2e3d4c5b6a' const FILE = `2026-10-12T00-00-00-000Z_${ID}.jsonl` @@ -159,6 +160,57 @@ describe('OMP known-workspace transcript lookup', () => { expect(io.open).toHaveBeenCalledTimes(1) }) + it('cannot establish a unique recovery match when a known bucket is unreadable', async () => { + await record('--repo-orca--') + io.open.mockRejectedValue(new Error('EACCES')) + expect( + await findOmpSessionByCwd({ + sessionsDir, + sessionId: ID, + cwd: '/repo/orca', + homeDir: join(root, 'home'), + env: {}, + requireUnique: true + }) + ).toBeNull() + expect(io.open).toHaveBeenCalled() + }) + + it('leaves recovery uncertain when a cwd bucket exceeds the enumeration bound', async () => { + const cwd = join(root, 'home', 'project') + await record('-project') + let yielded = 0 + let closed = false + io.open.mockImplementation(async () => ({ + async *[Symbol.asyncIterator]() { + try { + for (let index = 0; index < 100_002; index++) { + yielded++ + yield { + name: index === 0 ? FILE : `unrelated-${index}.jsonl`, + isFile: () => true, + isSymbolicLink: () => false + } + } + } finally { + closed = true + } + } + })) + expect( + await findOmpSessionByCwd({ + sessionsDir, + sessionId: ID, + cwd, + homeDir: join(root, 'home'), + env: { TMPDIR: join(root, 'temp'), TEMP: join(root, 'temp') }, + requireUnique: true + }) + ).toBeNull() + expect(yielded).toBe(100_001) + expect(closed).toBe(true) + }) + it.skipIf(process.platform === 'win32')('rejects symlinked groups and files', async () => { const outside = join(root, 'outside') await mkdir(outside, { recursive: true }) diff --git a/src/main/omp/omp-session-cwd-lookup.ts b/src/main/omp/omp-session-cwd-lookup.ts index 18fd11b3c968..549d720a620e 100644 --- a/src/main/omp/omp-session-cwd-lookup.ts +++ b/src/main/omp/omp-session-cwd-lookup.ts @@ -4,6 +4,8 @@ import { basename, extname, isAbsolute, join, relative, resolve } from 'node:pat import { PI_CWD_BUCKET_LAYOUT } from '../ai-vault/session-cwd-bucket-layouts' import { SESSION_FILE_ID_LAYOUTS } from '../native-chat/session-file-id-layouts' +const MAX_RECOVERY_DIRECTORY_ENTRIES = 100_000 + async function canonicalPath(path: string): Promise { return realpath(path).catch(() => resolve(path)) } @@ -47,28 +49,48 @@ async function cwdBuckets(cwd: string, home: string, env: NodeJS.ProcessEnv): Pr ].filter((name): name is string => name !== null) } -/** Import only a known workspace's direct session files, stopping at its first ID match. */ +/** Import a known workspace's direct session files; recovery additionally requires a unique match. */ export async function findOmpSessionByCwd(input: { sessionsDir: string sessionId: string cwd: string homeDir: string env: NodeJS.ProcessEnv + requireUnique?: boolean + signal?: AbortSignal strictErrors?: boolean }): Promise { const sessionId = input.sessionId.trim() if (!sessionId || !isAbsolute(input.cwd)) { return null } + let found: string | null = null + let scanned = 0 for (const bucket of await cwdBuckets(input.cwd, input.homeDir, input.env)) { + input.signal?.throwIfAborted() const dirPath = join(input.sessionsDir, bucket) try { const stat = await lstat(dirPath) if (!stat.isDirectory() || stat.isSymbolicLink()) { + if (input.requireUnique) { + return null + } continue } const directory = await opendir(dirPath) for await (const entry of directory) { + input.signal?.throwIfAborted() + if (input.requireUnique && ++scanned > MAX_RECOVERY_DIRECTORY_ENTRIES) { + return null + } + if ( + input.requireUnique && + extname(entry.name).toLowerCase() === '.jsonl' && + SESSION_FILE_ID_LAYOUTS.omp.fileMatchesId(entry.name, sessionId) && + (!entry.isFile() || entry.isSymbolicLink()) + ) { + return null + } if ( !entry.isFile() || entry.isSymbolicLink() || @@ -78,7 +100,13 @@ export async function findOmpSessionByCwd(input: { } const file = join(dirPath, entry.name) if (SESSION_FILE_ID_LAYOUTS.omp.fileMatchesId(file, sessionId)) { - return file + if (!input.requireUnique) { + return file + } + if (found && found !== file) { + return null + } + found = file } } } catch (error) { @@ -92,8 +120,14 @@ export async function findOmpSessionByCwd(input: { ) { throw error } - // A stale or unreadable bucket must not replace a missing history with a blank chat. + // An unreadable bucket cannot establish uniqueness for restart recovery. + if ( + input.requireUnique && + !(error instanceof Error && 'code' in error && error.code === 'ENOENT') + ) { + return null + } } } - return null + return found } diff --git a/src/main/opencode/opencode-acp-stored-messages.test.ts b/src/main/opencode/opencode-acp-stored-messages.test.ts index 43ca17f8a5d6..1bacbe586262 100644 --- a/src/main/opencode/opencode-acp-stored-messages.test.ts +++ b/src/main/opencode/opencode-acp-stored-messages.test.ts @@ -33,6 +33,7 @@ const readInProcess: OpenCodeTranscriptPageReader = async (args) => readOpenCode function readFrom(dataHome: string) { return openCodeStoredUserMessagesReader(readInProcess)({ env: { HOME: dataHome, USERPROFILE: dataHome, XDG_DATA_HOME: dataHome }, + cwd: dataHome, providerSessionId: 'ses_1', signal: new AbortController().signal }) @@ -71,6 +72,7 @@ describe('OpenCode stored user messages', () => { await expect( read({ env: { HOME: home, USERPROFILE: home }, + cwd: home, providerSessionId: 'ses_1', signal: new AbortController().signal })