diff --git a/.github/workflows/pi-provider-runtime.yml b/.github/workflows/pi-provider-runtime.yml
index 4cde4ab5f69..ac652d64e63 100644
--- a/.github/workflows/pi-provider-runtime.yml
+++ b/.github/workflows/pi-provider-runtime.yml
@@ -5,6 +5,7 @@ on:
     paths:
       - 'src/shared/commit-message-agent-specs-primary.ts'
       - 'tests/tools/pi-provider-runtime-smoke.mjs'
+      - 'src/main/pi/rpc-history-*.ts'
       - '.github/workflows/pi-provider-runtime.yml'
 permissions:
   contents: read
@@ -29,5 +30,5 @@ jobs:
       - uses: ./.github/actions/install-node-dependencies
       - name: Install pinned Pi runtime
         run: npm install --prefix .cache/pi-provider --ignore-scripts --no-audit --no-fund @earendil-works/pi-coding-agent@0.84.2
-      - name: Verify extension model generation before and after
+      - name: Verify extension model generation and session entry ids
         run: node tests/tools/pi-provider-runtime-smoke.mjs .cache/pi-provider/node_modules/@earendil-works/pi-coding-agent/dist/cli.js
diff --git a/src/main/daemon/ndjson.test.ts b/src/main/daemon/ndjson.test.ts
index 9c7c481eac3..e075539e139 100644
--- a/src/main/daemon/ndjson.test.ts
+++ b/src/main/daemon/ndjson.test.ts
@@ -266,6 +266,45 @@ describe('createNdjsonParser', () => {
     expect(records).toEqual([{ index: 0 }, { index: 1 }, { index: 2 }, { index: 3 }])
   })
 
