fix(omp): reconcile uncertain sends from owned native session history

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.
+
 <Callout title="Experimental">
   Transcript fidelity, streaming, and terminal parity are still under active tuning. Prefer the raw
   TUI when you need every OSC/status detail.

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: {

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(

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<string, string>
+  /** The workspace on the execution host that owns this launch and its store. */
+  cwd: string
   providerSessionId: string
   signal: AbortSignal
 }) => Promise<AcpStoredUserMessage[] | null>
@@ -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
       }),

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
+}

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<string, string> = { 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(() => ({

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(

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:<clientMessageId>`), 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