Add a shared timeline assembler for structured agent chats (not wired yet)

src/main/native-chat/agent-session-journal/journal-subagent-liveness.ts

@@ -52,8 +52,8 @@ export function staleSubagentRosterRevisions(
 ): JournalSubagentLivenessRevision[] {
   const revisions: JournalSubagentLivenessRevision[] = []
   for (const item of items) {
-    const body = item.body
-    if (body.kind !== 'message' || !body.blocks.some(hasStaleLiveWork)) {
+    const body = lostLiveWorkJournalBody(item.body)
+    if (!body) {
       continue
     }
     // A key that will not parse cannot be re-addressed, and appending under a
@@ -62,11 +62,20 @@ export function staleSubagentRosterRevisions(
     if (!identity || agentJournalItemKey(identity) !== item.itemId) {
       continue
     }
-    revisions.push({ identity, body: { ...body, blocks: settleBlocks(body.blocks) } })
+    revisions.push({ identity, body })
   }
   return revisions
 }
 
+/** The row once the host lost the session running its live work: every working child and
+ *  in-flight background task `unverifiable`, plain-text twins restated. Null when none is live. */
+export function lostLiveWorkJournalBody(body: AgentJournalItemBody): AgentJournalItemBody | null {
+  if (body.kind !== 'message' || !body.blocks.some(hasStaleLiveWork)) {
+    return null
+  }
+  return { ...body, blocks: settleBlocks(body.blocks) }
+}
+
 function hasStaleLiveWork(block: NativeChatBlock): boolean {
   return hasWorkingChild(block) || hasLiveBackgroundTask(block)
 }

src/main/native-chat/agent-session-journal/journal-terminal-settlement.ts

@@ -1,5 +1,6 @@
 import type { AgentJournalItemBody } from '../../../shared/agent-session-journal-types'
 import { isRunningAgentJournalTurn } from '../../../shared/agent-session-turn-record'
+import { cancelledJournalPromptBody } from './journal-prompt-body-bounds'
 
 /** True while an item is still awaiting the row that settles it, so a sink can
  *  treat that row as lifecycle-critical rather than sheddable under pressure. */
@@ -12,3 +13,15 @@ export function requiresTerminalSettlement(body: AgentJournalItemBody): boolean
   }
   return isRunningAgentJournalTurn(body)
 }
+
+/** The row that settles an item no one will finish: a running tool call fails, a pending prompt
+ *  is cancelled. Null for an item that needs none. Turn rows are each writer's own to end. */
+export function terminalAgentJournalBody(body: AgentJournalItemBody): AgentJournalItemBody | null {
+  if (body.kind === 'tool-call') {
+    return body.state === 'running' ? { ...body, state: 'failed' } : null
+  }
+  if (body.kind === 'approval' || body.kind === 'question') {
+    return body.resolution.state === 'pending' ? cancelledJournalPromptBody(body) : null
+  }
+  return null
+}

src/main/native-chat/agent-session-timeline/provider-timeline-assembler-attribution.test.ts

