Admit one provider event's journal writes as one queued operation, decided when it runs

src/main/claude/claude-context-facts.ts

@@ -25,7 +25,10 @@ import type { ClaudeOpenTurn } from './claude-open-turn'
 import { claudeRecord, claudeText } from './claude-structured-item-translation'
 import type { ClaudeTurnEnd } from './claude-turn-lifecycle-item'
 import { isRootClaudeFrame } from './claude-turn-opening'
-import { writeClaudeTurnRow, type ClaudeTurnRowTarget } from './claude-turn-row-revision'
+import {
+  writeAgentJournalTurnRow,
+  type AgentJournalTurnRowTarget
+} from '../native-chat/agent-session-timeline/agent-journal-turn-row-revision'
 
 const CONVERSATION_FRAME_TYPES = new Set(['assistant', 'user', 'stream_event'])
 
@@ -188,7 +191,7 @@ export class ClaudeContextFacts {
     this.windowHint = report.windowTokens
     const contextUsage: AgentSessionContextUsage =
       part === 'report' ? { used: { kind: 'report', ...report }, window } : { window }
-    writeClaudeTurnRow(
+    writeAgentJournalTurnRow(
       this.sink,
       target === null ? { newest: true } : { identity: target },
       { contextUsage },
@@ -211,9 +214,9 @@ export class ClaudeContextFacts {
     windowIfNoneHeld?: AgentSessionContextWindow
   ): void {
     const identity = this.turn.identity
-    const target: ClaudeTurnRowTarget = identity ? { identity } : { newest: true }
+    const target: AgentJournalTurnRowTarget = identity ? { identity } : { newest: true }
     // A context fact often lands with no later frame to publish it, so it publishes itself.
-    writeClaudeTurnRow(
+    writeAgentJournalTurnRow(
       this.sink,
       target,
       { contextUsage, ...(windowIfNoneHeld ? { windowIfNoneHeld } : {}) },

src/main/claude/claude-open-turn.ts

@@ -18,7 +18,7 @@ import {
   type ClaudeCurrentTurn,
   type ClaudeTurnEnd
 } from './claude-turn-lifecycle-item'
-import { writeClaudeTurnRow } from './claude-turn-row-revision'
+import { writeAgentJournalTurnRow } from '../native-chat/agent-session-timeline/agent-journal-turn-row-revision'
 import type { ClaudeCommandTurn } from './claude-command-turn'
 import { createClaudeTurnOpener, type ClaudeTurnSource } from './claude-turn-opening'
 
@@ -198,7 +198,7 @@ export class ClaudeOpenTurn {
     contextUsage?: AgentSessionContextUsage
   ): void {
     const item = claudeTurnLifecycleItem(turn, end)
-    writeClaudeTurnRow(
+    writeAgentJournalTurnRow(
       this.deps.sink,
       { identity: item.identity },
       { lifecycle: item.body, ...(contextUsage ? { contextUsage } : {}) },

src/main/native-chat/agent-session-journal/journal-lifecycle-batch-appender.ts

@@ -1,7 +1,15 @@
+import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key'
 import type { AgentJournalCursor } from '../../../shared/agent-session-journal-types'
 import type { JournalReducerState } from './journal-reducer'
-import { journalLifecycleBatchRowBuilder } from './journal-row-builders'
-import type { JournalLifecycleBatchInput } from './journal-store-contracts'
+import { partitionJournalLifecycleMutations } from './journal-lifecycle-batch-partition'
+import {
+  journalLifecycleBatchRowBuilder,
+  type JournalLifecycleMutationInput
+} from './journal-row-builders'
+import type {
+  JournalLifecycleBatchInput,
+  JournalResolvedLifecycleBatchInput
+} from './journal-store-contracts'
 import type { JournalRow } from './journal-row-schema'
 import { journalQueuedRejectionRowBuilders } from './journal-pending-submission-recovery'
 
@@ -70,6 +78,35 @@ export class JournalLifecycleBatchAppender {
       })
   }
 
+  /** The rows a resolved settlement writes, planned at its own turn in the queue: its mutations,
+   *  chosen then, in as many consecutive rows as they need, minus any already applied. */
+  planResolved(
+    input: JournalResolvedLifecycleBatchInput
+  ): ((seq: number, ts: number) => JournalRow)[] {
+    const mutations = input.resolve()
+    // Every chunk is built before any commits, so a second chunk naming the same item would
+    // reuse the first chunk's revision.
+    this.assertDistinctItems(mutations)
+    return partitionJournalLifecycleMutations(input.settlementId, mutations)
+      .filter((chunk) => !this.wasApplied(chunk.settlementId))
+      .map((chunk) =>
+        journalLifecycleBatchRowBuilder(this.deps.state, chunk.settlementId, chunk.mutations, input)
+      )
+  }
+
+  private assertDistinctItems(mutations: readonly JournalLifecycleMutationInput[]): void {
+    const { aliases } = this.deps.state()
+    const seen = new Set<string>()
+    for (const mutation of mutations) {
+      const itemId = agentJournalItemKey(mutation.identity)
+      const resolved = aliases.get(itemId) ?? itemId
+      if (seen.has(resolved)) {
+        throw new Error('journal_resolved_lifecycle_batch_names_item_twice')
+      }
+      seen.add(resolved)
+    }
+  }
+
   private wasApplied(settlementId: string): boolean {
     return this.deps.state().appliedSettlementIds.has(settlementId)
   }

src/main/native-chat/agent-session-journal/journal-row-writer.ts

@@ -76,33 +76,36 @@ export class JournalRowWriter {
   enqueueRows(
     plan: () => readonly ((seq: number, ts: number) => JournalRow)[]
   ): Promise<JournalRow[]> {
-    return this.deps.serialize(() => {
-      assertJournalWritable(this.deps.readOnly(), this.deps.sessionId)
-      const first = this.deps.nextSequence()
-      const ts = this.deps.now()
-      const rows = plan().map((build, index) => build(first + index, ts))
-      if (rows.length === 0) {
-        return rows
-      }
-      for (const row of rows) {
-        assertJournalFence(row.fence, this.deps.highestFence())
-      }
-      try {
-        this.deps.database().transaction((db) => {
-          for (const row of rows) {
-            insertJournalRow(db, this.deps.sessionId, row)
-            this.runBookkeeping(db, row)
-          }
-        })
-      } catch (error) {
-        this.deps.rolledBack?.()
-        throw error
-      }
-      for (const row of rows) {
-        this.deps.commit(row)
-      }
+    return this.deps.serialize(() => this.writeRows(plan))
+  }
+
+  /** `enqueueRows`' write, for a caller already running at its own turn in the queue. */
+  writeRows(plan: () => readonly ((seq: number, ts: number) => JournalRow)[]): JournalRow[] {
+    assertJournalWritable(this.deps.readOnly(), this.deps.sessionId)
+    const first = this.deps.nextSequence()
+    const ts = this.deps.now()
+    const rows = plan().map((build, index) => build(first + index, ts))
+    if (rows.length === 0) {
       return rows
-    })
+    }
+    for (const row of rows) {
+      assertJournalFence(row.fence, this.deps.highestFence())
+    }
+    try {
+      this.deps.database().transaction((db) => {
+        for (const row of rows) {
+          insertJournalRow(db, this.deps.sessionId, row)
+          this.runBookkeeping(db, row)
+        }
+      })
+    } catch (error) {
+      this.deps.rolledBack?.()
+      throw error
+    }
+    for (const row of rows) {
+      this.deps.commit(row)
+    }
+    return rows
   }
 
   /** Assign the next sequence, make the row durable, and fold it through the SAME reducer

src/main/native-chat/agent-session-journal/journal-step-writer.ts

@@ -0,0 +1,58 @@
+// Several writes run as ONE turn in the chat's write queue, one after another.
+//
+// Each step is planned from the fold with every step before it landed, and commits in its own
+// transaction. The steps share one queue body, so no other write lands between them, and the
+// first step that throws ends the run: the steps before it stay written and none after it runs.
+
+import type {
+  JournalItemAppendOptions,
+  JournalResolvedLifecycleBatchInput
+} from './journal-store-contracts'
+import type { JournalResolvedItem } from './journal-item-appender'
+import type { JournalLifecycleBatchAppender } from './journal-lifecycle-batch-appender'
+import type { JournalReducerState } from './journal-reducer'
+import { journalItemRowBuilder } from './journal-row-builders'
+import type { JournalRow } from './journal-row-schema'
+import type { JournalRowWriter } from './journal-row-writer'
+import type { JournalWriteBody } from './journal-write-queue'
+
+export type JournalStep =
+  | {
+      kind: 'item'
+      /** Read at the step's turn; null writes nothing. */
+      resolve: () => JournalResolvedItem | null
+      options: JournalItemAppendOptions
+    }
+  | { kind: 'settlement'; batch: JournalResolvedLifecycleBatchInput }
+
+export class JournalStepWriter {
+  constructor(
+    private readonly deps: {
+      serialize: <T>(run: JournalWriteBody<T>) => Promise<T>
+      state: () => JournalReducerState
+      writeRows: JournalRowWriter['writeRows']
+      planSettlement: JournalLifecycleBatchAppender['planResolved']
+    }
+  ) {}
+
+  /** Whether each step wrote; rejects with the error of the step that threw. */
+  append(steps: readonly JournalStep[]): Promise<boolean[]> {
+    return this.deps.serialize(() => {
+      const wrote: boolean[] = []
+      for (const step of steps) {
+        wrote.push(this.deps.writeRows(() => this.plan(step)).length > 0)
+      }
+      return wrote
+    })
+  }
+
+  private plan(step: JournalStep): ((seq: number, ts: number) => JournalRow)[] {
+    if (step.kind === 'settlement') {
+      return this.deps.planSettlement(step.batch)
+    }
+    const resolved = step.resolve()
+    return resolved === null
+      ? []
+      : [journalItemRowBuilder(this.deps.state, resolved.identity, resolved.body, step.options)]
+  }
+}

src/main/native-chat/agent-session-journal/journal-store-collaborators.ts

@@ -17,6 +17,7 @@ import { JournalStopMarks } from './journal-stop-marks'
 import { journalQueuePauseRestatement } from './queued-message-pause'
 import type { JournalReducerState } from './journal-reducer'
 import { JournalRowWriter } from './journal-row-writer'
+import { JournalStepWriter } from './journal-step-writer'
 import { restoreJournalStore } from './journal-store-restore'
 import type { JournalRow } from './journal-row-schema'
 import type { AgentSessionJournal } from './journal-store'
@@ -52,6 +53,7 @@ export type JournalStoreCollaborators = {
   epochController: JournalEpochController
   itemAppender: JournalItemAppender
   lifecycleBatchAppender: JournalLifecycleBatchAppender
+  stepWriter: JournalStepWriter
   queuedMessages: JournalQueuedMessages
   stopMarks: JournalStopMarks
   /** Restores the store's state from disk. Owned here because it needs the same
@@ -100,6 +102,12 @@ export function createJournalStoreCollaborators(host: JournalStoreHost): Journal
     inTransaction: (db, row) => queuedMessages.onRowInTransaction(db, row),
     rolledBack: () => queuedMessages.invalidate()
   })
+  const lifecycleBatchAppender = new JournalLifecycleBatchAppender({
+    state: host.state,
+    cursor: host.cursor,
+    enqueue: host.enqueue,
+    enqueueRows: (plan) => rowWriter.enqueueRows(plan)
+  })
   return {
     epochController,
     queuedMessages,
@@ -115,11 +123,12 @@ export function createJournalStoreCollaborators(host: JournalStoreHost): Journal
       state: host.state,
       enqueue: host.enqueue
     }),
-    lifecycleBatchAppender: new JournalLifecycleBatchAppender({
+    lifecycleBatchAppender,
+    stepWriter: new JournalStepWriter({
+      serialize: host.serialize,
       state: host.state,
-      cursor: host.cursor,
-      enqueue: host.enqueue,
-      enqueueRows: (plan) => rowWriter.enqueueRows(plan)
+      writeRows: (plan) => rowWriter.writeRows(plan),
+      planSettlement: (batch) => lifecycleBatchAppender.planResolved(batch)
     })
   }
 }

src/main/native-chat/agent-session-journal/journal-store-contracts.ts

@@ -80,6 +80,14 @@ export type JournalLifecycleBatchInput = {
   rejectsQueued?: AgentJournalDispatchRejection
 }
 
+export type JournalResolvedLifecycleBatchInput = Omit<
+  JournalLifecycleBatchInput,
+  'mutations' | 'rejectsQueued'
+> & {
+  /** Read from the fold with every earlier write landed; may return none. */
+  resolve: () => readonly JournalLifecycleMutationInput[]
+}
+
 export type JournalSubmissionInput = {
   clientMessageId: string
   payloadFingerprint: string

src/main/native-chat/agent-session-journal/journal-store.ts

@@ -8,6 +8,7 @@ import type {
   AgentJournalCursor,
   AgentJournalItemBody,
   AgentJournalItemIdentity,
+  AgentJournalRenderItem,
   AgentJournalSnapshot,
   AgentJournalSubmission,
   AgentJournalThreadGoal,
@@ -67,8 +68,9 @@ import type { JournalOperationReceipt, JournalRowWriter } from './journal-row-wr
 import type { JournalEpochController } from './journal-epoch-controller'
 import { JournalWriteQueue } from './journal-write-queue'
 import { createJournalStoreCollaborators } from './journal-store-collaborators'
-import type { JournalItemAppender, JournalResolvedItem } from './journal-item-appender'
+import type { JournalItemAppender } from './journal-item-appender'
 import type { JournalLifecycleBatchAppender } from './journal-lifecycle-batch-appender'
+import type { JournalStepWriter } from './journal-step-writer'
 import type { JournalStopMarks } from './journal-stop-marks'
 
 export { AgentSessionJournalError } from './journal-write-guards'
@@ -87,6 +89,7 @@ export class AgentSessionJournal {
   private readonly epochController: JournalEpochController
   private readonly itemAppender: JournalItemAppender
   private readonly lifecycleBatchAppender: JournalLifecycleBatchAppender
+  private readonly stepWriter: JournalStepWriter
   private readonly restore: () => Promise<void>
   /** Draft rows queued while the agent works; never reducer input or owed work. */
   readonly queuedMessages: JournalQueuedMessages
@@ -129,6 +132,7 @@ export class AgentSessionJournal {
     this.epochController = collaborators.epochController
     this.itemAppender = collaborators.itemAppender
     this.lifecycleBatchAppender = collaborators.lifecycleBatchAppender
+    this.stepWriter = collaborators.stepWriter
     this.queuedMessages = collaborators.queuedMessages
     this.stopMarks = collaborators.stopMarks
     this.restore = collaborators.restore
@@ -204,6 +208,9 @@ export class AgentSessionJournal {
   itemBody = (itemId: string): AgentJournalItemBody | null =>
     this.state.items.get(itemId)?.body ?? null
 
+  /** One reduced item with its attribution, for a writer that needs the turn a row joined. */
+  item = (itemId: string): AgentJournalRenderItem | null => this.state.items.get(itemId) ?? null
+
   /** Visits reduced items with the producer that wrote each, for a producer re-deriving what an
    *  earlier run of this session left. */
   visitItemsWithLinkage = (visit: JournalItemLinkageVisitor): void => {
@@ -276,12 +283,8 @@ export class AgentSessionJournal {
   }
 
   /** An upsert whose row is chosen from the fold at its own turn in the queue; null writes nothing. */
-  appendResolvedItem(
-    resolve: () => JournalResolvedItem | null,
-    options: JournalItemAppendOptions
-  ): Promise<JournalAppendResult | null> {
-    return this.itemAppender.appendResolved(resolve, options)
-  }
+  appendResolvedItem: JournalItemAppender['appendResolved'] = (resolve, options) =>
+    this.itemAppender.appendResolved(resolve, options)
 
   appendTombstone(
     identity: AgentJournalItemIdentity,
@@ -307,6 +310,9 @@ export class AgentSessionJournal {
     return this.lifecycleBatchAppender.append(input)
   }
 
+  /** Several writes as one turn in the queue; see `JournalStepWriter`. */
+  appendSteps: JournalStepWriter['append'] = (steps) => this.stepWriter.append(steps)
+
   /**
    * Write-ahead submission row. It is durable before the caller dispatches
    * anything, and it doubles as the optimistic user bubble so an accepted echo