+  it('discards only the paused oversized tail after delivering older records across re-pauses', () => {
+    const records: unknown[] = []
+    const rejected: unknown[] = []
+    let paused = false
+    const framer = createIncrementalNdjsonFramer(
+      (record) => {
+        records.push(record)
+        if (records.length <= 2) {
+          paused = true
+        }
+      },
+      (error) => rejected.push(error),
+      { maxLineBytes: 32, shouldPause: () => paused }
+    )
+    framer.feed('{"a":1}\n{"b":2}\n{"c":3}\n{"oversized":"')
+    framer.feed('x'.repeat(64 * 1024))
+    expect(rejected).toEqual([
+      expect.objectContaining({
+        prefix: expect.stringMatching(/^\{"oversized":/),
+        maxLineBytes: 64 * 1024
+      })
+    ])
+    expect(framer.takeTail()).toBe('')
+    paused = false
+    framer.resume()
+    expect(records).toEqual([{ a: 1 }, { b: 2 }])
+    paused = false
+    framer.feed('still discarded')
+    expect(records).toEqual([{ a: 1 }, { b: 2 }, { c: 3 }])
+    paused = true
+    framer.feed('"}\n{"d":4}\n{"e":')
+    expect(records).toHaveLength(3)
+    paused = false
+    framer.resume()
+    framer.feed('5}\n')
+    expect(records).toEqual([{ a: 1 }, { b: 2 }, { c: 3 }, { d: 4 }, { e: 5 }])
+    expect(rejected).toHaveLength(1)
+  })
+
   it('caps an actually incomplete paused suffix and recovers at the next delimiter', () => {
     const records: unknown[] = []
     const rejected: unknown[] = []
@@ -278,6 +317,8 @@ describe('createNdjsonParser', () => {
 
     framer.feed('{}\nx')
     framer.feed('y'.repeat(70_000))
+    expect(framer.takeTail()).toBe('')
+    framer.feed('more discarded bytes')
     paused = false
     framer.resume()
     framer.feed('\n{"recovered":true}\n')
diff --git a/src/main/jsonl-rpc/agent-connection.ts b/src/main/jsonl-rpc/agent-connection.ts
index d9c5667d961..bf44c9034c1 100644
--- a/src/main/jsonl-rpc/agent-connection.ts
+++ b/src/main/jsonl-rpc/agent-connection.ts
@@ -14,6 +14,7 @@ import {
   type JsonlRpcPeerHandlers,
   type JsonlRpcRecord
 } from './peer'
+import type { JsonlRpcRequestOptions } from './peer-oversized-reply'
 import { resolveJsonlRpcPeerOptions, type JsonlRpcPeerOptions } from './peer-limits'
 
 export type JsonlRpcAgentConnectionOptions = JsonlRpcPeerHandlers & {
@@ -151,7 +152,7 @@ export class JsonlRpcAgentConnection {
   request(
     command: string,
     params: Record<string, unknown> = {},
-    options: { timeoutMs?: number | null } = {}
+    options: JsonlRpcRequestOptions = {}
   ): Promise<unknown> {
     return this.peer.request(command, params, options)
   }
diff --git a/src/main/jsonl-rpc/peer-oversized-reply.ts b/src/main/jsonl-rpc/peer-oversized-reply.ts
new file mode 100644
index 00000000000..fa8bb629211
--- /dev/null
+++ b/src/main/jsonl-rpc/peer-oversized-reply.ts
@@ -0,0 +1,38 @@
+import type { NdjsonRejectedRecord } from '../../shared/main-process-ndjson-framer'
+
+/** A reply over the line cap, refused for the one request that asked to survive it. */
+export class JsonlRpcReplyTooLargeError extends Error {
+  constructor(
+    readonly command: string,
+    readonly observedBytes: number,
+    readonly maxBytes: number
+  ) {
+    super(`Agent reply to ${command} exceeded ${maxBytes} bytes`)
+    this.name = 'JsonlRpcReplyTooLargeError'
+  }
+}
+
+export type JsonlRpcRequestOptions = {
+  timeoutMs?: number | null
+  /** An optional read whose oversized reply fails only itself instead of the stream. */
+  oversizedReply?: 'reject'
+}
+
+// Pi echoes opaque ids before the response payload; the category survives its waiter's timeout.
+const OVERSIZED_REPLY_HEAD = /^\{"id":"(orca-optional-([1-9]\d*))","type":"response"(?=[,}])/
+
+/**
+ * Only an issued optional reply can be dropped alone. Complete-queue loss may also omit
+ * lifecycle records, even when its prefix identifies an optional reply.
+ */
+export function oversizedReplyRequest(
+  rejected: Extract<NdjsonRejectedRecord, { kind: 'line-too-long' }>,
+  highestOptionalAssigned: number
+): string | null {
+  const head = OVERSIZED_REPLY_HEAD.exec(rejected.prefix)
+  if (!head || rejected.droppedCompleteRecords) {
+    return null
+  }
+  const number = Number(head[2])
+  return Number.isSafeInteger(number) && number <= highestOptionalAssigned ? head[1] : null
+}
diff --git a/src/main/jsonl-rpc/peer.test.ts b/src/main/jsonl-rpc/peer.test.ts
index dc4b8e816b5..40711ed9795 100644
--- a/src/main/jsonl-rpc/peer.test.ts
+++ b/src/main/jsonl-rpc/peer.test.ts
@@ -159,6 +159,149 @@ describe('JSON-lines RPC peer', () => {
     expect(onRecord).not.toHaveBeenCalled()
   })
 
+  it('fails only an opted-in read whose reply is oversized, then keeps the stream', async () => {
+    const onRecord = vi.fn()
+    const { peer, input, reply } = fixture({ onRecord }, { maxLineBytes: 128 })
+    const big = { id: 'orca-optional-1', type: 'response', command: 'get_entries', success: true }
+    const entries = peer.request('get_entries', {}, { oversizedReply: 'reject' })
+    input.write(`${JSON.stringify({ ...big, data: { entries: ['x'.repeat(300)] } })}\n`)
+    await expect(entries).rejects.toThrow('exceeded 128 bytes')
+    reply({ type: 'agent_settled' })
+    const state = peer.request('get_state')
+    reply({ type: 'response', id: 'orca-1', command: 'get_state', success: true, data: {} })
+    expect(await state).toEqual({})
+    expect(onRecord).toHaveBeenCalledExactlyOnceWith({ type: 'agent_settled' })
+    // A read that did not opt in still ends the stream.
+    const plain = expect(peer.request('get_entries')).rejects.toThrow('size limit')
+    input.write(`${JSON.stringify({ ...big, id: 'orca-2', data: 'y'.repeat(300) })}\n`)
+    await plain
+    expect(peer.closed).toBe(true)
+  })
+
+  it('survives a late oversized optional reply after timeout and permits another read', async () => {
+    vi.useFakeTimers()
+    const onRecord = vi.fn()
+    const { peer, input, output, reply } = fixture({ onRecord }, { maxLineBytes: 128 })
+    const expired = expect(
+      peer.request('get_entries', {}, { oversizedReply: 'reject', timeoutMs: 10 })
+    ).rejects.toThrow('in time')
+    expect(JSON.parse(output.read().toString()).id).toBe('orca-optional-1')
+    await vi.advanceTimersByTimeAsync(10)
+    await expired
+    const state = peer.request('get_state')
+    input.write(
+      '{"id":"orca-optional-1","type":"response","command":"get_entries","success":true,"data":"'
+    )
+    input.write('x'.repeat(300))
+    input.write('"}\n')
+    reply({ id: 'orca-1', type: 'response', command: 'get_state', success: true, data: {} })
+    reply({ type: 'agent_settled' })
+    await expect(state).resolves.toEqual({})
+    const next = peer.request('get_entries', {}, { oversizedReply: 'reject' })
+    reply({
+      id: 'orca-optional-2',
+      type: 'response',
+      command: 'get_entries',
+      success: true,
+      data: []
+    })
+    await expect(next).resolves.toEqual([])
+    expect(onRecord).toHaveBeenCalledExactlyOnceWith({ type: 'agent_settled' })
+    expect(peer.closed).toBe(false)
+  })
+
+  it.each([undefined, 'different_command'])(
+    'uses local context for oversized optional replies with command %s',
+    async (command) => {
+      const { peer, input } = fixture({}, { maxLineBytes: 128 })
+      const pending = expect(
+        peer.request('get_entries', {}, { oversizedReply: 'reject' })
+      ).rejects.toThrow('reply to get_entries exceeded 128 bytes')
+      input.write(
+        `${JSON.stringify({ id: 'orca-optional-1', type: 'response', command, data: 'x'.repeat(300) })}\n`
+      )
+      await pending
+      expect(peer.closed).toBe(false)
+    }
+  )
+
+  it.each([
+    ['orca-optional-2', 'response'],
+    ['orca-optional-0', 'response'],
+    ['orca-optional-01', 'response'],
+    ['orca-optional-9007199254740992', 'response'],
+    ['orca-1', 'response'],
+    ['foreign', 'response'],
+    ['orca-optional-1', 'agent_end']
+  ])('closes on an oversized record without issued optional ownership: %s %s', async (id, type) => {
+    const { peer, input } = fixture({}, { maxLineBytes: 128 })
+    const pending = expect(
+      peer.request('get_entries', {}, { oversizedReply: 'reject' })
+    ).rejects.toThrow('size limit')
+    input.write(`${JSON.stringify({ id, type, data: 'x'.repeat(300) })}\n`)
+    await pending
+    expect(peer.closed).toBe(true)
+  })
+
+  it('closes when the response type has no structural field boundary', async () => {
+    const { peer, input } = fixture({}, { maxLineBytes: 128 })
+    const pending = expect(
+      peer.request('get_entries', {}, { oversizedReply: 'reject' })
+    ).rejects.toThrow('size limit')
+    input.write(`{"id":"orca-optional-1","type":"response"garbage${'x'.repeat(300)}`)
+    await pending
+    expect(peer.closed).toBe(true)
+  })
+
+  it('keeps required replies and queued events around an oversized paused optional suffix', async () => {
+    const records: unknown[] = []
+    const { peer, input } = fixture(
+      {
+        onRecord: (record) => {
+          records.push(record)
+          if (record.type === 'first' || record.type === 'second') {
+            peer.pauseReading()
+          }
+        }
+      },
+      { maxLineBytes: 256 }
+    )
+    const optional = expect(
+      peer.request('get_entries', {}, { oversizedReply: 'reject' })
+    ).rejects.toThrow('exceeded 256 bytes')
+    const state = peer.request('get_state')
+    // Already-read bytes can arrive within a burst after the consumer pauses.
+    input.emit(
+      'data',
+      '{"type":"first"}\n{"type":"second"}\n{"id":"orca-1","type":"response","command":"get_state","success":true,"data":{}}\n{"id":"orca-optional-1","type":"response","data":"'
+    )
+    input.emit('data', 'x'.repeat(70_000))
+    await optional
+    peer.resumeReading()
+    expect(records).toEqual([{ type: 'first' }, { type: 'second' }])
+    peer.resumeReading()
+    await expect(state).resolves.toEqual({})
+    input.emit('data', 'remaining optional bytes\n')
+    input.emit('data', '{"type":"agent_settled"}\n')
+    expect(records).toEqual([{ type: 'first' }, { type: 'second' }, { type: 'agent_settled' }])
+    expect(peer.closed).toBe(false)
+  })
+
+  it('closes on complete-queue loss even if the omitted batch begins with an optional reply', async () => {
+    const { peer, input } = fixture({ onRecord: () => peer.pauseReading() }, { maxLineBytes: 256 })
+    const pending = expect(
+      peer.request('get_entries', {}, { oversizedReply: 'reject' })
+    ).rejects.toThrow('size limit')
+    input.emit('data', '{"type":"first"}\n{"type":"queued"}\n')
+    const optional = '{"id":"orca-optional-1","type":"response","data":"'
+    input.emit(
+      'data',
+      `${optional}${'x'.repeat(64 * 1024 * 1024)}"}\n{"type":"tool_execution_end"}\n`
+    )
+    await pending
+    expect(peer.closed).toBe(true)
+  })
+
   it('enforces pending capacity, and releases it after timeout and invalid response', async () => {
     vi.useFakeTimers()
     const { peer, reply } = fixture({}, { maxPendingRequests: 1, requestTimeoutMs: 10 })
diff --git a/src/main/jsonl-rpc/peer.ts b/src/main/jsonl-rpc/peer.ts
index 992d7f987e2..7e934b9a278 100644
--- a/src/main/jsonl-rpc/peer.ts
+++ b/src/main/jsonl-rpc/peer.ts
@@ -11,6 +11,11 @@ import {
   resolveJsonlRpcPeerOptions,
   type JsonlRpcPeerOptions
 } from './peer-limits'
+import {
+  JsonlRpcReplyTooLargeError,
+  oversizedReplyRequest,
+  type JsonlRpcRequestOptions
+} from './peer-oversized-reply'
 
 const recordSchema = z.looseObject({ type: z.string() })
 const responseSchema = z.looseObject({
@@ -59,6 +64,7 @@ export class JsonlRpcPeer {
   private readonly writer: ProviderStdioWriteQueue
   private readonly framer: ReturnType<typeof createIncrementalNdjsonFramer>
   private nextId = 0
+  private nextOptionalId = 0
   private terminalError?: Error
   private inputEndError?: Error
   private readingPaused = false
@@ -83,7 +89,20 @@ export class JsonlRpcPeer {
       (record) => this.dispatch(record),
       (rejected) => {
         if (rejected.kind === 'line-too-long') {
-          this.close(new Error('JSON-lines RPC record exceeded the size limit'))
+          const id = oversizedReplyRequest(rejected, this.nextOptionalId)
+          if (id) {
+            const request = this.pending.get(id)
+            if (request) {
+              const error = new JsonlRpcReplyTooLargeError(
+                request.command,
+                rejected.observedBytes,
+                this.limits.maxLineBytes
+              )
+              this.rejectPending(id, error)
+            }
+          } else {
+            this.close(new Error('JSON-lines RPC record exceeded the size limit'))
+          }
         } else {
           this.diagnose('Ignored non-JSON agent output')
         }
@@ -138,7 +157,7 @@ export class JsonlRpcPeer {
   request(
     command: string,
     params: Record<string, unknown> = {},
-    options: { timeoutMs?: number | null } = {}
+    options: JsonlRpcRequestOptions = {}
   ): Promise<unknown> {
     if (this.terminalError) {
       return Promise.reject(this.terminalError)
@@ -155,7 +174,13 @@ export class JsonlRpcPeer {
     } catch (error) {
       return Promise.reject(error)
     }
-    const id = `orca-${++this.nextId}`
+    if (options.oversizedReply === 'reject' && this.nextOptionalId >= Number.MAX_SAFE_INTEGER) {
+      return Promise.reject(new Error('JSON-lines RPC optional request identity exhausted'))
+    }
+    const id =
+      options.oversizedReply === 'reject'
+        ? `orca-optional-${++this.nextOptionalId}`
+        : `orca-${++this.nextId}`
     const controller = new AbortController()
     return new Promise((resolve, reject) => {
       const timer =
diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts
index 01e1d246628..b49cf2bacce 100644
--- a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts
+++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts
@@ -292,21 +292,19 @@ async function decodeReplacementItems(input: {
 /**
  * A message whose only content is a tool invocation becomes a tool-call item so
  * the reducer renders it as one. Everything else stays a message item with its
- * blocks bounded in place.
+ * adoption limits applied only when supplied.
  */
-function legacyItemBody(
+export function legacyItemBody(
   message: NativeChatMessage,
-  limits: JournalPayloadLimits
+  limits?: JournalPayloadLimits
 ): AgentJournalItemBody {
   const only = message.blocks.length === 1 ? message.blocks[0] : undefined
   if (only?.type === 'tool-call') {
-    // Legacy transcripts are untrusted and can contain arbitrarily large tool
-    // arguments. Keep them on the same bounded path as live events before the
-    // replacement epoch is published.
+    // Adoption bounds bulk before publishing its replacement epoch; catch-up admits whole batches.
     return {
       kind: 'tool-call',
       name: only.name,
-      input: boundToolInput(only.input, limits),
+      input: limits ? boundToolInput(only.input, limits) : only.input,
       ...(only.callId !== undefined ? { callId: only.callId } : {}),
       state: 'completed'
     }
@@ -314,7 +312,7 @@ function legacyItemBody(
   return {
     kind: 'message',
     role: message.role,
-    blocks: message.blocks.map((block) => boundBlock(block, limits))
+    blocks: limits ? message.blocks.map((block) => boundBlock(block, limits)) : message.blocks
   }
 }
 
diff --git a/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts b/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts
index fb0e723f41e..85ac3f6b126 100644
--- a/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts
+++ b/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts
@@ -164,17 +164,9 @@ export function createPiTranscriptCollector<T>(
           }
           processedRows += 1
           processedBytes += Buffer.byteLength(row.raw_line, 'utf8')
-          const record = parseJsonObject(row.raw_line)
-          if (record) {
-            const message =
-              record.type === 'compaction' || record.type === 'branch_summary'
-                ? piSummaryMessage(record, row.fallback_id)
-                : asRecord(record.message)?.role === 'system'
-                  ? null
-                  : decodeOmpTranscriptLine(row.raw_line, row.fallback_id)
-            if (message) {
-              keep(toItem(message, identify(row.raw_line, row.line_index)))
-            }
+          const message = decodePiSessionEntryLine(row.raw_line, row.fallback_id)
+          if (message) {
+            keep(toItem(message, identify(row.raw_line, row.line_index)))
           }
           if (
             processedBytes >= 1024 * 1024 ||
@@ -211,6 +203,24 @@ export function createPiTranscriptCollector<T>(
   }
 }
 
+/** One saved Pi entry as a chat shows it: summaries for compaction and abandoned branches, nothing
+ *  for a system prompt, else the decoder Pi shares with omp. */
+export function decodePiSessionEntryLine(
+  line: string,
+  fallbackId: string
+): NativeChatMessage | null {
+  const record = parseJsonObject(line)
+  if (!record) {
+    return null
+  }
+  if (record.type === 'compaction' || record.type === 'branch_summary') {
+    return piSummaryMessage(record, fallbackId)
+  }
+  return asRecord(record.message)?.role === 'system'
+    ? null
+    : decodeOmpTranscriptLine(line, fallbackId)
+}
+
 function piSummaryMessage(
   record: Record<string, unknown>,
   fallbackId: string
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts
index 166990cfca7..a7d01fe8167 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts
@@ -20,6 +20,7 @@ describe('history catch-up runner', () => {
     let barriers = 0
     const journal = {
       epoch: 'epoch-1',
+      cursor: () => ({ epoch: 'epoch-1', sequence: 0 }),
       item: () => null,
       itemBody: () => null,
       canonicalItemId: (key: string) => key,
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-views.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-views.ts
index 54db9e893e2..81c417061dc 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-views.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-views.ts
@@ -25,6 +25,7 @@ export type StructuredAgentSessionRevisionJournal = Pick<
 export type StructuredAgentSessionHistoryJournal = Pick<
   AgentSessionJournal,
   | 'epoch'
+  | 'cursor'
   | 'visitItems'
   | 'visitItemsWithLinkage'
   | 'item'
diff --git a/src/main/native-chat/transcript-line-decoders-omp.ts b/src/main/native-chat/transcript-line-decoders-omp.ts
index bfe44fd87bf..7f0519feb00 100644
--- a/src/main/native-chat/transcript-line-decoders-omp.ts
+++ b/src/main/native-chat/transcript-line-decoders-omp.ts
@@ -13,6 +13,7 @@ import {
   type NativeChatMessage
 } from '../../shared/native-chat-types'
 import type { AgentSessionTokenUsage } from '../../shared/agent-session-context-usage'
+import { buildImageDataUri } from '../../shared/image-data-uri'
 import {
   asRecord,
   extractString,
@@ -268,12 +269,14 @@ function ompContentBlock(record: Record<string, unknown> | null): NativeChatBloc
       const name = extractString(record.name) ?? 'tool'
       return { type: 'tool-call', name, input: record.arguments }
     }
-    case 'image':
-      // Why: omp stores images as content-addressed blob handles
-      // (`blob:sha256:…`) rather than a path or URL, so there is nothing the
-      // renderer can load. Dropping the block matches how the Claude mapper
-      // treats an image record with neither source.
-      return null
+    case 'image': {
+      const data = extractString(record.data)
+      const url =
+        data && !data.startsWith('blob:')
+          ? buildImageDataUri(extractString(record.mimeType) ?? undefined, data)
+          : null
+      return url ? { type: 'image-ref', url } : null
+    }
     default:
       return null
   }
diff --git a/src/main/pi/rpc-history-capture.ts b/src/main/pi/rpc-history-capture.ts
new file mode 100644
index 00000000000..115c6d36731
--- /dev/null
+++ b/src/main/pi/rpc-history-capture.ts
@@ -0,0 +1,137 @@
+import type { AgentSessionProviderHandle } from '../../shared/agent-session-provider-handle'
+import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
+import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger'
+import { readPiEntries } from './rpc-history-catch-up'
+import { piBranchInTree, piBranchSince } from './rpc-history-import'
+import { encodePiHistoryClaim } from './rpc-history-journal-view'
+import type { PiRpcConnection, PiRpcSession } from './rpc-session'
+
+export type PiHistoryResumePointWrite = {
+  sessionId: string
+  fence: number
+  handle: AgentSessionProviderHandle
+}
+
+/** Where the chat's journal and Pi's entries were last known to meet; null is a session with no
+ *  entries yet. */
+export type PiHistoryAnchor = { entry: string | null }
+
+export type PiHistoryCaptureDeps = {
+  /** Revises the open's handle in place with where the chat's journal meets Pi's entries. */
+  persistResumePoint?: (write: PiHistoryResumePointWrite) => Promise<void>
+}
+
+export type PiHistoryCaptureParts = {
+  sessionId: string
+  fence: number
+  connection: Pick<PiRpcConnection, 'request'>
+  sink: StructuredAgentSessionEventSink | undefined
+  /** Resolves once the timeline holds no events the sink refused. */
+  drained: () => Promise<void>
+  /** Changes whenever the chat's Pi does anything after a turn. */
+  activity: () => number
+  handle: () => AgentSessionProviderHandle | null
+  persist?: PiHistoryCaptureDeps['persistResumePoint']
+  logger: StructuredAgentSessionLogger
+}
+
+/**
+ * After each turn the chat's Pi ran, saves the claim that its entries are shown: Pi's leaf, and the
+ * journal position its rows reached. Bookkeeping only: never awaited by the turn, never failing a
+ * send. A capture the chat moved past, or one that fails, saves nothing, so the next open treats
+ * that turn as not accounted for and skips rather than duplicates.
+ */
+export class PiHistoryCapture {
+  private anchor: PiHistoryAnchor | undefined
+  private queue: Promise<void> = Promise.resolve()
+
+  constructor(private readonly parts: () => PiHistoryCaptureParts) {}
+
+  /** Built with its session, before the parts it reads when a turn ends exist. */
+  static forSession(
+    session: Parameters<typeof piRpcHistoryCaptureParts>[0],
+    deps: () => Parameters<typeof piRpcHistoryCaptureParts>[1]
+  ): PiHistoryCapture {
+    return new PiHistoryCapture(() => piRpcHistoryCaptureParts(session, deps()))
+  }
+
+  /** Only an open that left the claim at Pi's current leaf lets captures continue from it. */
+  start(anchor: PiHistoryAnchor | undefined): void {
+    this.anchor = anchor
+  }
+
+  turnEnded(): void {
+    const deps = this.parts()
+    const anchor = this.anchor
+    if (!anchor || !deps.persist) {
+      return
+    }
+    const activity = deps.activity()
+    this.queue = this.queue
+      .then(() => this.capture(deps, anchor, activity))
+      .catch((error: unknown) =>
+        deps.logger.warn('Pi turn was not recorded for catch-up', {
+          scope: 'pi-history-capture',
+          sessionId: deps.sessionId,
+          error
+        })
+      )
+  }
+
+  private async capture(
+    deps: PiHistoryCaptureParts,
+    anchor: PiHistoryAnchor,
+    activity: number
+  ): Promise<void> {
+    const moved = () => this.anchor !== anchor || deps.activity() !== activity
+    await deps.drained()
+    const sink = deps.sink
+    const written = await sink?.written?.()
+    const history = sink?.journalHistory?.()
+    if (!written?.ok || !history || moved()) {
+      return
+    }
+    // Read before Pi's leaf: a row landing after this is not claimed.
+    const through = history.cursor()
+    const answer = await readPiEntries(deps.connection, anchor.entry)
+    const branch =
+      anchor.entry === null ? piBranchInTree(answer, null) : piBranchSince(answer, anchor.entry)
+    const handle = deps.handle()
+    if (moved() || !branch || answer.leafId === null || !handle) {
+      return
+    }
+    const entry = answer.leafId
+    // The anchor only keeps the next read small; the saved claim is what an open trusts.
+    this.anchor = { entry }
+    await deps.persist?.({
+      sessionId: deps.sessionId,
+      fence: deps.fence,
+      handle: {
+        ...handle,
+        resumeCursor: encodePiHistoryClaim({
+          entry,
+          epoch: through.epoch,
+          sequence: through.sequence
+        })
+      }
+    })
+  }
+}
+
+/** A session's parts, read when a turn ends: the capture is built before they exist. */
+function piRpcHistoryCaptureParts(
+  session: Pick<PiRpcSession, 'input' | 'connection' | 'lane' | 'turns' | 'openedLink'>,
+  deps: PiHistoryCaptureDeps & { logger: StructuredAgentSessionLogger }
+): PiHistoryCaptureParts {
+  return {
+    sessionId: session.input.identity.sessionId,
+    fence: session.input.fence,
+    connection: session.connection,
+    sink: session.input.events,
+    drained: () => session.lane.drained(),
+    activity: () => session.turns.activity,
+    handle: () => session.openedLink?.handle ?? null,
+    ...(deps.persistResumePoint ? { persist: deps.persistResumePoint } : {}),
+    logger: deps.logger
+  }
+}
diff --git a/src/main/pi/rpc-history-catch-up.test.ts b/src/main/pi/rpc-history-catch-up.test.ts
new file mode 100644
index 00000000000..a0bbd2bd97f
--- /dev/null
+++ b/src/main/pi/rpc-history-catch-up.test.ts
@@ -0,0 +1,343 @@
+// Turns made outside a Pi chat, through the path they take: Pi's `get_entries` → walk → staged
+// rows → on-disk journal, with the claim the handle would carry between opens.
+
+import { mkdtemp, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types'
+import { AGENT_JOURNAL_THREAD_SCOPE } from '../../shared/agent-session-journal-types'
+import { createTrackedJournalOpener } from '../native-chat/agent-session-journal/journal-host-database-test-support'
+import { createDeferredStructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
+import { testEventSinkLogging } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support'
+import { projectStructuredAgentSessionMessages } from '../../shared/structured-agent-session-message-projection'
+import { JsonlRpcResponseError } from '../jsonl-rpc/peer'
+import { catchUpPiSession } from './rpc-history-catch-up'
+import { PiHistoryCapture } from './rpc-history-capture'
+import { decodePiHistoryClaim, type PiHistoryClaim } from './rpc-history-journal-view'
+
+const SESSION = 'session-pi-catch-up'
+const logger = { warn: vi.fn(), error: vi.fn() }
+
+type Entry = { type: 'message'; id: string; parentId: string | null; message: unknown }
+function entry(
+  id: string,
+  parentId: string | null,
+  role: 'user' | 'assistant',
+  text: string
+): Entry {
+  return { type: 'message', id, parentId, message: { role, content: [{ type: 'text', text }] } }
+}
+
+/** Answers `get_entries` the way Pi 1.3.0 does (`rpc-mode.js:503-514`), from what it loaded. */
+function piWith(entries: Entry[], leafId = entries.at(-1)?.id ?? null) {
+  return {
+    request: async (command: string, params: Record<string, unknown> = {}) => {
+      expect(command).toBe('get_entries')
+      if (params.since === undefined) {
+        return { entries, leafId }
+      }
+      const index = entries.findIndex((candidate) => candidate.id === params.since)
+      if (index === -1) {
+        throw new JsonlRpcResponseError('get_entries', `Entry not found: ${String(params.since)}`)
+      }
+      return { entries: entries.slice(index + 1), leafId }
+    }
+  }
+}
+
+const journals = createTrackedJournalOpener()
+const cleanups: (() => Promise<void>)[] = []
+afterEach(async () => {
+  for (const cleanup of cleanups.splice(0)) {
+    await cleanup()
+  }
+  logger.warn.mockClear()
+})
+
+async function chat() {
+  const root = await mkdtemp(join(tmpdir(), 'orca-pi-catch-up-'))
+  const file = join(root, 'session.jsonl')
+  const journal = await journals.open({
+    identity: {
+      sessionId: SESSION,
+      workspaceId: 'workspace-1',
+      hostId: 'local',
+      agent: 'pi',
+      providerHandle: null
+    },
+    stateDirectory: root,
+    now: () => 1_000
+  })
+  const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging())
+  deferred.bind({ journal, fence: 1, publish: () => {} })
+  cleanups.push(async () => {
+    deferred.close()
+    await journals.closeAll()
+    await rm(root, { recursive: true, force: true })
+  })
+  let claim: PiHistoryClaim | null = null
+  let ordinal = 0
+  const live = (recordId: string): AgentJournalItemIdentity => ({
+    provider: 'legacy',
+    agent: 'pi',
+    sessionId: SESSION,
+    recordId
+  })
+  const save = async (entries: Entry[]) =>
+    writeFile(
+      file,
+      `${[{ type: 'session', id: 'pi-1' }, ...entries].map((e) => JSON.stringify(e)).join('\n')}\n`
+    )
+  return {
+    /** One open of the chat's Pi, which has just loaded `entries` from its file. */
+    open: async (entries: Entry[], leafId?: string) => {
+      await save(entries)
+      const result = await catchUpPiSession({
+        connection: piWith(entries, leafId),
+        sessionId: SESSION,
+        sessionFile: file,
+        lineage: new Set([file]),
+        claim,
+        sink: deferred.sink,
+        signal: new AbortController().signal,
+        assertCurrent: () => {},
+        logger
+      })
+      claim = result.claim
+      return result
+    },
+    /** A turn the chat ran: its prompt and answer rows, then the capture its end starts. */
+    turn: async (text: string, entries: Entry[], anchor: string | null) => {
+      ordinal += 1
+      deferred.sink.appendItem(
+        { provider: 'orca', clientMessageId: `m${ordinal}` },
+        { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] },
+        { turnScope: AGENT_JOURNAL_THREAD_SCOPE }
+      )
+      deferred.sink.appendItem(
+        live(`run:g:${ordinal}`),
+        { kind: 'turn', turnId: `run:g:${ordinal}`, state: 'completed' },
+        { turnScope: AGENT_JOURNAL_THREAD_SCOPE }
+      )
+      deferred.sink.appendItem(
+        live(`message:g:${ordinal}`),
+        { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: `answer ${text}` }] },
+        { turnScope: { kind: 'turn', turnItemId: `legacy/pi/${SESSION}/run:g:${ordinal}` } }
+      )
+      const persist = vi.fn(async ({ handle }: { handle: { resumeCursor?: string } }) => {
+        claim = decodePiHistoryClaim(handle.resumeCursor)
+      })
+      const capture = new PiHistoryCapture(() => ({
+        sessionId: SESSION,
+        fence: 1,
+        connection: piWith(entries),
+        sink: deferred.sink,
+        drained: async () => {},
+        activity: () => 0,
+        handle: () => ({ transport: 'jsonl-rpc', agent: 'pi', nativeId: file }),
+        persist,
+        logger
+      }))
+      capture.start({ entry: anchor })
+      capture.turnEnded()
+      await vi.waitFor(() => expect(persist).toHaveBeenCalledOnce())
+    },
+    /** A prompt handed to the chat's Pi whose turn never reached the journal (Orca died). */
+    crashedPrompt: async (text: string) => {
+      ordinal += 1
+      deferred.sink.appendItem(
+        { provider: 'orca', clientMessageId: `m${ordinal}` },
+        { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] },
+        { turnScope: AGENT_JOURNAL_THREAD_SCOPE }
+      )
+      await deferred.sink.written?.()
+    },
+    queuedPrompt: async (text: string) => {
+      await journal.appendSubmission({
+        clientMessageId: 'queued',
+        payloadFingerprint: 'queued',
+        body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] },
+        fence: 1,
+        handoverRecorded: true
+      })
+    },
+    projected: () =>
+      projectStructuredAgentSessionMessages(journal.snapshot().items, [], journal.submissions()),
+    items: () => journal.snapshot().items,
+    texts: () =>
+      journal.snapshot().items.flatMap((item) => {
+        if (item.body.kind === 'message') {
+          const text = item.body.blocks.map((block) => (block.type === 'text' ? block.text : ''))
+          return [`${item.body.role}:${text.join('')}`]
+        }
+        return item.body.kind === 'status' && item.body.presentation
+          ? [`(${item.body.presentation})`]
+          : []
+      })
+  }
+}
+
+describe('Pi chat catch-up', () => {
+  it('adds terminal turns once, in Pi order, under one divider, and places a rewind', async () => {
+    const c = await chat()
+    expect(await c.open([])).toEqual({ claim: null, current: true })
+    const own = [entry('u1', null, 'user', 'ask 1'), entry('a1', 'u1', 'assistant', 'answer ask 1')]
+    await c.turn('ask 1', own, null)
+
+    const fullAnswer = `${'long answer '.repeat(2000)}TAIL`
+    const outside = [
+      ...own,
+      entry('u2', 'a1', 'user', 'ask 2'),
+      entry('a2', 'u2', 'assistant', fullAnswer),
+      entry('u3', 'a2', 'user', 'ask 3'),
+      entry('a3', 'u3', 'assistant', 'answer 3')
+    ]
+    await c.open(outside)
+    const caughtUp = [
+      'user:ask 1',
+      'assistant:answer ask 1',
+      '(continued-outside)',
+      'user:ask 2',
+      `assistant:${fullAnswer}`,
+      'user:ask 3',
+      'assistant:answer 3'
+    ]
+    expect(c.texts()).toEqual(caughtUp)
+
+    // Reopening adds nothing.
+    await c.open(outside)
+    expect(c.texts()).toEqual(caughtUp)
+
+    // The terminal went back to ask 1's answer and continued from there.
+    const rewound = [
+      ...outside,
+      entry('u4', 'a1', 'user', 'ask 4'),
+      entry('a4', 'u4', 'assistant', 'answer 4')
+    ]
+    await c.open(rewound)
+    const afterRewind = [
+      ...caughtUp,
+      '(continued-outside-rewound)',
+      'user:ask 4',
+      'assistant:answer 4'
+    ]
+    expect(c.texts()).toEqual(afterRewind)
+
+    // Re-editing the first prompt starts a new root in the same session tree.
+    const rootRewind = [
+      ...rewound,
+      entry('u5', null, 'user', 'ask 5'),
+      entry('a5', 'u5', 'assistant', 'answer 5')
+    ]
+    expect(await c.open(rootRewind)).toMatchObject({ claim: { entry: 'a5' }, current: true })
+    const afterRootRewind = [
+      ...afterRewind,
+      '(continued-outside-rewound)',
+      'user:ask 5',
+      'assistant:answer 5'
+    ]
+    expect(c.texts()).toEqual(afterRootRewind)
+    await c.open(rootRewind)
+    expect(c.texts()).toEqual(afterRootRewind)
+    expect(logger.warn).not.toHaveBeenCalled()
+  })
+
+  it('keeps an image-only terminal prompt and its reply across repeated opens', async () => {
+    const c = await chat()
+    await c.open([])
+    const own = [entry('u1', null, 'user', 'ask 1'), entry('a1', 'u1', 'assistant', 'answer ask 1')]
+    await c.turn('ask 1', own, null)
+    const data =
+      'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+/lXcAAAAASUVORK5CYII='
+    const outside = [
+      ...own,
+      {
+        ...entry('u2', 'a1', 'user', ''),
+        message: {
+          role: 'user',
+          content: [
+            { type: 'text', text: '' },
+            { type: 'image', data, mimeType: 'image/png' }
+          ]
+        }
+      },
+      entry('a2', 'u2', 'assistant', 'A tiny image')
+    ]
+    expect(await c.open(outside)).toMatchObject({ claim: { entry: 'a2' }, current: true })
+    const imported = c.items()
+    expect(imported.slice(-3).map((item) => item.body)).toEqual([
+      expect.objectContaining({ kind: 'status', presentation: 'continued-outside' }),
+      {
+        kind: 'message',
+        role: 'user',
+        blocks: [{ type: 'image-ref', url: `data:image/png;base64,${data}` }]
+      },
+      { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'A tiny image' }] }
+    ])
+    await c.open(outside)
+    expect(c.items()).toEqual(imported)
+    expect(logger.warn).not.toHaveBeenCalled()
+  })
+
+  it('imports outside work above the queued prompt that starts a resting Pi', async () => {
+    const c = await chat()
+    await c.open([])
+    const own = [entry('u1', null, 'user', 'ask 1'), entry('a1', 'u1', 'assistant', 'answer ask 1')]
+    await c.turn('ask 1', own, null)
+    await c.queuedPrompt('ask next')
+    const outside = [
+      ...own,
+      entry('u2', 'a1', 'user', 'outside'),
+      entry('a2', 'u2', 'assistant', 'outside answer')
+    ]
+    expect(await c.open(outside)).toMatchObject({ claim: { entry: 'a2' }, current: true })
+    const projected = c.projected()
+    expect(
+      projected
+        .filter((message) => message.role === 'user' || message.role === 'assistant')
+        .map((message) =>
+          message.blocks
+            .filter((block) => block.type === 'text')
+            .map((block) => block.text)
+            .join('')
+        )
+    ).toEqual(['ask 1', 'answer ask 1', 'outside', 'outside answer', 'ask next'])
+    await c.open(outside)
+    expect(c.projected()).toEqual(projected)
+    expect(logger.warn).not.toHaveBeenCalled()
+  })
+
+  it('skips a suffix it cannot split after a crash, then shows later terminal turns', async () => {
+    const c = await chat()
+    await c.open([])
+    const own = [entry('u1', null, 'user', 'ask 1'), entry('a1', 'u1', 'assistant', 'answer ask 1')]
+    await c.turn('ask 1', own, null)
+
+    // The chat's Pi saved ask 2 before Orca died; then a terminal wrote ask 3.
+    await c.crashedPrompt('ask 2')
+    const mixed = [
+      ...own,
+      entry('u2', 'a1', 'user', 'ask 2'),
+      entry('u3', 'u2', 'user', 'ask 3'),
+      entry('a3', 'u3', 'assistant', 'answer 3')
+    ]
+    await c.open(mixed)
+    expect(c.texts()).toEqual(['user:ask 1', 'assistant:answer ask 1', 'user:ask 2'])
+
+    // The skip is saved: the next terminal turn shows, nothing before it does.
+    await c.open([
+      ...mixed,
+      entry('u4', 'a3', 'user', 'ask 4'),
+      entry('a4', 'u4', 'assistant', 'answer 4')
+    ])
+    expect(c.texts()).toEqual([
+      'user:ask 1',
+      'assistant:answer ask 1',
+      'user:ask 2',
+      '(continued-outside)',
+      'user:ask 4',
+      'assistant:answer 4'
+    ])
+  })
+})
diff --git a/src/main/pi/rpc-history-catch-up.ts b/src/main/pi/rpc-history-catch-up.ts
new file mode 100644
index 00000000000..3a1499fc042
--- /dev/null
+++ b/src/main/pi/rpc-history-catch-up.ts
@@ -0,0 +1,161 @@
+import { waitForPromiseWithSignal } from '../../shared/abort-signal-reason'
+import { JsonlRpcResponseError } from '../jsonl-rpc/peer'
+import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
+import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger'
+import { runHistoryCatchUp } from '../native-chat/agent-session-wire/structured-agent-session-history-catch-up'
+import {
+  createPiHistoryImportPlan,
+  piBranchInTree,
+  piBranchSince,
+  piEntriesAnswerSchema,
+  type PiEntriesAnswer,
+  type PiSessionEntry
+} from './rpc-history-import'
+import {
+  readPiHistoryJournalView,
+  samePiHistoryJournalView,
+  usablePiHistoryClaim,
+  type PiHistoryClaim
+} from './rpc-history-journal-view'
+import type { PiRpcConnection } from './rpc-session'
+import { readPiSessionLastEntryId } from './rpc-session-file-tail'
+
+export const PI_HISTORY_CATCH_UP_TIMEOUT_MS = 30_000
+
+/** What a catch-up leaves the chat with: the claim to save on this open's link, and whether it
+ *  names Pi's current leaf (a capture may then continue from it) or was only carried forward. */
+export type PiHistoryCatchUp = { claim: PiHistoryClaim | null; current: boolean }
+
+/** `get_entries`, whose oversized answer fails only this read. */
+export async function readPiEntries(
+  connection: Pick<PiRpcConnection, 'request'>,
+  since: string | null,
+  timeoutMs?: number
+): Promise<PiEntriesAnswer> {
+  return piEntriesAnswerSchema.parse(
+    await connection.request('get_entries', since === null ? {} : { since }, {
+      oversizedReply: 'reject',
+      ...(timeoutMs === undefined ? {} : { timeoutMs })
+    })
+  )
+}
+
+// Pi has no writer lock: a terminal can write the conversation only while the chat's Pi isn't
+// running, and every start of the chat's Pi opens here, with Pi's view freshly loaded from disk.
+
+/** Optional reconciliation never gates the open; a failure keeps the claim the chat already had. */
+export async function catchUpPiSession(input: {
+  connection: Pick<PiRpcConnection, 'request'>
+  sessionId: string
+  sessionFile: string
+  /** Every file of this conversation: entries keep their ids across a directory fork. */
+  lineage: ReadonlySet<string>
+  claim: PiHistoryClaim | null
+  sink: StructuredAgentSessionEventSink | undefined
+  signal: AbortSignal
+  assertCurrent: () => void
+  timeoutMs?: number
+  logger: StructuredAgentSessionLogger
+}): Promise<PiHistoryCatchUp> {
+  const carried: PiHistoryCatchUp = { claim: input.claim, current: false }
+  if (!input.sink) {
+    return carried
+  }
+  // Set once Pi's current leaf is known; it stands only if nothing after it fails.
+  let leaf: { entry: string | null } | undefined
+  const outcome = await runHistoryCatchUp({
+    agentName: 'Pi',
+    scope: 'pi-history-catch-up',
+    sessionId: input.sessionId,
+    sink: input.sink,
+    signal: input.signal,
+    assertCurrent: input.assertCurrent,
+    timeoutMs: input.timeoutMs ?? PI_HISTORY_CATCH_UP_TIMEOUT_MS,
+    logger: input.logger,
+    readView: (history) => readPiHistoryJournalView(history, input.lineage),
+    sameView: samePiHistoryJournalView,
+    prepare: async ({ view: readView, journal, signal, deadline, check, warn }) => {
+      const view = readView()
+      const read = async (since: string | null) => {
+        check()
+        const answer = await waitForPromiseWithSignal(
+          readPiEntries(input.connection, since, Math.max(1, deadline - Date.now())),
+          signal
+        )
+        check()
+        return answer
+      }
+      const claim = usablePiHistoryClaim(input.claim, view)
+      // Without a claim, any Pi history the journal already shows (adopted, or an older build's
+      // turns) leaves nothing to split it by; with one, neither does the chat's own work after it.
+      const known = view.ownWorkSequence > 0 || view.accounted.size > 0
+      if (claim ? view.ownWorkSequence > claim.sequence : known) {
+        leaf = { entry: await currentLeaf(input.sessionFile, read) }
+        return null
+      }
+      let answer = claim ? await readSince(read, claim.entry) : null
+      let branch: { path: PiSessionEntry[]; rewound: boolean } | null =
+        answer && claim ? piBranchSince(answer, claim.entry) : null
+      if (!answer || !branch) {
+        answer = await read(null)
+        const inTree = piBranchInTree(answer, claim?.entry ?? null)
+        if (inTree === 'unknown-claim') {
+          leaf = { entry: answer.leafId }
+          return null
+        }
+        if (!inTree) {
+          warn('incomplete-history')
+          return null
+        }
+        branch = inTree
+      }
+      leaf = { entry: answer.leafId }
+      return createPiHistoryImportPlan({
+        sessionFile: input.sessionFile,
+        journal,
+        path: branch.path.filter((entry) => !view.accounted.has(entry.id)),
+        rewound: branch.rewound,
+        divider: view.hasRows,
+        check
+      })
+    }
+  })
+  const history = input.sink.journalHistory?.()
+  if (outcome === 'skipped' || !leaf || !history) {
+    return carried
+  }
+  const cursor = history.cursor()
+  return {
+    claim:
+      leaf.entry === null
+        ? null
+        : { entry: leaf.entry, epoch: cursor.epoch, sequence: cursor.sequence },
+    current: true
+  }
+}
+
+/** A claim Pi no longer has (a replaced file) answers as if the whole tree had to be read. */
+async function readSince(
+  read: (since: string | null) => Promise<PiEntriesAnswer>,
+  entry: string
+): Promise<PiEntriesAnswer | null> {
+  try {
+    return await read(entry)
+  } catch (error) {
+    if (error instanceof JsonlRpcResponseError && error.message.startsWith('Entry not found')) {
+      return null
+    }
+    throw error
+  }
+}
+
+/** Pi's leaf, asked cheaply after the file's last entry when the file still matches what Pi
+ *  loaded, else from the whole tree. */
+async function currentLeaf(
+  sessionFile: string,
+  read: (since: string | null) => Promise<PiEntriesAnswer>
+): Promise<string | null> {
+  const last = await readPiSessionLastEntryId(sessionFile).catch(() => null)
+  const hinted = last === null ? null : await readSince(read, last)
+  return (hinted ?? (await read(null))).leafId
+}
diff --git a/src/main/pi/rpc-history-import.ts b/src/main/pi/rpc-history-import.ts
new file mode 100644
index 00000000000..d2f1bf7deee
--- /dev/null
+++ b/src/main/pi/rpc-history-import.ts
@@ -0,0 +1,146 @@
+import { z } from 'zod'
+import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key'
+import {
+  AGENT_JOURNAL_THREAD_SCOPE,
+  type AgentJournalItemIdentity
+} from '../../shared/agent-session-journal-types'
+import { legacyItemBody } from '../native-chat/agent-session-journal/journal-legacy-import'
+import { decodePiSessionEntryLine } from '../native-chat/agent-session-journal/journal-pi-active-branch'
+import { journalLifecycleItemMutation } from '../native-chat/agent-session-journal/journal-row-builders'
+import type { StructuredAgentSessionTransitionJournal } from '../native-chat/agent-session-wire/structured-agent-session-transition'
+import {
+  continuedOutsideNotice,
+  continuedOutsideRecordId,
+  createHistoryCatchUpBatch,
+  type HistoryCatchUpPlan
+} from '../native-chat/agent-session-wire/structured-agent-session-history-catch-up'
+
+const piEntrySchema = z.looseObject({
+  id: z.string().min(1),
+  parentId: z.string().min(1).nullable(),
+  type: z.string()
+})
+export type PiSessionEntry = z.infer<typeof piEntrySchema>
+
+/** `get_entries`: the entries after `since` in the order Pi appended them, and its leaf. */
+export const piEntriesAnswerSchema = z.object({
+  entries: z.array(piEntrySchema),
+  leafId: z.string().min(1).nullable()
+})
+export type PiEntriesAnswer = z.infer<typeof piEntriesAnswerSchema>
+
+/** The answer to `get_entries since: claim`: the active branch after the claim, oldest first, when
+ *  the leaf's `parentId` chain reaches the claim inside it; null when the terminal went back to an
+ *  earlier point, which only the whole tree can place. The answer lists in append order and keeps
+ *  abandoned branches, so the branch is always walked from the leaf. */
+export function piBranchSince(
+  answer: PiEntriesAnswer,
+  claim: string
+): { path: PiSessionEntry[]; rewound: false } | null {
+  const walked = walkFromLeaf(answer, (id) => id === claim)
+  return walked?.stop === claim ? { path: walked.path, rewound: false } : null
+}
+
+/**
+ * The whole tree's active branch after the newest entry the chat accounts for: the claim, or the
+ * nearest of its ancestors where the terminal branched off (rewound). Without a claim, the whole
+ * branch. 'unknown-claim' when the claim isn't in this conversation at all.
+ */
+export function piBranchInTree(
+  answer: PiEntriesAnswer,
+  claim: string | null
+): { path: PiSessionEntry[]; rewound: boolean } | 'unknown-claim' | null {
+  const byId = new Map(answer.entries.map((entry) => [entry.id, entry]))
+  const accounted = new Set<string>()
+  for (let id = claim; id !== null && !accounted.has(id); id = byId.get(id)?.parentId ?? null) {
+    if (!byId.has(id)) {
+      return 'unknown-claim'
+    }
+    accounted.add(id)
+  }
+  const walked = walkFromLeaf(answer, (id) => accounted.has(id))
+  if (!walked) {
+    return null
+  }
+  return { path: walked.path, rewound: walked.stop !== claim }
+}
+
+function walkFromLeaf(
+  answer: PiEntriesAnswer,
+  stopsAt: (id: string) => boolean
+): { path: PiSessionEntry[]; stop: string | null } | null {
+  const byId = new Map(answer.entries.map((entry) => [entry.id, entry]))
+  const path: PiSessionEntry[] = []
+  let current = answer.leafId
+  while (current !== null && !stopsAt(current)) {
+    const entry = byId.get(current)
+    if (!entry) {
+      return null
+    }
+    // Each id is visited once: a cycle ends the walk as incomplete history.
+    byId.delete(current)
+    path.push(entry)
+    current = entry.parentId
+  }
+  return { path: path.toReversed(), stop: current }
+}
+
+/** Rows keyed as adoption keys them (`legacy/pi/<session file>/<entry id>`), so both dedupe. */
+export function piHistoryEntryIdentity(
+  sessionFile: string,
+  entryId: string
+): AgentJournalItemIdentity {
+  return { provider: 'legacy', agent: 'pi', sessionId: sessionFile, recordId: entryId }
+}
+
+/** Outside entries as journal rows, staged whole before anything is written; entries the journal
+ *  already holds are left out first, so an earlier import never vetoes newer entries. Null when
+ *  nothing on the path shows as a row. */
+export function createPiHistoryImportPlan(input: {
+  sessionFile: string
+  journal: StructuredAgentSessionTransitionJournal
+  path: readonly PiSessionEntry[]
+  rewound: boolean
+  /** Only a chat that already shows something needs the divider. */
+  divider: boolean
+  check: () => void
+}): HistoryCatchUpPlan | null {
+  const batch = createHistoryCatchUpBatch()
+  const entryIds: string[] = []
+  for (const entry of input.path) {
+    input.check()
+    const identity = piHistoryEntryIdentity(input.sessionFile, entry.id)
+    if (input.journal.item(input.journal.canonicalItemId(agentJournalItemKey(identity))) !== null) {
+      continue
+    }
+    const message = decodePiSessionEntryLine(JSON.stringify(entry), entry.id)
+    if (!message) {
+      continue
+    }
+    const body = legacyItemBody(message)
+    if (entryIds.length === 0 && input.divider) {
+      batch.push(
+        journalLifecycleItemMutation(
+          { turnScope: AGENT_JOURNAL_THREAD_SCOPE },
+          piHistoryEntryIdentity(input.sessionFile, continuedOutsideRecordId(entry.id)),
+          continuedOutsideNotice(input.rewound)
+        )
+      )
+    }
+    // Imported prompts are history, never sends delivery recovery could claim.
+    batch.push(
+      journalLifecycleItemMutation(
+        {
+          turnScope: AGENT_JOURNAL_THREAD_SCOPE,
+          ...(body.kind === 'message' && body.role === 'user' ? { echoOf: null } : {})
+        },
+        identity,
+        body
+      )
+    )
+    entryIds.push(entry.id)
+  }
+  return batch.size === 0
+    ? null
+    : batch.finish('pi-history-import', [input.journal.epoch, input.sessionFile, entryIds])
+}
diff --git a/src/main/pi/rpc-history-journal-view.ts b/src/main/pi/rpc-history-journal-view.ts
new file mode 100644
index 00000000000..1101041bfd2
--- /dev/null
+++ b/src/main/pi/rpc-history-journal-view.ts
@@ -0,0 +1,109 @@
+import { z } from 'zod'
+import {
+  agentJournalSubmissionKey,
+  parseAgentJournalItemKey
+} from '../../shared/agent-session-journal-item-key'
+import { isQueuedAgentJournalSubmission } from '../../shared/agent-session-queued-submission'
+import type { StructuredAgentSessionHistoryJournal } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
+
+/**
+ * Pi's entries up to `entry` are accounted for (shown, or deliberately skipped) by the chat's journal
+ * up to `sequence` of `epoch`. Saved as the Pi handle's resume cursor, which only Pi's adapter reads.
+ */
+export type PiHistoryClaim = { entry: string; epoch: string; sequence: number }
+
+const claimSchema = z.object({
+  entry: z.string().min(1),
+  epoch: z.string().min(1),
+  sequence: z.number().int().nonnegative()
+})
+
+export function encodePiHistoryClaim(claim: PiHistoryClaim): string {
+  return JSON.stringify(claim)
+}
+
+/** A cursor an older build left, or one this build can't read, is no claim at all. */
+export function decodePiHistoryClaim(value: string | undefined): PiHistoryClaim | null {
+  if (value === undefined) {
+    return null
+  }
+  try {
+    const parsed = claimSchema.safeParse(JSON.parse(value))
+    return parsed.success ? parsed.data : null
+  } catch {
+    return null
+  }
+}
+
+export type PiHistoryJournalView = {
+  epoch: string
+  sequence: number
+  hasRows: boolean
+  /** Pi entries the journal already holds as rows, from any file of this conversation. */
+  accounted: ReadonlySet<string>
+  /** Where the chat's own newest work was created: a turn its Pi ran, or a prompt handed to it. */
+  ownWorkSequence: number
+}
+
+export function readPiHistoryJournalView(
+  journal: StructuredAgentSessionHistoryJournal,
+  lineage: ReadonlySet<string>
+): PiHistoryJournalView {
+  const accounted = new Set<string>()
+  const submissions = journal.submissions()
+  const queued = new Set(
+    submissions
+      .filter(isQueuedAgentJournalSubmission)
+      .map((submission) => agentJournalSubmissionKey(submission.clientMessageId))
+  )
+  let hasRows = false
+  let ownWorkSequence = 0
+  journal.visitItems((itemId, sequence, body) => {
+    hasRows = true
+    const identity = parseAgentJournalItemKey(itemId)
+    if (
+      identity?.provider === 'legacy' &&
+      identity.agent === 'pi' &&
+      lineage.has(identity.sessionId)
+    ) {
+      accounted.add(identity.recordId)
+      return
+    }
+    if (
+      body.kind === 'turn' ||
+      (identity?.provider === 'orca' && body.kind === 'message' && !queued.has(itemId))
+    ) {
+      ownWorkSequence = Math.max(ownWorkSequence, sequence)
+    }
+  })
+  // A prompt Pi accepted before its turn row landed still counts from its acceptance.
+  for (const submission of submissions) {
+    const receipt = journal.receiptFor(submission.clientMessageId)
+    if (receipt?.cursor.epoch === journal.epoch) {
+      ownWorkSequence = Math.max(ownWorkSequence, receipt.cursor.sequence)
+    }
+  }
+  const cursor = journal.cursor()
+  return { epoch: cursor.epoch, sequence: cursor.sequence, hasRows, accounted, ownWorkSequence }
+}
+
+/** Usable only in the journal it describes, and never past what that journal holds. */
+export function usablePiHistoryClaim(
+  claim: PiHistoryClaim | null,
+  view: PiHistoryJournalView
+): PiHistoryClaim | null {
+  return claim && claim.epoch === view.epoch && claim.sequence <= view.sequence ? claim : null
+}
+
+export function samePiHistoryJournalView(
+  a: PiHistoryJournalView,
+  b: PiHistoryJournalView
+): boolean {
+  return (
+    a.epoch === b.epoch &&
+    a.hasRows === b.hasRows &&
+    a.ownWorkSequence === b.ownWorkSequence &&
+    a.accounted.size === b.accounted.size &&
+    [...a.accounted].every((id) => b.accounted.has(id))
+  )
+}
diff --git a/src/main/pi/rpc-launch-resolution.ts b/src/main/pi/rpc-launch-resolution.ts
index f4c5ec61c00..d2195fea53d 100644
--- a/src/main/pi/rpc-launch-resolution.ts
+++ b/src/main/pi/rpc-launch-resolution.ts
@@ -5,6 +5,7 @@ import { agentSessionRefusalError } from '../../shared/agent-session-wire-refusa
 import {
   agentSessionProviderHandleChainHead,
   agentSessionProviderHandleKey,
+  type AgentSessionProviderHandleChain,
   type AgentSessionProviderHandleLink
 } from '../../shared/agent-session-provider-handle'
 import { requireLegacyAgentSessionAccountHome } from '../../shared/agent-session-account-home'
@@ -32,6 +33,8 @@ import {
 export type PiRpcResolvedLaunch = PiRpcLaunchOptions & {
   previous: AgentSessionProviderHandleLink | null
   replacement?: 'unsaved' | 'restore-failed'
+  /** The files of the conversation this launch continues, newest first; empty for a new one. */
+  lineage?: readonly string[]
 }
 export type PiRpcLaunchPartDeps = {
   resolveCommand?: typeof resolveCliCommand
@@ -197,6 +200,9 @@ export function piRpcLaunchPart(
         command,
         fullAccess: deps.resolveFullAccess?.() ?? true,
         previous,
+        ...(sessionFile || forkFile
+          ? { lineage: piConversationFiles(record.providerHandleChain) }
+          : {}),
         ...(sessionFile ? { sessionFile } : {}),
         ...(forkFile ? { forkFile } : {}),
         ...(replacement ? { replacement } : {})
@@ -207,6 +213,22 @@ export function piRpcLaunchPart(
   }
 }
 
+/** The head back through resumes and directory forks, which copy every entry with its id, to the
+ *  session the conversation started as. */
+function piConversationFiles(chain: AgentSessionProviderHandleChain): string[] {
+  const files: string[] = []
+  for (const link of chain.toReversed()) {
+    if (link.handle.transport !== 'jsonl-rpc' || link.handle.agent !== 'pi') {
+      break
+    }
+    files.push(link.handle.nativeId)
+    if (link.origin !== 'resumed' && link.origin !== 'forked') {
+      break
+    }
+  }
+  return files
+}
+
 export function piRpcProviderLink(
   launch: PiRpcResolvedLaunch,
   file: string,
diff --git a/src/main/pi/rpc-runtime-registration.ts b/src/main/pi/rpc-runtime-registration.ts
index aa20e80f636..346665bece1 100644
--- a/src/main/pi/rpc-runtime-registration.ts
+++ b/src/main/pi/rpc-runtime-registration.ts
@@ -11,6 +11,7 @@ import { PI_RPC_AGENT } from './rpc-agent-definition'
 import { piRpcLaunchPart } from './rpc-launch-resolution'
 import { PiRpcSessionAdapter } from './rpc-session-adapter'
 import { PI_RPC_RUNTIME_METADATA } from './rpc-runtime-metadata'
+import { reviseAgentSessionProviderResumePoint } from '../runtime/agent-session-provider-handle-transition'
 
 function createPiRpcAdapter(context: StructuredAgentAdapterContext): StructuredAgentRuntimeAdapter {
   const { deps } = context
@@ -35,6 +36,11 @@ function createPiRpcAdapter(context: StructuredAgentAdapterContext): StructuredA
         ...(outcome.state === 'accepted' ? { providerIdentity: outcome.providerIdentity } : outcome)
       })
     },
+    persistResumePoint: async ({ sessionId, fence, handle }) => {
+      await context.store.transitionHandoff(sessionId, (record) =>
+        reviseAgentSessionProviderResumePoint({ record, fence, handle, now: Date.now() })
+      )
+    },
     logger: deps.logger
   })
 }
diff --git a/src/main/pi/rpc-session-file-tail.ts b/src/main/pi/rpc-session-file-tail.ts
new file mode 100644
index 00000000000..f09a79425fc
--- /dev/null
+++ b/src/main/pi/rpc-session-file-tail.ts
@@ -0,0 +1,62 @@
+import { open } from 'node:fs/promises'
+import { z } from 'zod'
+
+const BLOCK_BYTES = 64 * 1024
+/** One entry line is never longer than a reply the peer accepts. */
+const MAX_TAIL_BYTES = 8 * 1024 * 1024
+
+/**
+ * The id of the newest entry Pi saved, read backwards from the file's end. Only a hint for a cheap
+ * `get_entries since` that asks Pi for its leaf: Pi answers from what it loaded, which the file
+ * may no longer match, so the answer, not this, is what counts.
+ */
+export async function readPiSessionLastEntryId(file: string): Promise<string | null> {
+  const handle = await open(file, 'r')
+  try {
+    let position = (await handle.stat()).size
+    // The newest line not yet ended by an earlier newline, oldest bytes first.
+    let partial: Buffer[] = []
+    let partialBytes = 0
+    // A last line without its newline is still being appended; it is not an entry yet.
+    let complete = false
+    while (position > 0 && partialBytes <= MAX_TAIL_BYTES) {
+      const length = Math.min(BLOCK_BYTES, position)
+      position -= length
+      const block = Buffer.alloc(length)
+      await handle.read(block, 0, length, position)
+      let end = length
+      for (let index = length - 1; index >= 0; index--) {
+        if (block[index] !== 0x0a) {
+          continue
+        }
+        const id = complete ? entryId([block.subarray(index + 1, end), ...partial]) : null
+        if (id !== null) {
+          return id
+        }
+        complete = true
+        partial = []
+        partialBytes = 0
+        end = index
+      }
+      partial.unshift(block.subarray(0, end))
+      partialBytes += end
+    }
+    return position === 0 && complete ? entryId(partial) : null
+  } finally {
+    await handle.close()
+  }
+}
+
+const entryHeadSchema = z.looseObject({ type: z.unknown(), id: z.string().min(1) })
+
+function entryId(parts: Buffer[]): string | null {
+  let record: unknown
+  try {
+    record = JSON.parse(Buffer.concat(parts).toString('utf8'))
+  } catch {
+    return null
+  }
+  const head = entryHeadSchema.safeParse(record)
+  // The header names the session, not an entry.
+  return head.success && head.data.type !== 'session' ? head.data.id : null
+}
diff --git a/src/main/pi/rpc-session-startup.ts b/src/main/pi/rpc-session-startup.ts
index 518ebe4650b..1b5a458f33a 100644
--- a/src/main/pi/rpc-session-startup.ts
+++ b/src/main/pi/rpc-session-startup.ts
@@ -18,9 +18,14 @@ import { piRpcProviderLink, type PiRpcResolvedLaunch } from './rpc-launch-resolu
 import { piModelOptionId, readPiRpcCommands, readPiRpcSessionOptions } from './rpc-options'
 import { piRpcStateSchema, type PiRpcState } from './rpc-protocol'
 import type { PiRpcConnection, PiRpcSessionDeps } from './rpc-session'
+import { catchUpPiSession } from './rpc-history-catch-up'
+import type { PiHistoryAnchor } from './rpc-history-capture'
+import { decodePiHistoryClaim, encodePiHistoryClaim } from './rpc-history-journal-view'
 
 export class PiRpcSessionStartup {
   commands?: AgentSessionSlashCommand[]
+  /** Set by an open that left its claim at Pi's current leaf. */
+  historyAnchor?: PiHistoryAnchor
 
   constructor(
     private readonly input: StructuredAgentSessionAcquireInput,
@@ -51,13 +56,36 @@ export class PiRpcSessionStartup {
       Date.now()
     )
     this.setState(state)
+    // Only the conversation this open continues can carry the claim; a replacement starts over.
+    const continued = launch.sessionFile !== undefined || launch.forkFile !== undefined
+    const history = await this.wait(
+      catchUpPiSession({
+        connection: this.connection,
+        sessionId: this.input.identity.sessionId,
+        sessionFile: state.sessionFile,
+        lineage: new Set([...(launch.lineage ?? []), state.sessionFile]),
+        claim: continued ? decodePiHistoryClaim(launch.previous?.handle.resumeCursor) : null,
+        sink: this.input.events,
+        signal,
+        assertCurrent: () => this.assertLive(signal),
+        logger: this.deps.logger
+      }),
+      signal
+    )
+    this.assertLive(signal)
+    if (history.current) {
+      this.historyAnchor = { entry: history.claim?.entry ?? null }
+    }
     const current: AgentSessionOptionsResult['current'] = {
       model: state.model ? piModelOptionId(state.model.provider, state.model.id) : '',
       ...(state.thinkingLevel ? { effort: state.thinkingLevel } : {})
     }
     this.assertLive(signal)
     current.confirmed = [...(current.model ? ['model'] : []), ...(current.effort ? ['effort'] : [])]
-    return { link, reportedOptions: current }
+    const claimed = history.claim
+      ? { ...link, handle: { ...link.handle, resumeCursor: encodePiHistoryClaim(history.claim) } }
+      : link
+    return { link: claimed, reportedOptions: current }
   }
 
   /** After the open: the listing and the `/` commands reach the host as later reports, and a read
diff --git a/src/main/pi/rpc-session.ts b/src/main/pi/rpc-session.ts
index 4f934fe9857..f241bd50e1d 100644
--- a/src/main/pi/rpc-session.ts
+++ b/src/main/pi/rpc-session.ts
@@ -25,6 +25,7 @@ import type { PiRpcResolvedLaunch } from './rpc-launch-resolution'
 import type { StructuredAgentSessionProviderOpenResult } from '../native-chat/agent-session-wire/structured-agent-session-provider-open-contract'
 import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle'
 import { PiRpcSessionStartup } from './rpc-session-startup'
+import { PiHistoryCapture, type PiHistoryCaptureDeps } from './rpc-history-capture'
 
 export type PiRpcConnection = Pick<
   JsonlRpcAgentConnection,
@@ -40,7 +41,7 @@ export type PiRpcConnection = Pick<
   | 'lastCloseResult'
   | 'onExit'
 >
-export type PiRpcSessionDeps = {
+export type PiRpcSessionDeps = PiHistoryCaptureDeps & {
   openConnection?: (
     launch: ProviderProcessLaunch,
     handlers: JsonlRpcAgentConnectionOptions
@@ -66,6 +67,7 @@ export class PiRpcSession {
   readonly heldSends = new Set<string>()
   readonly optionProof = { current: new Map<string, string>(), writes: new Map<string, symbol>() }
   readonly startup: PiRpcSessionStartup
+  readonly history = PiHistoryCapture.forSession(this, () => this.deps)
   initialOptions: Readonly<Record<string, string>> = {}
   openedLink: AgentSessionProviderHandleLink | null = null
   requestedClose = false
@@ -109,6 +111,7 @@ export class PiRpcSession {
       request: (command, params, options) => this.connection.request(command, params, options),
       settled: (id, outcome) => this.settle(id, outcome),
       idle: () => this.dialogs.cancelAll(),
+      turnEnded: () => this.history.turnEnded(),
       failed: (error) => this.fail(error),
       diagnostic: (error) =>
         deps.logger.warn('Pi context usage could not be read', {
@@ -178,6 +181,7 @@ export class PiRpcSession {
       ...(result.reportedOptions.effort ? { effort: result.reportedOptions.effort } : {})
     }
     this.openedLink = result.link ?? null
+    this.history.start(this.startup.historyAnchor)
     this.optionProof.current = new Map(Object.entries(this.initialOptions))
     void this.startup
       .readAfterOpen(this.generation, result.reportedOptions)
diff --git a/src/main/pi/rpc-turns.ts b/src/main/pi/rpc-turns.ts
index 34cd3e7bd23..e04f5a518ce 100644
--- a/src/main/pi/rpc-turns.ts
+++ b/src/main/pi/rpc-turns.ts
@@ -5,6 +5,7 @@ import type {
 } from '../native-chat/agent-session-wire/structured-agent-session-adapter'
 import type { ProviderTimelineEvent } from '../native-chat/agent-session-timeline/provider-timeline-event'
 import type { JsonlRpcRecord } from '../jsonl-rpc/peer'
+import type { JsonlRpcRequestOptions } from '../jsonl-rpc/peer-oversized-reply'
 import type { JsonlRpcTimelineLane } from '../jsonl-rpc/timeline-lane'
 import { PiRpcContextUsage } from './rpc-context-usage'
 import { PiRpcMessages } from './rpc-messages'
@@ -20,10 +21,12 @@ export type PiRpcTurnDeps = {
   request: (
     command: string,
     params?: Record<string, unknown>,
-    options?: { timeoutMs?: number | null }
+    options?: JsonlRpcRequestOptions
   ) => Promise<unknown>
   settled: (id: string, outcome: AgentSessionDispatchOutcome) => void
   idle: () => void
+  /** A turn's end reached the timeline. */
+  turnEnded?: () => void
   failed: (error: Error) => void
   diagnostic?: (error: unknown) => void
 }
@@ -89,6 +92,10 @@ export class PiRpcTurns {
   get working(): boolean {
     return this.active !== undefined
   }
+  /** Moves whenever Pi or a send does anything. */
+  get activity(): number {
+    return this.revision
+  }
 
   async submit(
     id: string,
@@ -284,6 +291,7 @@ export class PiRpcTurns {
     this.settling = false
     this.messages.reset()
     this.deps.idle()
+    this.deps.turnEnded?.()
   }
 
   private refreshUsage(): void {
diff --git a/src/shared/main-process-ndjson-framer.ts b/src/shared/main-process-ndjson-framer.ts
index abd66cf97b1..5d21814ddb0 100644
--- a/src/shared/main-process-ndjson-framer.ts
+++ b/src/shared/main-process-ndjson-framer.ts
@@ -23,6 +23,7 @@ export type NdjsonRejectedRecord =
       maxLineBytes: number
       observedBytes: number
       prefix: string
+      droppedCompleteRecords?: true
     }
   | { kind: 'invalid-json'; line: string; error: Error }
 
@@ -39,10 +40,6 @@ export type IncrementalNdjsonFramerOptions = {
   shouldPause?: () => boolean
 }
 
-function errorFrom(value: unknown): Error {
-  return value instanceof Error ? value : new Error(String(value))
-}
-
 function boundedUtf8Prefix(value: string, maxBytes: number): string {
   if (Buffer.byteLength(value, 'utf8') <= maxBytes) {
     return value
@@ -80,6 +77,7 @@ export function createIncrementalNdjsonFramer(
   let prefixSegments: string[] = []
   let prefixBytes = 0
   let discardingOversizedLine = false
+  let discardingPausedSuffix = false
   let pendingInput: string | null = null
   let pausedCompleteInput: string[] = []
   let pausedCompleteInputBytes = 0
@@ -118,6 +116,7 @@ export function createIncrementalNdjsonFramer(
         kind: 'line-too-long',
         maxLineBytes: maxQueuedCompleteInputBytes,
         observedBytes: pausedCompleteInputBytes + completeBytes,
+        droppedCompleteRecords: true,
         prefix: ownRetainedString(boundedUtf8Prefix(complete, REJECTED_LINE_PREFIX_MAX_BYTES))
       })
       return
@@ -133,19 +132,25 @@ export function createIncrementalNdjsonFramer(
     }
     const suffixBytes = Buffer.byteLength(suffix, 'utf8')
     if (suffixBytes > maxPendingInputBytes) {
+      pendingInput = null
+      discardingPausedSuffix = true
       onRejected({
         kind: 'line-too-long',
         maxLineBytes: maxPendingInputBytes,
         observedBytes: suffixBytes,
         prefix: ownRetainedString(boundedUtf8Prefix(suffix, REJECTED_LINE_PREFIX_MAX_BYTES))
       })
-      pendingInput = null
-      discardingOversizedLine = true
       return
     }
     pendingInput = detachFrom(suffix, source)
   }
 
+  const retainPausedInput = (input: string, source = input): void => {
+    const newlineIndex = input.lastIndexOf('\n')
+    queuePausedCompleteInput(newlineIndex === -1 ? '' : input.slice(0, newlineIndex + 1), source)
+    retainPendingSuffix(newlineIndex === -1 ? input : input.slice(newlineIndex + 1), source)
+  }
+
   const process = (input: string): void => {
     let cursor = 0
     while (cursor < input.length) {
@@ -193,7 +198,11 @@ export function createIncrementalNdjsonFramer(
             try {
               parsed = JSON.parse(line)
             } catch (error) {
-              onRejected({ kind: 'invalid-json', line, error: errorFrom(error) })
+              onRejected({
+                kind: 'invalid-json',
+                line,
+                error: error instanceof Error ? error : new Error(String(error))
+              })
               continue
             }
             onRecord(parsed, line)
@@ -202,12 +211,7 @@ export function createIncrementalNdjsonFramer(
       }
 
       if (options.shouldPause?.() && cursor < input.length) {
-        const remainder = input.slice(cursor)
-        const newlineIndex = remainder.lastIndexOf('\n')
-        const complete = newlineIndex === -1 ? '' : remainder.slice(0, newlineIndex + 1)
-        const suffix = newlineIndex === -1 ? remainder : remainder.slice(newlineIndex + 1)
-        queuePausedCompleteInput(complete, input)
-        retainPendingSuffix(suffix, input)
+        retainPausedInput(input.slice(cursor), input)
         return
       }
     }
@@ -215,36 +219,32 @@ export function createIncrementalNdjsonFramer(
 
   return {
     feed(chunk): void {
-      if (chunk.length === 0) {
-        return
-      }
       if (pausedCompleteInputBytes > 0 && !options.shouldPause?.()) {
         const queued = pausedCompleteInput
         pausedCompleteInput = []
         pausedCompleteInputBytes = 0
         process(queued.join(''))
       }
-      if (pendingInput !== null || pausedCompleteInputBytes > 0) {
-        // Complete records are retained as a separate bounded queue. The pending
-        // input limit applies only to the final, actually incomplete record.
-        if (pendingInput === null) {
-          if (options.shouldPause?.()) {
-            const newlineIndex = chunk.lastIndexOf('\n')
-            const complete = newlineIndex === -1 ? '' : chunk.slice(0, newlineIndex + 1)
-            const suffix = newlineIndex === -1 ? chunk : chunk.slice(newlineIndex + 1)
-            queuePausedCompleteInput(complete, chunk)
-            retainPendingSuffix(suffix, chunk)
-            return
-          }
-          process(chunk)
+      const endedPausedSuffix = discardingPausedSuffix
+      // The dropped tail follows the queued records; only new bytes can finish it.
+      if (discardingPausedSuffix) {
+        const newlineIndex = chunk.indexOf('\n')
+        if (newlineIndex === -1) {
           return
         }
-        const combined = pendingInput + chunk
-        const newlineIndex = combined.lastIndexOf('\n')
-        const complete = newlineIndex === -1 ? '' : combined.slice(0, newlineIndex + 1)
-        const suffix = newlineIndex === -1 ? combined : combined.slice(newlineIndex + 1)
-        queuePausedCompleteInput(complete, combined)
-        retainPendingSuffix(suffix, combined)
+        discardingPausedSuffix = false
+        chunk = chunk.slice(newlineIndex + 1)
+      }
+      if (chunk.length === 0) {
+        return
+      }
+
+      if (pendingInput !== null) {
+        retainPausedInput(pendingInput + chunk)
+        return
+      }
+      if ((pausedCompleteInputBytes > 0 || endedPausedSuffix) && options.shouldPause?.()) {
+        retainPausedInput(chunk)
         return
       }
       process(chunk)
@@ -275,7 +275,10 @@ export function createIncrementalNdjsonFramer(
       }
     },
     takeTail(): string {
-      const tail = discardingOversizedLine ? '' : `${lineSegments.join('')}${pendingInput ?? ''}`
+      const tail =
+        discardingOversizedLine || discardingPausedSuffix
+          ? ''
+          : `${lineSegments.join('')}${pendingInput ?? ''}`
       clearLine()
       pendingInput = null
       return tail
@@ -283,6 +286,7 @@ export function createIncrementalNdjsonFramer(
     reset(): void {
       clearLine()
       discardingOversizedLine = false
+      discardingPausedSuffix = false
       pendingInput = null
       pausedCompleteInput = []
       pausedCompleteInputBytes = 0
@@ -300,14 +304,9 @@ export function encodeNdjson(msg: unknown, maxLineBytes = NDJSON_MAX_LINE_BYTES)
   return `${line}\n`
 }
 
-export type NdjsonParser = {
-  feed(chunk: string): void
-  reset(): void
-}
+export type NdjsonParser = Pick<IncrementalNdjsonFramer, 'feed' | 'reset'>
 
-export type NdjsonParserOptions = {
-  maxLineBytes?: number
-}
+export type NdjsonParserOptions = Pick<IncrementalNdjsonFramerOptions, 'maxLineBytes'>
 
 export function createNdjsonParser(
   onMessage: (msg: unknown) => void,
diff --git a/tests/tools/pi-provider-runtime-smoke.mjs b/tests/tools/pi-provider-runtime-smoke.mjs
index d095cb72446..04309165098 100644
--- a/tests/tools/pi-provider-runtime-smoke.mjs
+++ b/tests/tools/pi-provider-runtime-smoke.mjs
@@ -2,6 +2,8 @@ import assert from 'node:assert/strict'
 import { once } from 'node:events'
 import { mkdtemp, mkdir, writeFile, rm } from 'node:fs/promises'
 import { createServer } from 'node:http'
+import { spawn } from 'node:child_process'
+import { createInterface } from 'node:readline'
 import { createRequire } from 'node:module'
 import { tmpdir } from 'node:os'
 import { join, resolve } from 'node:path'
@@ -111,6 +113,92 @@ try {
   assert.equal(results[1].code, 0, results[1].stderr)
   assert.match(results[1].stdout, /fixture-generated-commit/)
   assert.equal(results[1].requests, 1)
+  // The Pi catch-up's contract: entries keep stable ids, and `since` returns what was appended after
+  // one of them, or refuses an id the session doesn't have.
+  const sessionFile = join(scratch, 'catch-up.jsonl')
+  const rpc = spawn(process.execPath, [piCli, '--mode', 'rpc', '--session', sessionFile], {
+    cwd: scratch,
+    env: {
+      PATH: process.env.PATH,
+      SystemRoot: process.env.SystemRoot,
+      WINDIR: process.env.WINDIR,
+      HOME: scratch,
+      USERPROFILE: scratch,
+      ORCA_BACKGROUND_LAUNCH: '1',
+      PI_CODING_AGENT_DIR: dir
+    },
+    stdio: ['pipe', 'pipe', 'inherit']
+  })
+  const replies = new Map()
+  let settled
+  const settledOnce = new Promise((resolve) => (settled = resolve))
+  let fail
+  const failed = new Promise((_, reject) => (fail = reject))
+  const deadline = setTimeout(
+    () => fail(new Error('Pi RPC smoke timed out after 30 seconds')),
+    30000
+  )
+  rpc.on('error', fail)
+  rpc.on('close', (code, signal) =>
+    fail(new Error(`Pi RPC closed (code ${code}, signal ${signal})`))
+  )
+  rpc.stdin.on('error', fail)
+  rpc.stdout.on('error', fail)
+  const lines = createInterface({ input: rpc.stdout })
+  lines.on('line', (line) => {
+    try {
+      const record = JSON.parse(line)
+      if (record.type === 'response' && replies.has(record.id)) {
+        const reply = replies.get(record.id)
+        replies.delete(record.id)
+        reply(record, line)
+      } else if (record.type === 'agent_settled' || record.type === 'agent_end') {
+        settled()
+      }
+    } catch (error) {
+      fail(error)
+    }
+  })
+  let next = 0
+  const ask = (type, params = {}, id = `smoke-${++next}`) =>
+    new Promise((resolve) => {
+      replies.set(id, (record, line) => {
+        if (id === 'orca-optional-1') {
+          assert.ok(line.startsWith('{"id":"orca-optional-1","type":"response",'))
+          assert.equal(record.id, id)
+        }
+        resolve(record)
+      })
+      rpc.stdin.write(`${JSON.stringify({ ...params, type, id })}\n`)
+    })
+  try {
+    await Promise.race([
+      failed,
+      (async () => {
+        assert.equal((await ask('prompt', { message: 'hello' })).success, true)
+        await settledOnce
+        const all = await ask('get_entries', {}, 'orca-optional-1')
+        assert.equal((await ask('get_state')).success, true)
+        assert.equal(all.success, true, all.error)
+        const ids = all.data.entries.map((entry) => entry.id)
+        assert.ok(ids.length >= 2 && ids.every((id) => typeof id === 'string' && id.length > 0))
+        assert.equal(all.data.leafId, ids.at(-1))
+        const since = await ask('get_entries', { since: ids[0] })
+        assert.deepEqual(
+          since.data.entries.map((entry) => entry.id),
+          ids.slice(1)
+        )
+        const unknown = await ask('get_entries', { since: 'not-an-entry' })
+        assert.equal(unknown.success, false)
+        assert.match(unknown.error, /Entry not found/)
+        results.push({ variant: 'rpc-get-entries', entries: ids.length })
+      })()
+    ])
+  } finally {
+    clearTimeout(deadline)
+    lines.close()
+    rpc.kill()
+  }
   console.log(
     JSON.stringify(
       {