@@ -0,0 +1,223 @@
+// Where a request, a send or a background task belongs is decided from the turn the event names
+// and the journal as it stands, not from what an earlier event left in memory.
+
+import { afterEach, describe, expect, it } from 'vitest'
+import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key'
+import { agentJournalTurnBody } from '../../../shared/agent-session-turn-record'
+import {
+  backgroundTask,
+  backgroundTaskState,
+  closeProviderTimelineRigs,
+  openProviderTimelineRig,
+  openUnboundProviderTimelineAssembler,
+  pendingApproval,
+  providerItemId,
+  providerTurnItemId,
+  type ProviderTimelineRig
+} from './provider-timeline-assembler-test-support'
+
+afterEach(closeProviderTimelineRigs)
+
+const answered = {
+  ...pendingApproval,
+  resolution: {
+    state: 'resolved' as const,
+    selectedOptionId: 'allow',
+    resolvedBy: 'phone',
+    resolvedAt: 1_500
+  }
+}
+
+function identityOf(itemId: string) {
+  const identity = parseAgentJournalItemKey(itemId)
+  if (!identity) {
+    throw new Error(`${itemId} did not parse`)
+  }
+  return identity
+}
+
+/** `p1` opened in `t1` and answered by a client, `t1` over, `t2` running. */
+async function answeredInAnEarlierTurn(): Promise<ProviderTimelineRig> {
+  const rig = await openProviderTimelineRig()
+  rig.assembler.apply({ type: 'turn.open', turn: 't1', at: 1_000 })
+  rig.assembler.apply({ type: 'request.open', request: 'p1', body: pendingApproval })
+  await rig.journal.appendItem(identityOf(providerItemId('request', 'p1')), answered, {
+    fence: 1,
+    turnScope: { kind: 'turn', turnItemId: providerTurnItemId('t1') }
+  })
+  rig.assembler.apply({ type: 'turn.end', turn: 't1', at: 2_000, state: 'completed' })
+  rig.assembler.apply({ type: 'turn.open', turn: 't2', at: 3_000 })
+  await rig.rows()
+  return rig
+}
+
+const second = providerItemId('request', 'p1', { incarnation: 2 })
+
+describe('a request key reused in another turn', () => {
+  it('opens a new prompt in the live turn it names after a withdrawn one', async () => {
+    const rig = await openProviderTimelineRig()
+    rig.assembler.apply({ type: 'turn.open', turn: 't1', at: 1_000 })
+    rig.assembler.apply({ type: 'request.open', request: 'p1', body: pendingApproval })
+    rig.assembler.apply({ type: 'request.withdrawn', request: 'p1' })
+    rig.assembler.apply({ type: 'turn.end', turn: 't1', at: 2_000, state: 'completed' })
+    rig.assembler.apply({ type: 'turn.open', turn: 't2', at: 3_000 })
+    await rig.rows()
+    const opened = rig.assembler.apply({
+      type: 'request.open',
+      request: 'p1',
+      body: pendingApproval,
+      join: { turn: 't2' }
+    })
+    expect(opened.dropped).toBeUndefined()
+    expect(await rig.row(second)).toMatchObject({
+      body: { resolution: { state: 'pending' } },
+      turnScope: { kind: 'turn', turnItemId: providerTurnItemId('t2') }
+    })
+    expect((await rig.row(providerItemId('request', 'p1')))?.body).toMatchObject({
+      resolution: { state: 'cancelled' }
+    })
+  })
+
+  it('opens a new prompt in the open turn after an answered one, leaving the answer', async () => {
+    const rig = await answeredInAnEarlierTurn()
+    expect(
+      rig.assembler.apply({ type: 'request.open', request: 'p1', body: pendingApproval }).dropped
+    ).toBeUndefined()
+    expect((await rig.row(second))?.body).toMatchObject({ resolution: { state: 'pending' } })
+    expect((await rig.row(providerItemId('request', 'p1')))?.body).toMatchObject({
+      resolution: { state: 'resolved', selectedOptionId: 'allow' }
+    })
+  })
+
+  it('opens a new row for a request id the previous process used, after a restart', async () => {
+    const rig = await openProviderTimelineRig()
+    rig.assembler.apply({ type: 'turn.open', turn: 't1', at: 1_000 })
+    rig.assembler.apply({ type: 'request.open', request: '0', body: pendingApproval })
+    await rig.rows()
+    // JSON-RPC ids restart with the provider process: the new child asks under `0` again.
+    const restarted = await rig.restart()
+    restarted.apply({ type: 'turn.open', turn: 't2', at: 3_000 })
+    expect(
+      restarted.apply({ type: 'request.open', request: '0', body: pendingApproval }).dropped
+    ).toBeUndefined()
+    // The sweep cancelled the old process's prompt; the new one is its own row, still pending.
+    expect((await rig.row(providerItemId('request', '0')))?.body).toMatchObject({
+      resolution: { state: 'cancelled' }
+    })
+    expect(await rig.row(providerItemId('request', '0', { generation: 'gen-2' }))).toMatchObject({
+      body: { resolution: { state: 'pending' } },
+      turnScope: { kind: 'turn', turnItemId: providerTurnItemId('t2') }
+    })
+  })
+
+  it('asks nothing in a turn another writer ended while the open was queued', async () => {
+    const rig = await answeredInAnEarlierTurn()
+    const { assembler, bind } = openUnboundProviderTimelineAssembler(rig.journal)
+    const join = { turn: 't2' }
+    assembler.apply({ type: 'request.open', request: 'p1', body: pendingApproval, join })
+    const running = await rig.turn('t2')
+    if (!running) {
+      throw new Error('t2 is running')
+    }
+    await rig.journal.appendItem(
+      identityOf(providerTurnItemId('t2')),
+      agentJournalTurnBody({ ...running, state: 'interrupted', completedAt: 4_000 }),
+      { fence: 1, turnScope: { kind: 'thread' } }
+    )
+    await bind()
+    expect(
+      await rig.row(providerItemId('request', 'p1', { generation: 'gen-unbound' }))
+    ).toBeUndefined()
+  })
+})
+
+describe('a send that names its turn', () => {
+  it('opens the turn it names when that turn opens later', async () => {
+    const rig = await openProviderTimelineRig()
+    rig.assembler.apply({
+      type: 'input.accepted',
+      clientMessageId: 'send1',
+      requestedAt: 900,
+      join: { turn: 't1' }
+    })
+    rig.assembler.apply({ type: 'turn.open', turn: 't1', at: 1_000 })
+    expect(await rig.turn('t1')).toMatchObject({ userItemId: 'orca:send1', requestedAt: 900 })
+  })
+
+  it('never replaces an opener another writer gave the running turn', async () => {
+    const rig = await openProviderTimelineRig()
+    rig.assembler.apply({ type: 'turn.open', turn: 't1', at: 1_000 })
+    const running = await rig.turn('t1')
+    if (!running) {
+      throw new Error('t1 is running')
+    }
+    await rig.journal.appendItem(
+      identityOf(providerTurnItemId('t1')),
+      agentJournalTurnBody({ ...running, userItemId: 'orca:first', requestedAt: 800 }),
+      { fence: 1, turnScope: { kind: 'thread' } }
+    )
+    rig.assembler.apply({ type: 'input.accepted', clientMessageId: 'later', requestedAt: 900 })
+    expect(await rig.turn('t1')).toMatchObject({ userItemId: 'orca:first', requestedAt: 800 })
+  })
+
+  it('waits for its own turn rather than the next one to open', async () => {
+    const rig = await openProviderTimelineRig()
+    for (const [send, turn] of [
+      ['send-b', 't2'],
+      ['send-a', 't1']
+    ] as const) {
+      rig.assembler.apply({
+        type: 'input.accepted',
+        clientMessageId: send,
+        requestedAt: 900,
+        join: { turn }
+      })
+    }
+    rig.assembler.apply({ type: 'turn.open', turn: 't1', at: 1_000 })
+    rig.assembler.apply({ type: 'turn.end', at: 2_000, state: 'completed' })
+    rig.assembler.apply({ type: 'turn.open', turn: 't3', at: 3_000 })
+    rig.assembler.apply({ type: 'turn.end', at: 4_000, state: 'completed' })
+    rig.assembler.apply({ type: 'turn.open', turn: 't2', at: 5_000 })
+    expect((await rig.turn('t1'))?.userItemId).toBe('orca:send-a')
+    // A turn no send named opens as the provider's own.
+    expect((await rig.turn('t3'))?.userItemId).toBe(providerTurnItemId('t3'))
+    expect((await rig.turn('t2'))?.userItemId).toBe('orca:send-b')
+  })
+})
+
+describe('a background task', () => {
+  it('outlives its turn, and settles on its own update', async () => {
+    const rig = await openProviderTimelineRig()
+    rig.assembler.apply({ type: 'turn.open', turn: 't1', at: 1_000 })
+    rig.assembler.apply({ type: 'item.open', item: 'bg', body: backgroundTask('bg', 'working') })
+    rig.assembler.apply({ type: 'turn.end', turn: 't1', at: 2_000, state: 'completed' })
+    expect(await backgroundTaskState(rig, 'bg')).toBe('working')
+    expect(
+      rig.assembler.apply({ type: 'item.close', item: 'bg', body: backgroundTask('bg', 'done') })
+        .dropped
+    ).toBeUndefined()
+    expect(await backgroundTaskState(rig, 'bg')).toBe('done')
+    // A straggler progress report never re-lights it.
+    expect(
+      rig.assembler.apply({
+        type: 'item.update',
+        item: 'bg',
+        body: backgroundTask('bg', 'working')
+      }).dropped
+    ).toBe('item-settled')
+    expect(await backgroundTaskState(rig, 'bg')).toBe('done')
+  })
+
+  it('is left unverifiable, not exited, when the session ends with it in flight', async () => {
+    const rig = await openProviderTimelineRig()
+    rig.assembler.apply({ type: 'turn.open', turn: 't1', at: 1_000 })
+    rig.assembler.apply({ type: 'item.open', item: 'bg', body: backgroundTask('bg', 'working') })
+    rig.assembler.apply({ type: 'item.open', item: 'ran', body: backgroundTask('ran', 'working') })
+    rig.assembler.apply({ type: 'item.close', item: 'ran', body: backgroundTask('ran', 'done') })
+    rig.assembler.apply({ type: 'session.ended', verdict: { state: 'unverifiable' } })
+    expect((await rig.row(providerItemId('item', 'bg')))?.body).toEqual(
+      backgroundTask('bg', 'unverifiable')
+    )
+    expect(await backgroundTaskState(rig, 'ran')).toBe('done')
+  })
+})

