fix(native-chat): one ordered journal writer

src/main/claude/claude-background-task-row-journal.test.ts

@@ -111,25 +111,19 @@ describe('Claude background task row journal', () => {
     expect(appendAndPublish).toHaveBeenCalledTimes(2)
   })
 
-  it('keeps fallback append coalescing separate from its ordered publication', () => {
-    const calls: { operation: 'append' | 'publish'; coalescingKey?: string }[] = []
+  it('appends the row, then publishes it', () => {
+    const calls: ('append' | 'publish')[] = []
     const task = row()
     const sink: StructuredAgentSessionEventSink = {
       appendItem: vi.fn(),
       appendTombstone: vi.fn(),
       publish: vi.fn(),
-      tryAppendResolvedItem: vi.fn((_identity, _body, _resolve, options) => {
-        calls.push({
-          operation: 'append',
-          ...(options?.coalescingKey ? { coalescingKey: options.coalescingKey } : {})
-        })
+      tryAppendResolvedItem: vi.fn(() => {
+        calls.push('append')
         return { accepted: true as const }
       }),
-      tryPublish: vi.fn((options) => {
-        calls.push({
-          operation: 'publish',
-          ...(options?.coalescingKey ? { coalescingKey: options.coalescingKey } : {})
-        })
+      tryPublish: vi.fn(() => {
+        calls.push('publish')
         return { accepted: true as const }
       })
     }
@@ -143,13 +137,7 @@ describe('Claude background task row journal', () => {
         () => AGENT_JOURNAL_THREAD_SCOPE
       )
     ).toEqual({ accepted: true })
-    expect(calls).toEqual([
-      {
-        operation: 'append',
-        coalescingKey: JSON.stringify(['claude-background-task', 'task-1', 'tool-1'])
-      },
-      { operation: 'publish' }
-    ])
+    expect(calls).toEqual(['append', 'publish'])
     expect(task.lastSerialized).not.toBeNull()
   })
 

src/main/claude/claude-background-task-row-journal.ts

@@ -156,12 +156,7 @@ export function writeClaudeBackgroundTaskRow(
   }
   beforeAppend?.()
   const identity = claudeBackgroundTaskIdentity(id, row.generation)
-  // Generation is translator-local and resets when a provider stream is
-  // recreated. Keep unresolved writes from distinct provider runs queued side
-  // by using the provider's parent tool identity as the coalescing discriminator.
-  const coalescingKey = JSON.stringify(['claude-background-task', id, row.toolUseId ?? null])
   const appendOptions = {
-    coalescingKey,
     turnScope: turnScope(),
     ...(lifecycle ? { lifecycle: true } : {})
   }

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

@@ -209,7 +209,7 @@ export class ClaudeOpenTurn {
       { lifecycle: item.body, ...(contextUsage ? { contextUsage } : {}) },
       { publish: false, options: item.options }
     )
-    // Preserve first-work evidence when completion arrives before the journal drains.
+    // Keyed apart, so this never replaces the start's publication while it still waits to run.
     this.deps.sink.publish({ coalescingKey: item.publishCoalescingKey })
   }
 }

src/main/claude/claude-structured-journal-translation-background-tasks.test.ts

@@ -15,6 +15,7 @@ import {
 import type { AgentSessionJournal } from '../native-chat/agent-session-journal/journal-store'
 import { createClaudeJournalTranslator } from './claude-structured-journal-translation'
 import { blockOf } from './claude-background-task-row-test-support'
+import { withJournalQueueMembers } from '../native-chat/agent-session-wire/structured-agent-session-journal-double-test-support'
 import { testEventSinkLogging } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support'
 
 // The frames below are the ones the reported session actually carried: two real
@@ -144,7 +145,7 @@ function persistedTarget(
 ): StructuredAgentSessionEventTarget {
   const journal =
     // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: this test double implements the journal methods exercised by the deferred sink.
-    {
+    withJournalQueueMembers({
       appendItem: async (identity: AgentJournalItemIdentity, body: AgentJournalItemBody) => {
         persisted.set(agentJournalItemKey(identity), body)
         return { cursor: { epoch: 'test', sequence: persisted.size }, itemId: '', revision: 1 }
@@ -163,7 +164,7 @@ function persistedTarget(
       }) satisfies StructuredAgentSessionLinkageJournal['visitItemsWithLinkage'],
       itemBody: (itemId: string) => persisted.get(itemId) ?? null,
       epoch: 'test'
-    } as unknown as AgentSessionJournal
+    }) as unknown as AgentSessionJournal
   return { journal, fence: 1, publish: vi.fn() }
 }
 

src/main/claude/claude-structured-prompt-ownership.ts

@@ -87,8 +87,8 @@ export async function cancelClaudeStructuredTurn(input: {
     )
   }
   // Judge against the published journal, because that is the only turn a client could have been
-  // shown — but only while it HAS an answer. The journal drains through a serialized async queue,
-  // so a null read means the row has not landed yet, not that nothing is running; falling back to
+  // shown — but only while it HAS an answer. A null read can mean the row never landed (a sink not
+  // yet bound, or refusing under backpressure), not that nothing is running; falling back to
   // the in-memory turn there keeps Stop from being gated on bookkeeping. No live turn either way
   // means nothing has published an identity this request can contradict.
   const ownsRequestedTurn = (): boolean => {

src/main/claude/claude-structured-session-reading-control.test.ts

@@ -20,6 +20,7 @@ import {
   identityFor,
   PROVIDER_SESSION_ID
 } from './claude-structured-session-test-support'
+import { withJournalQueueMembers } from '../native-chat/agent-session-wire/structured-agent-session-journal-double-test-support'
 import { testEventSinkLogging } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support'
 
 function controlledSink(): {
@@ -49,7 +50,7 @@ function persistedTarget(
 ): StructuredAgentSessionEventTarget {
   const journal =
     // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: This test double implements the journal methods exercised by the deferred sink.
-    {
+    withJournalQueueMembers({
       appendItem: async (identity: AgentJournalItemIdentity, body: AgentJournalItemBody) => {
         persisted.set(agentJournalItemKey(identity), body)
         return { cursor: { epoch: 'test', sequence: persisted.size }, itemId: '', revision: 1 }
@@ -68,7 +69,7 @@ function persistedTarget(
       }) satisfies StructuredAgentSessionLinkageJournal['visitItemsWithLinkage'],
       itemBody: (itemId: string) => persisted.get(itemId) ?? null,
       epoch: 'test'
-    } as unknown as AgentSessionJournal
+    }) as unknown as AgentSessionJournal
   return { journal, fence: 1, publish: vi.fn() }
 }
 

src/main/claude/claude-subagent-group-row.ts

@@ -92,13 +92,12 @@ export function writeClaudeSubagentGroupRow(
   group: RosterGroup
 ): void {
   const agents = [...group.entries.values()].map((tracked) => tracked.entry)
-  const options = { coalescingKey: `claude-subagents:${group.groupId}`, turnScope: group.turnScope }
   if (agents.length === 0) {
     // The row's last child turned out not to be a subagent. An empty roster is
     // not a roster of nothing, so the row goes rather than reading "Ran 0".
     if (group.lastSerialized !== null) {
       group.lastSerialized = null
-      sink.appendTombstone(group.identity, options)
+      sink.appendTombstone(group.identity)
       sink.publish()
     }
     return
@@ -110,8 +109,6 @@ export function writeClaudeSubagentGroupRow(
     return
   }
   group.lastSerialized = serialized
-  sink.appendItem(group.identity, body, options)
-  // Publish keeps the sink's own coalescing slot: sharing the row's key makes
-  // each queued publish evict the append it was meant to flush.
+  sink.appendItem(group.identity, body, { turnScope: group.turnScope })
   sink.publish()
 }

src/main/claude/claude-subagent-roster.test.ts

@@ -14,6 +14,7 @@ import {
   type StructuredAgentSessionEventSink
 } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
 import { ClaudeSubagentRoster } from './claude-subagent-roster'
+import { withJournalQueueMembers } from '../native-chat/agent-session-wire/structured-agent-session-journal-double-test-support'
 import { testEventSinkLogging } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support'
 
 const TURN_1 = 'claude-session:turn-1'
@@ -456,13 +457,14 @@ describe('ClaudeSubagentRoster — through the real sink queue', () => {
   it('lands every revision, not just the one that was already in flight', async () => {
     const appended: AgentJournalItemBody[] = []
     let published = 0
-    const journal = {
+    // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a double for the journal members this path calls; the helper adds the in-order ones.
+    const journal = withJournalQueueMembers({
       appendItem: async (_identity: AgentJournalItemIdentity, body: AgentJournalItemBody) => {
         appended.push(body)
         return { cursor: { epoch: 'e', sequence: appended.length } }
       },
       appendTombstone: async () => ({ epoch: 'e', sequence: 0 })
-    } as unknown as AgentSessionJournal
+    }) as unknown as AgentSessionJournal
     const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging())
     deferred.bind({
       journal,
@@ -477,8 +479,7 @@ describe('ClaudeSubagentRoster — through the real sink queue', () => {
       currentTurnScope: () => AGENT_JOURNAL_THREAD_SCOPE
     })
 
-    // The first append is in flight while the rest are submitted, so a publish
-    // sharing the row's coalescing key would evict them.
+    // The first append is in flight while the rest are submitted; each must still land.
     roster.observeSystemFrame(started({ task_id: 'task-1', description: 'One' }))
     roster.observeSystemFrame(started({ task_id: 'task-2', description: 'Two' }))
     roster.observeSystemFrame(

src/main/codex/codex-structured-item-streams.ts

@@ -1,4 +1,3 @@
-import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key'
 import {
   AGENT_JOURNAL_THREAD_SCOPE,
   type AgentJournalRowAttribution
@@ -118,10 +117,8 @@ export function createCodexStructuredItemStreams(
     if (!translated.body) {
       return true
     }
-    return appendCodexItemAndPublish(deps.sink, state.identity, translated.body, {
-      coalescingKey: `checkpoint:${agentJournalItemKey(state.identity)}`,
-      ...attributionOf(key)
-    }).accepted
+    return appendCodexItemAndPublish(deps.sink, state.identity, translated.body, attributionOf(key))
+      .accepted
   }
 
   const persist = (key: string, text: string, force: boolean): boolean => {

src/main/codex/codex-structured-journal-generic-frames.ts

@@ -167,7 +167,6 @@ export class CodexJournalGenericFrames {
       }
       // A summary across evicted turns names no producer and belongs to no turn.
       const options = {
-        coalescingKey: `provider-frame-suppressed:codex:${bucket}`,
         ...(summary.producer
           ? this.deps.attributionFor(summary.producer.threadId, summary.producer.turnId)
           : { turnScope: AGENT_JOURNAL_THREAD_SCOPE })

src/main/codex/codex-structured-journal-goal-resume.test.ts

@@ -12,6 +12,7 @@ import {
 } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
 import { CodexJournalGoals } from './codex-structured-journal-goals'
 import { MAX_CODEX_GOAL_THREADS } from './codex-structured-journal-limits'
+import { withJournalQueueMembers } from '../native-chat/agent-session-wire/structured-agent-session-journal-double-test-support'
 import { testEventSinkLogging } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support'
 
 const THREAD = '01a08cc2-f96e-76d0-bb74-88b9bc0b03fc'
@@ -49,7 +50,7 @@ function goalJournal(
     ...options
   })
   // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a fake journal exposing only the members the goal translator and deferred sink call.
-  const journal = {
+  const journal = withJournalQueueMembers({
     get epoch() {
       return `epoch-${epochNumber}`
     },
@@ -81,7 +82,7 @@ function goalJournal(
         visit(item.itemId, item.sequence, item.body)
       }
     }
-  } as unknown as StructuredAgentSessionEventTarget['journal']
+  }) as unknown as StructuredAgentSessionEventTarget['journal']
   const target = {
     journal,
     fence: 1,

src/main/codex/codex-structured-journal-translation-settlement.test.ts

@@ -25,6 +25,7 @@ import {
   CODEX_USER_INPUT_METHOD
 } from './codex-structured-prompt-replies'
 import type { CodexStructuredSessionEvent } from './codex-structured-session-adapter'
+import { withJournalQueueMembers } from '../native-chat/agent-session-wire/structured-agent-session-journal-double-test-support'
 import { testEventSinkLogging } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support'
 
 const SESSION_ID = 'session-1'
@@ -105,7 +106,8 @@ function deferredTarget(
 ): StructuredAgentSessionEventTarget {
   return {
     fence: 7,
-    journal: {
+    // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a double for the journal members this path calls; the helper adds the in-order ones.
+    journal: withJournalQueueMembers({
       appendItem: vi.fn(async (_identity: AgentJournalItemIdentity, body: AgentJournalItemBody) => {
         log.push(body)
         return { cursor: { epoch: 'e', sequence: log.length } }
@@ -121,7 +123,7 @@ function deferredTarget(
           return { epoch: 'e', sequence: log.length }
         }
       )
-    } as unknown as StructuredAgentSessionEventTarget['journal'],
+    }) as unknown as StructuredAgentSessionEventTarget['journal'],
     publish: vi.fn(() => {
       publishes.push('publish')
     })

src/main/codex/codex-structured-journal-translation-turns.ts

@@ -123,7 +123,7 @@ export function publishCodexTurnLifecycle(input: {
   } else {
     input.sink.appendItem(identity, body, appendOptions)
   }
-  // Preserve first-work evidence when completion arrives before the journal drains.
+  // Keyed apart, so a completion's publication never replaces a start one still waiting to run.
   const publishOptions = {
     lifecycle: true,
     ...(input.state === 'running'

src/main/codex/codex-structured-journal-translation.test.ts

@@ -18,6 +18,7 @@ import {
   CODEX_USER_INPUT_METHOD
 } from './codex-structured-prompt-replies'
 import type { CodexStructuredSessionEvent } from './codex-structured-session-adapter'
+import { withJournalQueueMembers } from '../native-chat/agent-session-wire/structured-agent-session-journal-double-test-support'
 import { testEventSinkLogging } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support'
 
 const SESSION_ID = 'session-1'
@@ -103,7 +104,8 @@ function deferredTarget(
 ): StructuredAgentSessionEventTarget {
   return {
     fence: 7,
-    journal: {
+    // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a double for the journal members this path calls; the helper adds the in-order ones.
+    journal: withJournalQueueMembers({
       appendItem: vi.fn(async (_identity: AgentJournalItemIdentity, body: AgentJournalItemBody) => {
         log.push(body)
         return { cursor: { epoch: 'e', sequence: log.length } }
@@ -119,7 +121,7 @@ function deferredTarget(
           return { epoch: 'e', sequence: log.length }
         }
       )
-    } as unknown as StructuredAgentSessionEventTarget['journal'],
+    }) as unknown as StructuredAgentSessionEventTarget['journal'],
     publish: vi.fn(() => {
       publishes.push('publish')
     })