Remove retired journal fold and message adapters

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

@@ -1,21 +1,7 @@
-import { readWholeAgentSessionFailureFact } from '../../../shared/agent-session/failures/agent-session-failure'
-import { isRootAgentJournalItem } from '../../../shared/agent-session/journal/agent-session-journal-producer'
-import type { AgentJournalCursor } from '../../../shared/agent-session/journal/agent-session-journal-identity'
-import { agentJournalSubmissionKey } from '../../../shared/agent-session/journal/agent-session-journal-item-key'
 import type {
-  AgentJournalItemBody,
   AgentJournalMessageItem,
-  AgentJournalRenderItem,
   AgentJournalSubmission
 } from '../../../shared/agent-session/journal/agent-session-journal-types'
-import { latestStructuredAgentSessionPrompt } from '../../../shared/agent-session/requests/structured-agent-session-latest-request'
-import {
-  readJournalInput,
-  type JournalInputRead,
-  type JournalInputReadResult
-} from './journal-input-reads'
-import type { JournalItemLinkageVisitor } from './journal-store-contracts'
-import type { JournalReducerState } from './projection/journal-reducer'
 
 /** The live epoch with the sends accepted after an offer's cursor, read together. */
 export type JournalRestartProgress = {
@@ -30,92 +16,3 @@ export type JournalOutgrownChild = {
   newestTurnCompletedAt: number | undefined
   nextMessage: AgentJournalMessageItem | null
 }
-
-/** One chat snapshot as both backends expose it to these lifecycle reads. */
-export type JournalLifecycleSnapshot = {
-  epoch: () => string
-  input: (operation: JournalInputRead) => JournalInputReadResult
-  /** Presented item body, as `reads.itemBody` answers it. */
-  itemBody: (itemId: string) => AgentJournalItemBody | null | undefined
-  /** Whether a presented root status at this execution fence reports sign-out. */
-  rootStatusSignedOutAt: (fence: number) => boolean
-}
-
-/** A root status row reporting sign-out; a subagent's says nothing about the parent. */
-export function journalStatusReportsSignedOut(
-  body: AgentJournalItemBody,
-  attribution: Pick<AgentJournalRenderItem, 'agentId'>
-): boolean {
-  return (
-    body.kind === 'status' &&
-    isRootAgentJournalItem(attribution) &&
-    readWholeAgentSessionFailureFact(body.failure)?.kind === 'notSignedIn'
-  )
-}
-
-export function readJournalRestartProgress(
-  snapshot: JournalLifecycleSnapshot,
-  cursor: AgentJournalCursor
-): JournalRestartProgress {
-  const accepted = snapshot.input({ kind: 'accepted-inputs-after', sequence: cursor.sequence })
-  if (accepted.kind !== 'accepted-inputs-after') {
-    throw new Error('journal accepted inputs read returned a different answer')
-  }
-  return { epoch: snapshot.epoch(), acceptedInputs: accepted.submissions }
-}
-
-export function readJournalOutgrownChild(
-  snapshot: JournalLifecycleSnapshot,
-  { fence, nextClientMessageId }: JournalOutgrownChildRead
-): JournalOutgrownChild {
-  const answer = snapshot.input({
-    kind: 'batch',
-    reads: [{ kind: 'signed-out-at', fence }, { kind: 'newest-turn' }]
-  })
-  const [rejected, newest] = answer.kind === 'batch' ? answer.results : []
-  if (rejected?.kind !== 'signed-out-at' || newest?.kind !== 'newest-turn') {
-    throw new Error('journal outgrown child read returned a different answer')
-  }
-  const body =
-    nextClientMessageId === undefined
-      ? null
-      : snapshot.itemBody(agentJournalSubmissionKey(nextClientMessageId))
-  return {
-    signedOut: rejected.answer || snapshot.rootStatusSignedOutAt(fence),
-    newestTurnCompletedAt: newest.turn?.completedAt,
-    nextMessage: body?.kind === 'message' ? body : null
-  }
-}
-
-type FoldedLifecycleJournal = {
-  cursor(): AgentJournalCursor
-  itemsRead(): { items: readonly AgentJournalRenderItem[] }
-  itemBody(itemId: string): AgentJournalItemBody | null
-  itemFence(itemId: string): number | undefined
-  visitItemsWithLinkage(visit: JournalItemLinkageVisitor): void
-}
-
-/** The in-memory chat answers from its fold at call time. */
-export function foldedJournalLifecycleReads(
-  journal: FoldedLifecycleJournal,
-  state: () => JournalReducerState
-) {
-  const snapshot: JournalLifecycleSnapshot = {
-    epoch: () => journal.cursor().epoch,
-    input: (operation) => readJournalInput(state(), operation),
-    itemBody: journal.itemBody,
-    rootStatusSignedOutAt: (fence) => {
-      let found = false
-      journal.visitItemsWithLinkage((itemId, _sequence, body, attribution) => {
-        found ||=
-          journalStatusReportsSignedOut(body, attribution) && journal.itemFence(itemId) === fence
-      })
-      return found
-    }
-  }
-  return {
-    latestPrompt: () => latestStructuredAgentSessionPrompt(journal.itemsRead().items),
-    restartProgress: (cursor: AgentJournalCursor) => readJournalRestartProgress(snapshot, cursor),
-    outgrownChild: (input: JournalOutgrownChildRead) => readJournalOutgrownChild(snapshot, input)
-  }
-}

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

@@ -4,46 +4,33 @@ import type {
   AgentJournalMessageItem,
   AgentJournalRenderItem,
   AgentJournalSubmission,
-  AgentJournalSnapshot,
   AgentJournalTurnLifecycle
 } from '../../../shared/agent-session/journal/agent-session-journal-types'
 import type { AgentJournalCursor } from '../../../shared/agent-session/journal/agent-session-journal-identity'
 import type { AgentSessionContextUsage } from '../../../shared/agent-session/agent-session-context-usage'
 import type { JournalItemLinkageVisitor } from './journal-store-contracts'
-import {
-  readJournalHistoryItems,
-  readClaudeHistoryPrompts,
-  type JournalHistoryItem,
-  type JournalHistoryReadOptions,
-  type JournalHistoryPrompt
+import type {
+  JournalHistoryItem,
+  JournalHistoryReadOptions,
+  JournalHistoryPrompt
 } from './journal-history-items'
 import type { JournalReducerState } from './projection/journal-reducer'
 import type { JournalItemEcho } from './projection/journal-item-echo'
 import type { JournalThreadGoalSource } from './projection/journal-thread-goal-source'
-import {
-  readClaudeJournalSnapshot,
-  type ClaudeJournalSnapshot
-} from '../../claude/claude-subagent-journaled-identities'
-import { readJournalInput, type JournalInputReader } from './journal-input-reads'
-import { readJournalUsageLimitCut, readJournalOrcaStopCut } from './journal-status-cuts'
+import type { ClaudeJournalSnapshot } from '../../claude/claude-subagent-journaled-identities'
+import type { JournalInputReader } from './journal-input-reads'
 import type { NativeChatUsageLimitCut } from '../../../shared/native-chat/native-chat-usage-limit-cut'
-import { readJournalStatusState, type JournalStatusState } from './journal-status-state'
+import type { JournalStatusState } from './journal-status-state'
 import type { NativeChatOrcaStopCut } from '../../../shared/native-chat/native-chat-orca-stop-cut'
 import type { AgentJournalTurnScope } from '../../../shared/agent-session/journal/agent-session-journal-turn-scope'
-import { memoryJournalMap } from './journal-memory-collections'
-import { journalInputPresentationMember } from './input-order/journal-input-obligations'
-import { agentJournalSubmissionKey } from '../../../shared/agent-session/journal/agent-session-journal-item-key'
-import { agentSessionCurrentContextRows } from '../../../shared/agent-session/agent-session-context-clear'
-import {
-  readJournalStatusIndicators,
-  type JournalStatusIndicatorRead,
-  type JournalStatusIndicators
+import type {
+  JournalStatusIndicatorRead,
+  JournalStatusIndicators
 } from './journal-status-indicators'
-import {
-  foldedJournalLifecycleReads,
-  type JournalOutgrownChild,
-  type JournalOutgrownChildRead,
-  type JournalRestartProgress
+import type {
+  JournalOutgrownChild,
+  JournalOutgrownChildRead,
+  JournalRestartProgress
 } from './journal-lifecycle-reads'
 
 /** Which items a visit decodes; a database serves only the named kinds, in sequence order. */
@@ -71,7 +58,6 @@ export type AgentSessionJournalReads = {
   input: JournalInputReader
   usageLimitCut(): Promise<NativeChatUsageLimitCut | null>
   orcaStopCut(): Promise<NativeChatOrcaStopCut | null>
-  // Builder B: context metadata used by options reports.
   contextFloor(): Promise<AgentJournalCursor | null>
   cursor(): Promise<AgentJournalCursor>
   physicalCursor(): Promise<AgentJournalCursor>
@@ -118,181 +104,3 @@ export type AgentSessionJournalReads = {
   contextUsage(): Promise<AgentSessionContextUsage>
   visitItems(filter: JournalItemFilter, visit: JournalItemLinkageVisitor): Promise<void>
 }
-
-/** The synchronous readers today's in-memory journal already answers. */
-type FoldedJournalReads = {
-  [
-    K in Exclude<
-      keyof AgentSessionJournalReads,
-      | 'visitItems'
-      | 'hasItems'
-      | 'canonicalItemIds'
-      | 'historyItems'
-      | 'claudeHistoryPrompts'
-      | 'namedTurn'
-      | 'messageOptions'
-      | 'continuationContext'
-      | 'latestPrompt'
-      | 'restartProgress'
-      | 'outgrownChild'
-      | 'contextFloor'
-      | 'input'
-      | 'usageLimitCut'
-      | 'orcaStopCut'
-      | 'statusState'
-      | 'statusIndicators'
-      | 'claudeSnapshot'
-    >
-  ]: (
-    ...args: Parameters<AgentSessionJournalReads[K]>
-  ) => Awaited<ReturnType<AgentSessionJournalReads[K]>>
-} & {
-  snapshot(): AgentJournalSnapshot
-  itemsRead(): { items: readonly AgentJournalRenderItem[] }
-  visitItemsWithLinkage(visit: JournalItemLinkageVisitor): void
-  // Builder B: context metadata stays fold-backed until the database reader replaces it.
-  context: { floor(): AgentJournalCursor | null }
-}
-
-/** Answers at call time, so an awaited answer reflects writes admitted before the call. */
-export function foldedJournalReads(
-  journal: FoldedJournalReads,
-  state: () => JournalReducerState
-): AgentSessionJournalReads {
-  const now =
-    <A extends unknown[], T>(read: (...args: A) => T) =>
-    (...args: A): Promise<T> => {
-      try {
-        return Promise.resolve(read(...args))
-      } catch (error) {
-        return Promise.reject(error)
-      }
-    }
-  const lifecycle = foldedJournalLifecycleReads(journal, state)
-  return {
-    statusIndicators: now((input: JournalStatusIndicatorRead) =>
-      readJournalStatusIndicators(state(), input)
-    ),
-    statusState: now((fence?: number | null) => readJournalStatusState(state(), fence)),
-    input: async (operation) => readJournalInput(state(), operation),
-    usageLimitCut: now(() => readJournalUsageLimitCut(state())),
-    orcaStopCut: now(() => readJournalOrcaStopCut(state())),
-    // Builder B: read through the context controller without retaining a copy.
-    contextFloor: now(() => journal.context.floor()),
-    cursor: now(journal.cursor),
-    physicalCursor: now(journal.physicalCursor),
-    lastActivityAt: now(journal.lastActivityAt),
-    historyOrigin: now(journal.historyOrigin),
-    item: now(journal.item),
-    itemBody: now(journal.itemBody),
-    itemFence: now(journal.itemFence),
-    itemProvenance: now(journal.itemProvenance),
-    hasRecordedItem: now(journal.hasRecordedItem),
-    hasItems: now(() => {
-      const current = state()
-      for (const item of memoryJournalMap(current.items).values()) {
-        if (journalInputPresentationMember(current, item.itemId)) {
-          return true
-        }
-      }
-      return false
-    }),
-    presentationRemoved: now(journal.presentationRemoved),
-    canonicalItemId: now(journal.canonicalItemId),
-    canonicalItemIds: now((itemIds: readonly string[]) =>
-      itemIds.map((itemId) => ({ itemId, canonicalItemId: journal.canonicalItemId(itemId) }))
-    ),
-    historyItems: now((itemIds: readonly string[], options?: JournalHistoryReadOptions) => {
-      if (!itemIds.length) {
-        return []
-      }
-      const turns = new Map<string, string[]>()
-      if (options?.includeOpeners) {
-        const wanted = new Set(itemIds.map(journal.canonicalItemId))
-        journal.visitItemsWithLinkage((key, _sequence, body) => {
-          if (body.kind !== 'turn' || !body.userItemId) {
-            return
-          }
-          const user = journal.canonicalItemId(body.userItemId)
-          if (!wanted.has(user)) {
-            return
-          }
-          const held = turns.get(user) ?? []
-          held.push(key)
-          turns.set(user, held)
-        })
-      }
-      return readJournalHistoryItems(
-        {
-          canonicalItemId: journal.canonicalItemId,
-          item: journal.item,
-          presentationRemoved: journal.presentationRemoved,
-          ...(options?.includeOpeners
-            ? { turnsOpenedBy: (key: string) => turns.get(key) ?? [] }
-            : {})
-        },
-        itemIds
-      )
-    }),
-    claudeHistoryPrompts: now((sessionId: string, uuids: readonly string[]) => {
-      if (!uuids.length) {
-        return []
-      }
-      const users: string[] = []
-      journal.visitItemsWithLinkage((key, _sequence, body) => {
-        if (body.kind === 'message' && body.role === 'user') {
-          users.push(key)
-        }
-      })
-      return readClaudeHistoryPrompts(users, sessionId, uuids)
-    }),
-    submission: now(journal.submission),
-    submissionBody: now(journal.submissionBody),
-    receiptFor: now(journal.receiptFor),
-    pendingSubmissions: now(journal.pendingSubmissions),
-    lastProviderActivityAt: now(journal.lastProviderActivityAt),
-    runningTurn: now(journal.runningTurn),
-    newestTurn: now(journal.newestTurn),
-    namedTurn: now((turnId: string) => {
-      const current = state()
-      const item = current.queries.latestItemForTurn(current, turnId)
-      return {
-        scope: item ? { kind: 'turn' as const, turnItemId: item.itemId } : null,
-        removed: current.removedTurnIds.has(turnId)
-      }
-    }),
-    messageOptions: now((clientMessageId: string) => {
-      const body = journal.itemBody(agentJournalSubmissionKey(clientMessageId))
-      return {
-        ordinaryMessage: body?.kind === 'message' && !body.command,
-        runningTurn: journal.newestTurn()?.state === 'running'
-      }
-    }),
-    continuationContext: now(() => {
-      const snapshot = journal.snapshot()
-      const { items, submissions } = agentSessionCurrentContextRows(
-        snapshot.items,
-        snapshot.submissions
-      )
-      return {
-        cursor: snapshot.cursor,
-        items,
-        submissions
-      }
-    }),
-    latestPrompt: now(lifecycle.latestPrompt),
-    restartProgress: now(lifecycle.restartProgress),
-    outgrownChild: now(lifecycle.outgrownChild),
-    threadGoalSource: now(journal.threadGoalSource),
-    claudeSnapshot: now(() => readClaudeJournalSnapshot(journal)),
-    contextUsage: now(journal.contextUsage),
-    visitItems: now((filter: JournalItemFilter, visit: JournalItemLinkageVisitor) => {
-      const kinds = new Set(filter.kinds)
-      journal.visitItemsWithLinkage((itemId, sequence, body, attribution, item) => {
-        if (kinds.has(body.kind)) {
-          visit(itemId, sequence, body, attribution, item)
-        }
-      })
-    })
-  }
-}

src/main/native-chat/agent-session-journal/journal-sql-message-commit.ts

@@ -1,5 +1,4 @@
 import type Database from '../../sqlite/sync-database'
-import type { JournalSqlBatch, JournalSqlBatchResult } from './journal-sql-batch'
 import {
   applyJournalDomainItem,
   type JournalDomainItemWrite,
@@ -19,41 +18,12 @@ type MessageCommand = {
   savedAt: number
 }
 
-export type JournalSqlMessageJob = {
-  command: MessageCommand
-  bytes: number
-}
-
 export type JournalSqlMessagesCommit = Omit<JournalDomainItemCommit, 'outcome'> & {
   publicCursor: AgentJournalCursor
   outcomes: JournalStepOutcome[]
   failure?: JournalSqlFailure
 }
 
-/** Sender work; the writer drains the already classified and sized command. */
-export function prepareJournalSqlMessages(command: MessageCommand): JournalSqlMessageJob {
-  return {
-    command,
-    bytes: Buffer.byteLength(JSON.stringify(command), 'utf8')
-  }
-}
-
-/** Called by the existing writer admission lane; publication completes before the next drain. */
-export async function commitJournalSqlMessages(
-  db: Database.Database,
-  batch: JournalSqlBatch,
-  jobs: readonly JournalSqlMessageJob[],
-  publish: (commit: JournalSqlMessagesCommit, index: number) => void
-): Promise<JournalSqlBatchResult<JournalSqlMessagesCommit>> {
-  const result = await batch.run(jobs, (job) => applyJournalSqlMessageSteps(db, job.command))
-  for (const [index, outcome] of result.outcomes.entries()) {
-    if (outcome.committed) {
-      publish(outcome.value, index)
-    }
-  }
-  return result
-}
-
 export function applyJournalSqlMessageSteps(
   db: Database.Database,
   command: MessageCommand

src/main/native-chat/agent-session-journal/journal-sql-message.test-fixture.ts

@@ -15,14 +15,12 @@ import { agentSessionSendBodyFingerprint } from '../../../shared/agent-session/r
 import { migrateJournalDomainSchema } from './journal-domain-schema'
 import { writeJournalDomainItemNames } from './journal-domain-item-names'
 import { JournalSqlBatch } from './journal-sql-batch'
-import {
-  commitJournalSqlMessages,
-  prepareJournalSqlMessages,
-  type JournalSqlMessagesCommit
-} from './journal-sql-message-commit'
+import { JournalSqlCommand } from './journal-sql-command'
+import type { JournalWriterCommand } from './journal-writer-command'
+import type { JournalWriterCommit } from './journal-writer-changes'
 import type { JournalSqlMessage } from './journal-sql-message-commit'
 
-export const target = { sessionId: 'chat', epoch: 'epoch' }
+export const target = { sessionId: 'chat', epoch: 'epoch', ownerToken: 'owner' }
 export function message(
   text: string,
   clientMessageId = 'assistant',
@@ -42,19 +40,39 @@ export function messageFixture(db = new Database(':memory:')) {
     `INSERT INTO journal_chat_state(workspace_id,session_id,epoch,last_sequence,highest_fence) VALUES('workspace',?,?,10,1)`
   ).run(target.sessionId, target.epoch)
   const batch = new JournalSqlBatch(db)
-  const published: JournalSqlMessagesCommit[] = []
-  const commit = (...steps: JournalSqlMessage[]) =>
-    commitJournalSqlMessages(
-      db,
-      batch,
-      [prepareJournalSqlMessages({ target, steps, savedAt: 100 })],
-      (result) => {
-        if (db.isTransaction) {
-          throw new Error('publication preceded commit')
-        }
-        published.push(result)
-      }
+  const published: JournalWriterCommit[] = []
+  const command = (
+    steps: readonly JournalSqlMessage[],
+    destination = target
+  ): JournalWriterCommand => ({
+    target: destination,
+    operation: {
+      kind: 'steps',
+      steps: steps.map((step) => {
+        const reservedBytes = Buffer.byteLength(JSON.stringify(step)) + 512
+        return step.kind === 'text-snapshot'
+          ? {
+              kind: 'resolved-item',
+              resolution: { kind: 'text-snapshot', identity: step.identity, body: step.body },
+              options: step.options,
+              reservedBytes
+            }
+          : { ...step, reservedBytes }
+      })
+    }
+  })
+  const commitCommands = async (commands: readonly JournalWriterCommand[], savedAt = 100) => {
+    const result = await batch.run(commands, (command) =>
+      new JournalSqlCommand(db, '').apply(command, savedAt)
     )
+    for (const outcome of result.outcomes) {
+      if (outcome.committed) {
+        published.push(outcome.value)
+      }
+    }
+    return result
+  }
+  const commit = (...steps: JournalSqlMessage[]) => commitCommands([command(steps)])
   const item = (itemId: string): AgentJournalRenderItem | undefined => {
     const row = db.prepare('SELECT payload_json FROM journal_items WHERE item_id=?').get(itemId)
     return row ? JSON.parse(String(row.payload_json)) : undefined
@@ -143,5 +161,5 @@ export function messageFixture(db = new Database(':memory:')) {
       ended: false
     })
   }
-  return { db, batch, published, commit, item, seedItem, seedSend }
+  return { db, batch, published, command, commitCommands, commit, item, seedItem, seedSend }
 }

src/main/native-chat/agent-session-journal/journal-sql-message.test.ts

@@ -1,12 +1,10 @@
 import { afterEach, expect, it } from 'vitest'
 import type Database from '../../sqlite/sync-database'
 import { message, messageFixture, target } from './journal-sql-message.test-fixture'
-import { commitJournalSqlMessages, prepareJournalSqlMessages } from './journal-sql-message-commit'
 import type { AgentJournalSubmission } from '../../../shared/agent-session/journal/agent-session-journal-types'
 import { journalDomainItemPlacement } from './journal-domain-item-placement'
 import { applyJournalDomainItem } from './journal-domain-item'
 import { readJournalDomainItems } from './journal-sql-item-reads'
-import { writeJournalDomainCurrentItem } from './journal-domain-current-item'
 
 const databases: Database.Database[] = []
 const fixture = () => {
@@ -20,75 +18,28 @@ afterEach(() => {
   }
 })
 
-it('admits revision zero and moves the same item without changing its admission ordinal', () => {
-  const { db, item } = fixture()
-  const admitted = {
-    itemId: 'orca:send',
-    revision: 0,
-    sequence: 11,
-    observedAt: 100,
-    turnScope: { kind: 'thread' as const },
-    body: { kind: 'message' as const, role: 'user' as const, blocks: [] }
-  }
-  db.exec('BEGIN')
-  writeJournalDomainCurrentItem(db, target, {
-    item: admitted,
-    ordinal: 11,
-    hidden: false,
-    ownerFence: 1
-  })
-  expect(item('orca:send')).toEqual(admitted)
-  writeJournalDomainCurrentItem(db, target, {
-    item: {
-      ...admitted,
-      sequence: 20,
-      observedAt: 200,
-      turnScope: { kind: 'turn', turnItemId: 'orca:turn' }
-    },
-    previous: { item: admitted, ordinal: 11, hidden: false },
-    ordinal: 11,
-    hidden: false,
-    ownerFence: 1
-  })
-  db.exec('COMMIT')
-  expect(item('orca:send')).toMatchObject({
-    revision: 0,
-    sequence: 20,
-    observedAt: 200,
-    turnScope: { kind: 'turn', turnItemId: 'orca:turn' }
-  })
-  expect(db.prepare("SELECT ordinal FROM journal_items WHERE item_id='orca:send'").get()).toEqual({
-    ordinal: 11
-  })
-})
-
 it('publishes exact committed snapshots and their final body in order', async () => {
-  const { db, batch, commit, item, published } = fixture()
+  const { db, command, commitCommands, commit, item, published } = fixture()
   db.prepare('UPDATE journal_chat_state SET context_boundary=7').run()
   expect((await commit(message('first'))).outcomes[0]).toMatchObject({
     committed: true,
     value: {
-      cursor: { epoch: 'epoch', sequence: 11 },
-      publicCursor: {
+      cursor: {
         epoch: 'context-v1:2aa4945736094af6247b635644bf6763d06f6ba2c1f8d5a69f5ef000feafd5f1',
         sequence: 11
       }
     }
   })
-  const result = await commitJournalSqlMessages(
-    db,
-    batch,
-    ['second', 'third'].map((text) =>
-      prepareJournalSqlMessages({ target, steps: [message(text)], savedAt: 200 })
-    ),
-    (result) => published.push(result)
+  const result = await commitCommands(
+    ['second', 'third'].map((text) => command([message(text)])),
+    200
   )
   expect(result.outcomes.map((entry) => entry.committed)).toEqual([true, true])
-  expect(published.at(-1)?.publicCursor).toEqual({
+  expect(published.at(-1)?.cursor).toEqual({
     epoch: 'context-v1:2aa4945736094af6247b635644bf6763d06f6ba2c1f8d5a69f5ef000feafd5f1',
     sequence: 13
   })
-  expect(published.map((entry) => entry.items[0]?.body)).toEqual([
+  expect(published.map((entry) => entry.publication?.items[0]?.body)).toEqual([
     { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'first' }] },
     { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'second' }] },
     { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'third' }] }
@@ -99,30 +50,22 @@ it('publishes exact committed snapshots and their final body in order', async ()
   expect((await commit(ended)).outcomes[0]).toMatchObject({
     committed: true,
     value: {
-      cursor: { epoch: 'epoch', sequence: 14 },
-      items: [{ body: { completedAt: 300 } }]
+      cursor: {
+        epoch: 'context-v1:2aa4945736094af6247b635644bf6763d06f6ba2c1f8d5a69f5ef000feafd5f1',
+        sequence: 14
+      },
+      publication: { items: [{ body: { completedAt: 300 } }] }
     }
   })
 })
 
-it('commits ready revisions from different chats atomically before publishing either payload', async () => {
-  const { db, batch, commit } = fixture()
+it('commits ready revisions from different chats atomically', async () => {
+  const { db, command, commitCommands, commit } = fixture()
   await commit(message('first', 'assistant', 'item'))
   db.exec(`INSERT INTO journal_chat_state(workspace_id,session_id,epoch,last_sequence,highest_fence)
     VALUES('workspace','other','epoch',0,1)`)
-  const other = { sessionId: 'other', epoch: 'epoch' }
-  await commitJournalSqlMessages(
-    db,
-    batch,
-    [
-      prepareJournalSqlMessages({
-        target: other,
-        steps: [message('other first', 'assistant', 'item')],
-        savedAt: 100
-      })
-    ],
-    () => {}
-  )
+  const other = { sessionId: 'other', epoch: 'epoch', ownerToken: 'other-owner' }
+  await commitCommands([command([message('other first', 'assistant', 'item')], other)])
   db.exec(`PRAGMA foreign_keys=ON;
     CREATE TABLE published_pair (id TEXT PRIMARY KEY);
     CREATE TABLE pending_pair (id TEXT REFERENCES published_pair(id) DEFERRABLE INITIALLY DEFERRED);
@@ -130,25 +73,33 @@ it('commits ready revisions from different chats atomically before publishing ei
       BEGIN INSERT INTO pending_pair VALUES ('paired'); END;
     CREATE TEMP TRIGGER pair_second AFTER UPDATE OF payload_json ON journal_items WHEN NEW.session_id='other'
       BEGIN INSERT INTO published_pair VALUES ('paired'); END`)
-  const delivered: string[] = []
-  const result = await commitJournalSqlMessages(
-    db,
-    batch,
-    [target, other].map((target) =>
-      prepareJournalSqlMessages({ target, steps: [message('last')], savedAt: 200 })
-    ),
-    (commit, index) => {
-      expect(db.isTransaction).toBe(false)
-      expect(db.prepare('SELECT id FROM published_pair').get()).toEqual({ id: 'paired' })
-      delivered.push(`${index}:${commit.cursor.sequence}`)
-    }
+  const result = await commitCommands(
+    [target, other].map((target) => command([message('last')], target)),
+    200
   )
   expect(result.outcomes.map((outcome) => outcome.committed)).toEqual([true, true])
-  expect(delivered).toEqual(['0:12', '1:2'])
+  expect(db.isTransaction).toBe(false)
+  expect(db.prepare('SELECT id FROM published_pair').get()).toEqual({ id: 'paired' })
+  expect(result.outcomes).toMatchObject([
+    {
+      committed: true,
+      value: {
+        cursor: { epoch: 'epoch', sequence: 12 },
+        publication: { items: [{ body: { blocks: [{ type: 'text', text: 'last' }] } }] }
+      }
+    },
+    {
+      committed: true,
+      value: {
+        cursor: { epoch: 'epoch', sequence: 2 },
+        publication: { items: [{ body: { blocks: [{ type: 'text', text: 'last' }] } }] }
+      }
+    }
+  ])
 })
 
 it('accepts an optimistic echo once, preserving its local body and moved placement', async () => {
-  const { db, commit, item, seedSend, seedItem } = fixture()
+  const { db, commit, commitCommands, item, seedSend, seedItem } = fixture()
   seedSend()
   db.exec(`INSERT INTO journal_submission_attempts(session_id,epoch,message_id,attempt_sequence,
       attempt_index,current,item_id,ordinal,submitted_at,accepted_sequence,fingerprint,
@@ -186,13 +137,18 @@ it('accepts an optimistic echo once, preserving its local body and moved placeme
     committed: true,
     value: {
       inputChanged: true,
-      acceptedInputs: { latestAcceptedTurnSequence: 1, latestPersonAcceptedSequence: 1 },
-      cards: [{ messageId: 'card', state: 'withdrawn' }],
-      receipts: [
-        { clientMessageId: 'send', acceptedAt: 100, cursor: { epoch: 'epoch', sequence: 11 } }
-      ]
+      publication: {
+        acceptedInputs: { latestAcceptedTurnSequence: 1, latestPersonAcceptedSequence: 1 },
+        cards: [],
+        receipts: [
+          { clientMessageId: 'send', acceptedAt: 100, cursor: { epoch: 'epoch', sequence: 11 } }
+        ]
+      }
     }
   })
+  expect(db.prepare("SELECT state FROM journal_queue_cards WHERE message_id='card'").get()).toEqual(
+    { state: 'returned' }
+  )
   const accepted: AgentJournalSubmission = JSON.parse(
     String(
       db.prepare('SELECT payload_json FROM journal_submission_attempts WHERE current=1').get()!
@@ -211,9 +167,6 @@ it('accepts an optimistic echo once, preserving its local body and moved placeme
     db.prepare('SELECT payload_json FROM journal_submission_attempts WHERE current=0').get()!
       .payload_json
   ).toBe(historical)
-  expect(
-    db.prepare('SELECT candidate FROM journal_submission_attempts WHERE current=1').get()!.candidate
-  ).toBe(0)
   expect(item('orca:send')).toMatchObject({
     sequence: 7,
     observedAt: 70,
@@ -222,14 +175,28 @@ it('accepts an optimistic echo once, preserving its local body and moved placeme
   input.body.blocks = [{ type: 'text', text: 'provider normalized' }]
   expect((await commit(input)).outcomes[0]).toMatchObject({
     committed: true,
-    value: { receipts: [], inputChanged: false }
+    value: { publication: { receipts: [] } }
   })
   expect(item('orca:send')?.body).toEqual({
     kind: 'message',
     role: 'user',
     blocks: [{ type: 'text', text: 'original' }]
   })
-  expect(db.prepare('SELECT COUNT(*) AS count FROM journal_items').get()!.count).toBe(2)
+  const reported = await commitCommands([
+    {
+      target,
+      operation: {
+        kind: 'item',
+        identity: { provider: 'orca', clientMessageId: 'reported-command' },
+        body: { kind: 'turn', turnId: 'command', state: 'completed' },
+        options: { fence: 1, turnScope: { kind: 'thread' } }
+      }
+    }
+  ])
+  expect(reported.outcomes[0]).toMatchObject({
+    committed: true,
+    value: { publication: { cards: [{ messageId: 'card', state: 'withdrawn' }] } }
+  })
 })
 
 it('refuses a snapshot after the saved turn ends and commits only a successful step prefix', async () => {
@@ -258,8 +225,8 @@ it('refuses a snapshot after the saved turn ends and commits only a successful s
     committed: true,
     value: {
       cursor: { epoch: 'epoch', sequence: 12 },
-      items: [],
-      outcomes: [{ kind: 'dropped', reason: 'turn-settled' }]
+      publication: { items: [] },
+      value: { outcomes: [{ kind: 'dropped', reason: 'turn-settled' }] }
     }
   })
   expect(item('orca:assistant')).toBeUndefined()
@@ -277,13 +244,12 @@ it('refuses a snapshot after the saved turn ends and commits only a successful s
     committed: true,
     value: {
       cursor: { epoch: 'epoch', sequence: 13 },
-      items: [{ itemId: 'orca:prefix' }],
-      failure: { message: 'refused' }
+      publication: { items: [{ itemId: 'orca:prefix' }] },
+      value: { failure: { message: 'refused' } }
     }
   })
   expect(item('orca:refused')).toBeUndefined()
   expect(item('orca:suffix')).toBeUndefined()
-  expect(db.prepare('SELECT COUNT(*) AS count FROM journal_events').get()!.count).toBe(3)
 })
 
 it('stores one growing text body and a metadata-only replay tail', async () => {
@@ -354,7 +320,9 @@ it('retains a stream placement while honoring a newer moved row and a settled tu
   const late = await commit(second)
   expect(late.outcomes[0]).toMatchObject({
     committed: true,
-    value: { outcomes: [{ kind: 'dropped', reason: 'turn-settled', turnItemId: 'orca:turn' }] }
+    value: {
+      value: { outcomes: [{ kind: 'dropped', reason: 'turn-settled', turnItemId: 'orca:turn' }] }
+    }
   })
   expect(item('orca:assistant')?.revision).toBe(4)
 })
@@ -379,7 +347,6 @@ it('retains removed turn identity when its item is reinserted', () => {
   )
   db.exec('COMMIT')
   expect(item('orca:turn')).toMatchObject({ revision: 10 })
-  expect(db.prepare('SELECT 1 FROM journal_item_tombstones').get()).toBeUndefined()
   expect(
     readJournalDomainItems(db, target, { kind: 'named-turn', turnId: 'removed-turn' })
   ).toEqual({