src/main/native-chat/agent-session-timeline/provider-timeline-assembler-earlier-turn-end.test.ts

@@ -0,0 +1,62 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import {
+  closeProviderTimelineRigs,
+  messageText,
+  openProviderTimelineRig,
+  providerItemId,
+  providerTurnId,
+  providerTurnItemId,
+  runningTool
+} from './provider-timeline-assembler-test-support'
+
+afterEach(closeProviderTimelineRigs)
+
+describe("an earlier turn's late end leaves the open turn alone", () => {
+  it("keeps the open turn's activity line", async () => {
+    const rig = await openProviderTimelineRig()
+    const activity: unknown[] = []
+    const assembler = rig.assemble({
+      sink: { ...rig.sink, setActivity: (each) => activity.push(each) }
+    })
+    assembler.apply({ type: 'turn.open', turn: 't1', at: 1_000 })
+    assembler.apply({ type: 'turn.end', turn: 't1', at: 1_500, state: 'completed' })
+    assembler.apply({ type: 'turn.open', turn: 't2', at: 2_000 })
+    assembler.apply({ type: 'activity', text: 'Thinking' })
+    activity.length = 0
+    expect(
+      assembler.apply({ type: 'turn.end', turn: 't1', at: 2_100, state: 'completed' })
+    ).toEqual({ admission: { accepted: true } })
+    expect(activity).toEqual([])
+    expect(assembler.openTurnId).toBe(providerTurnId('t2'))
+  })
+
+  it("keeps the open turn's anonymous reply one message", async () => {
+    const rig = await openProviderTimelineRig()
+    rig.assembler.apply({ type: 'turn.open', turn: 't1', at: 1_000 })
+    rig.assembler.apply({ type: 'turn.end', turn: 't1', at: 1_500, state: 'completed' })
+    rig.assembler.apply({ type: 'turn.open', turn: 't2', at: 2_000 })
+    const delta = { type: 'text.delta', item: { stream: 'reply' }, channel: 'assistant' } as const
+    rig.assembler.apply({ ...delta, text: 'Hello' })
+    rig.assembler.apply({ type: 'turn.end', turn: 't1', at: 2_100, state: 'completed' })
+    rig.assembler.apply({ ...delta, text: ' world' })
+    const texts = (await rig.rows()).flatMap((row) =>
+      row.body.kind === 'message' ? [[row.turnScope, messageText(row.body)]] : []
+    )
+    expect(texts).toEqual([[{ kind: 'turn', turnItemId: providerTurnItemId('t2') }, 'Hello world']])
+  })
+
+  it("leaves the open turn's running work to that turn's own end", async () => {
+    const rig = await openProviderTimelineRig()
+    rig.assembler.apply({ type: 'turn.open', turn: 't1', at: 1_000 })
+    rig.assembler.apply({ type: 'turn.open', turn: 't2', at: 2_000 })
+    rig.assembler.apply({ type: 'item.open', item: 'call', body: runningTool('read') })
+    rig.assembler.apply({ type: 'turn.end', turn: 't1', at: 2_100, state: 'completed' })
+    expect((await rig.row(providerItemId('item', 'call')))?.body).toMatchObject({
+      state: 'running'
+    })
+    rig.assembler.apply({ type: 'turn.end', turn: 't2', at: 2_200, state: 'interrupted' })
+    expect((await rig.row(providerItemId('item', 'call')))?.body).toMatchObject({
+      state: 'failed'
+    })
+  })
+})