perf(native-chat): reuse narrow SQLite work reads

Read canonical current work and queue eligibility in one snapshot, and reuse validated work through the existing journal observation entry. Preserve full status publication, live authority validation, optional queue failures and final writer claims. Frozen700 matches the reviewed 27-file source manifest and measured tree 6423b1cfbb. Cold and changed-head reads meet the one-point budget; unchanged publication frames improve. Disclose report698 strict-rule misses accepted by the coordinator as noise: the approximately 4us warm in-memory hit has paired p50 ratio 1.047; one writer-ack cell (empty / 2 subscribers / changed) is +5.5% at p50, 0.259 to 0.274 ms, while completion is approximately twice as fast.
2026-11-15 10:27:33 +00:00 · 2026-11-12 10:14:19 -08:00
parent 4a4dc76107
commit b98a2544bf
27 changed files with 989 additions and 318 deletions
@@ -1,12 +1,8 @@
 import { decodePersistedAgentSessionRecord } from '../../../shared/agent-session/lifecycle/agent-session-record-stored-form'
 import { isReadableAgentSessionStoreRecord } from '../../runtime/agent-session/records/agent-session-store-row-rules'
-import type { JournalCurrentWorkAuthority } from './journal-observer-read'
+import type { JournalCurrentWorkAuthority, JournalObserverResult } from './journal-observer-read'
 import { JournalSqlStatusContext } from './journal-sql-status-context'
-import { readJournalSqlCurrentWork } from './journal-sql-work-read'
-import {
-  journalDomainFirstQueuedInput,
-  journalDomainEnteredInputs
-} from './journal-domain-input-populations'
+import { readJournalSqlCurrentWork } from './journal-sql-current-work'
 import {
  nextJournalDomainQueueCard,
  type JournalDomainQueuePick
@@ -17,47 +13,68 @@ import type { JournalSqlQueueFacts } from './journal-sql-queued-messages'
 /** Send, automatic claim and publication derive the same barriers from current SQL rows. */
 export function journalDomainInputEligibility(
  scope: JournalSqlScope,
-  authority: JournalCurrentWorkAuthority
+  authority: JournalCurrentWorkAuthority,
+  context = new JournalSqlStatusContext(scope),
+  work = readJournalSqlCurrentWork(context, authority)
 ) {
-  const row = scope.db
-    .prepare('SELECT record_json FROM agent_session_records WHERE session_id=?')
-    .get(scope.sessionId)
-  const saved = row && scope.decode<unknown>(row.record_json)
-  if (row && !isReadableAgentSessionStoreRecord(scope.sessionId, saved)) {
-    throw new Error('input admission requires a readable session record')
-  }
-  const record = isReadableAgentSessionStoreRecord(scope.sessionId, saved)
-    ? decodePersistedAgentSessionRecord(saved).record
-    : null
-  const blocked =
-    !record || record.rewind?.phase === 'prepared' || record.rewind?.phase === 'provider-succeeded'
-  const context = new JournalSqlStatusContext(scope)
-  const work = readJournalSqlCurrentWork(context, authority)
+  const blocked = journalDomainInputRecordBlocked(scope, context)
  const prompts = [...context.select({ kind: 'pending-prompts' })].flatMap(({ item }) => {
    const displayed = work.isCurrentItem(item.itemId) ? context.displayed(item.itemId) : null
    return displayed ? [displayed] : []
  })
-  const pending = journalDomainFirstQueuedInput(scope)
-  const entered = journalDomainEnteredInputs(scope)
  return {
    blocked,
    context,
    work,
    prompts,
-    pending,
-    handover: [...entered].some((send) => work.owesSend(send)),
+    pending: context.firstQueuedInput(),
+    handover: [...context.enteredSubmissions()].some((send) => work.owesSend(send)),
    activeTurnId: work.activeTurn()?.turnId
  }
 }

+function journalDomainInputRecordBlocked(scope: JournalSqlScope, context: JournalSqlStatusContext) {
+  const row =
+    context.inputHead ??
+    scope.db
+      .prepare('SELECT record_json FROM agent_session_records WHERE session_id=?')
+      .get(scope.sessionId)
+  const saved = row?.record_json == null ? null : scope.decode<unknown>(row.record_json)
+  if (row?.record_json != null && !isReadableAgentSessionStoreRecord(scope.sessionId, saved)) {
+    throw new Error('input admission requires a readable session record')
+  }
+  const record = isReadableAgentSessionStoreRecord(scope.sessionId, saved)
+    ? decodePersistedAgentSessionRecord(saved).record
+    : null
+  return (
+    !record || record.rewind?.phase === 'prepared' || record.rewind?.phase === 'provider-succeeded'
+  )
+}
+
+/** A validated work answer already owns prompt, handover and active-turn policy. */
+export function journalDomainResolvedInputEligibility(
+  context: JournalSqlStatusContext,
+  work: JournalObserverResult['work']
+) {
+  return {
+    blocked: journalDomainInputRecordBlocked(context.scope, context),
+    context,
+    prompts: work.actionablePrompts,
+    pending: context.firstQueuedInput(),
+    handover: work.owedEnteredSubmissions.length > 0,
+    activeTurnId: work.activeTurn?.turnId
+  }
+}
+
 export function selectJournalEligibleQueueCard(
  scope: JournalSqlScope,
  authority: JournalCurrentWorkAuthority,
  now: number,
  facts: JournalSqlQueueFacts,
-  allowSteer: boolean
+  allowSteer: boolean,
+  resolved?: Omit<ReturnType<typeof journalDomainInputEligibility>, 'work'>
 ): JournalDomainQueuePick & { steer: boolean; command: boolean } {
-  const state = journalDomainInputEligibility(scope, authority)
+  const state = resolved ?? journalDomainInputEligibility(scope, authority)
  const none = { kind: 'selected' as const, card: null, steer: false, command: false }
  const steer = state.activeTurnId !== undefined
  if (
@@ -70,13 +87,18 @@ export function selectJournalEligibleQueueCard(
    return none
  }
  return {
-    ...nextJournalDomainQueueCard(scope.db, scope, {
-      reopenFloor: facts.reopenFloor,
-      now,
-      steer,
-      automatic: true,
-      mail: facts.mail
-    }),
+    ...nextJournalDomainQueueCard(
+      scope.db,
+      scope,
+      {
+        reopenFloor: facts.reopenFloor,
+        now,
+        steer,
+        automatic: true,
+        mail: facts.mail
+      },
+      state.context.inputHead
+    ),
    steer,
    command: state.activeTurnId?.startsWith('compact:') === true
  }
@@ -27,14 +27,14 @@ export function* journalDomainUnsettledInputs(scope: JournalSqlScope, pendingOnl
  let after: [number, string] = [-1, '']
  for (;;) {
    const rows = scope.db
-      .prepare(`SELECT message_id,ordinal FROM journal_submission_attempts
+      .prepare(`SELECT message_id,ordinal,payload_json FROM journal_submission_attempts
      WHERE session_id=? AND epoch=? AND current=1 AND dispatch_state IN ('pending','unknown')
      ${pendingOnly ? "AND dispatch_state='pending'" : ''}
      AND (ordinal,message_id)>(?,?) ORDER BY ordinal,message_id LIMIT ?`)
      .all(...scope.prefix, ...after, PAGE)
    for (const row of rows) {
      scope.examined(row)
-      const input = journalDomainSubmission(scope, String(row.message_id))
+      const input = scope.decode<AgentJournalSubmission>(row.payload_json)
      if (input) {
        yield input
      }
@@ -5,7 +5,8 @@ import {
  createQueuePauseContext,
  type QueuePauseContext
 } from './queued-messages/queued-message-pause'
-import { readJournalSqlAcceptedInputs } from './journal-sql-accepted-inputs'
+import { JOURNAL_SQL_ACCEPTED_INPUT_COLUMNS } from './journal-sql-accepted-inputs'
+import type { SqliteRow } from '../../sqlite/sqlite-statement'
 import { readJournalDomainQueueFailure } from './journal-domain-queue-failures'
 import { invalidateJournalDomainQueue } from './journal-domain-queue-revision'

@@ -16,15 +17,19 @@ export function readJournalDomainQueuePauseContext(
  db: Database.Database,
  target: Target,
  reopenFloor: AgentJournalCursor | null,
-  messageIds: readonly string[]
+  messageIds: readonly string[],
+  snapshot?: { head: SqliteRow; clearMembers: ReadonlySet<string> }
 ): QueuePauseContext {
  const prefix = [target.sessionId, target.epoch] as const
-  const head = db
-    .prepare(`SELECT queue_latest_stop_json, queue_manual_stop_json,
+  const head =
+    snapshot?.head ??
+    db
+      .prepare(`SELECT queue_latest_stop_json, queue_manual_stop_json,
    queue_resumed_sequence, queue_reopened_sequence, queue_clear_sequence,
-    queue_clear_operation_id, queue_clear_lifted FROM journal_chat_state
+    queue_clear_operation_id, queue_clear_lifted, ${JOURNAL_SQL_ACCEPTED_INPUT_COLUMNS}
+    FROM journal_chat_state head
    WHERE session_id=? AND epoch=?`)
-    .get(...prefix)
+      .get(...prefix)
  if (!head) {
    throw new Error('journal queue pause targets a retired chat epoch')
  }
@@ -47,15 +52,20 @@ export function readJournalDomainQueuePauseContext(
          }
        })
  }
-  const clearMembers = new Set(
-    db
-      .prepare(`SELECT member.message_id FROM json_each(?) wanted
+  const clearMembers =
+    snapshot?.clearMembers ??
+    new Set(
+      db
+        .prepare(`SELECT member.message_id FROM json_each(?) wanted
    CROSS JOIN journal_queue_clear_members member WHERE member.session_id=? AND member.epoch=?
      AND member.message_id=wanted.value`)
-      .all(JSON.stringify(messageIds), ...prefix)
-      .map((row) => String(row.message_id))
-  )
-  const accepted = readJournalSqlAcceptedInputs(db, prefix)
+        .all(JSON.stringify(messageIds), ...prefix)
+        .map((row) => String(row.message_id))
+    )
+  const accepted = {
+    latestAcceptedTurnSequence: Number(head.latest_accepted ?? 0),
+    latestPersonAcceptedSequence: Number(head.latest_person_accepted ?? 0)
+  }
  return createQueuePauseContext(
    {
      epoch: target.epoch,
@@ -1,4 +1,5 @@
-import { afterEach, expect, it } from 'vitest'
+import { readJournalSqlPoint } from './journal-sql-point-read'
+import { afterEach, expect, it, vi } from 'vitest'
 import Database from '../../sqlite/sync-database'
 import type { AgentJournalMessageItem } from '../../../shared/agent-session/journal/agent-session-journal-types'
 import { agentSessionFailureWords } from '../../../shared/agent-session/failures/agent-session-failure-words'
@@ -90,6 +91,108 @@ it('answers opening and owed phase membership in one work snapshot and honors a
  db.exec(`UPDATE journal_items SET hidden=1,turn_marker=1,context_clear=0,root=1,
    sequence=99 WHERE item_id='orca:send'`)
  expect(read(1)).toEqual({ kind: 'opening-send', opening: false })
+  const sql = vi.spyOn(db, 'prepare')
+  const input = readJournalSqlPoint(db, {
+    sessionId: 'chat',
+    operation: {
+      kind: 'work',
+      input: {
+        authority,
+        operation: { kind: 'input', now: 100, facts: { reopenFloor: null }, allowSteer: false }
+      }
+    }
+  })
+  expect(input).toMatchObject({
+    kind: 'work',
+    answer: { result: { kind: 'input', state: { working: true }, queue: { card: null } } }
+  })
+  const seeks = [
+    ...new Set(
+      sql.mock.calls
+        .map(([query]) => query)
+        .filter(
+          (query) =>
+            query.includes('raw_boundary AS') ||
+            query.includes('AS work_floor') ||
+            query.includes('payload_json FROM journal_submission_attempts')
+        )
+    )
+  ]
+  sql.mockRestore()
+  const plans = seeks.map((query) =>
+    db
+      .prepare(`EXPLAIN QUERY PLAN ${query}`)
+      .all()
+      .map((row) => String(row.detail))
+      .join('\n')
+  )
+  expect(plans.join('\n')).toContain('journal_items_raw_turn_marker')
+  expect(plans.join('\n')).toContain('journal_items_turn_marker')
+  expect(plans.join('\n')).toContain('journal_items_status_pending')
+  expect(plans.join('\n')).toContain('journal_items_context_clear')
+  expect(plans.join('\n')).toContain('journal_items_raw_context_clear')
+  expect(plans.join('\n')).toContain('journal_submission_unsettled')
+  for (const plan of plans) {
+    expect(plan).not.toMatch(/SCAN (?:journal_items|item|source)\b|TEMP B-TREE/)
+  }
+  const resolved = readJournalSqlWork(context(), { authority, operation: { kind: 'state' } })
+  const resolvedQueue = () =>
+    readJournalSqlPoint(db, {
+      sessionId: 'chat',
+      operation: {
+        kind: 'queue',
+        operation: {
+          kind: 'eligible-input-card',
+          authority,
+          now: 100,
+          facts: { reopenFloor: null },
+          allowSteer: false,
+          resolved
+        }
+      }
+    })
+  expect(resolvedQueue()).toMatchObject({
+    kind: 'queue',
+    result: {
+      kind: 'eligible-input-card',
+      card: null,
+      work: { result: { kind: 'input', state: { working: true }, queue: { card: null } } }
+    }
+  })
+  db.prepare(
+    "INSERT INTO agent_session_records(session_id,record_json) VALUES ('chat','unreadable')"
+  ).run()
+  expect(
+    readJournalSqlPoint(db, {
+      sessionId: 'chat',
+      operation: {
+        kind: 'work',
+        input: {
+          authority,
+          operation: { kind: 'input', now: 100, facts: { reopenFloor: null }, allowSteer: false }
+        }
+      }
+    })
+  ).toMatchObject({
+    kind: 'work',
+    answer: {
+      result: {
+        state: { working: true },
+        queue: { failure: { message: expect.stringContaining('JSON') } }
+      }
+    }
+  })
+  expect(resolvedQueue()).toMatchObject({
+    kind: 'queue',
+    result: {
+      work: {
+        result: {
+          state: { working: true },
+          queue: { failure: { message: expect.stringContaining('JSON') } }
+        }
+      }
+    }
+  })
 })
 const cleanup: Database.Database[] = []
 afterEach(() => {
@@ -1,3 +1,4 @@
+import type { SqliteRow } from '../../sqlite/sqlite-statement'
 import type Database from '../../sqlite/sync-database'
 import { journalMailSenderKey, type JournalMailCustody } from './journal-writer-host-facts'
 import type { AgentMessageSource } from '../../../shared/agent-session/agent-session-message-source'
@@ -40,7 +41,8 @@ export type JournalDomainQueuePick =
 export function nextJournalDomainQueueCard(
  db: Database.Database,
  target: JournalDomainSubmissionTarget,
-  selection: JournalDomainQueueSelection
+  selection: JournalDomainQueueSelection,
+  pauseHead?: SqliteRow
 ): JournalDomainQueuePick {
  const { steer, automatic } = selection
  let after: { position: number; messageId: string } | undefined
@@ -52,6 +54,13 @@ export function nextJournalDomainQueueCard(
        WHERE lease.session_id=card.session_id AND lease.epoch=card.epoch
          AND lease.message_id=card.message_id AND lease.card_revision=card.revision
          AND lease.expires_at>?) AS edit_held
+      ${
+        pauseHead
+          ? `,EXISTS(SELECT 1 FROM journal_queue_clear_members member
+        WHERE member.session_id=card.session_id AND member.epoch=card.epoch
+          AND member.message_id=card.message_id) AS clear_member`
+          : ''
+      }
      FROM journal_queue_cards AS card WHERE card.session_id=? AND card.epoch=?
        AND card.state IN ('waiting','returned')
        ${after ? 'AND (card.position,card.message_id)>(?,?)' : ''}
@@ -69,7 +78,15 @@ export function nextJournalDomainQueueCard(
      db,
      target,
      selection.reopenFloor,
-      rows.map((row) => String(row.message_id))
+      rows.map((row) => String(row.message_id)),
+      pauseHead
+        ? {
+            head: pauseHead,
+            clearMembers: new Set(
+              rows.filter((row) => row.clear_member).map((row) => String(row.message_id))
+            )
+          }
+        : undefined
    )
    for (const row of rows) {
      if (automatic && row.edit_held) {
@@ -26,3 +26,12 @@ export function readJournalSqlAcceptedInputs(
    latestPersonAcceptedSequence: Number(person?.accepted_sequence ?? 0)
  }
 }
+
+export const JOURNAL_SQL_ACCEPTED_INPUT_COLUMNS = `
+  (SELECT accepted_sequence FROM journal_submission_attempts
+    WHERE session_id=head.session_id AND epoch=head.epoch AND dispatch_state='accepted'
+      AND accepted_sequence IS NOT NULL ORDER BY accepted_sequence DESC LIMIT 1) AS latest_accepted,
+  (SELECT accepted_sequence FROM journal_submission_attempts
+    WHERE session_id=head.session_id AND epoch=head.epoch AND dispatch_state='accepted'
+      AND accepted_sequence IS NOT NULL AND admitted_personal=1
+    ORDER BY accepted_sequence DESC LIMIT 1) AS latest_person_accepted`
@@ -0,0 +1,50 @@
+import {
+  StructuredAgentSessionWorkPolicy,
+  structuredAgentSessionLiveFence,
+  structuredAgentSessionOwnerUnverifiable
+} from '../agent-session-wire/observations/structured-agent-session-work-policy'
+import { journalWroteBeforeOpen } from './replay/journal-cursor'
+import type { JournalSqlStatusContext } from './journal-sql-status-context'
+import type { JournalCurrentWorkAuthority } from './journal-observer-read'
+
+export function readJournalSqlCurrentWork(
+  context: JournalSqlStatusContext,
+  authority: JournalCurrentWorkAuthority
+) {
+  const handingOver = new Set(authority.handingOver)
+  const continuing = new Set(authority.continuingUnadmitted)
+  const finalizing = new Set(authority.finalizing)
+  const held =
+    authority.child?.heldSends === undefined ? undefined : new Set(authority.child.heldSends)
+  const evidence = {
+    record: authority.record,
+    ended: authority.ended,
+    replaced: authority.replaced,
+    child:
+      authority.child === null
+        ? null
+        : authority.child === undefined
+          ? undefined
+          : { fence: authority.child.fence }
+  }
+  return new StructuredAgentSessionWorkPolicy(
+    {
+      item: context.displayed,
+      itemFence: context.itemFence,
+      runningTurn: context.runningTurn,
+      recoverableSubmissions: context.recoverableSubmissions,
+      wroteBeforeOpen: (sequence) =>
+        journalWroteBeforeOpen({ epoch: context.scope.epoch }, authority.openedThrough, sequence),
+      continuingInput: (id) =>
+        continuing.has(id) && context.submission(id)?.dispatchState === 'pending',
+      finalizingInput: (id) => finalizing.has(id),
+      handingOver: (id) => handingOver.has(id)
+    },
+    structuredAgentSessionLiveFence(evidence),
+    authority.operationalRevision,
+    authority.ended,
+    authority.replaced,
+    structuredAgentSessionOwnerUnverifiable(evidence),
+    () => held
+  )
+}
@@ -94,7 +94,15 @@ function read(
    case 'work':
      return {
        kind: 'work',
-        answer: readJournalSqlWork(new JournalSqlStatusContext(scope), operation.input)
+        answer: readJournalSqlWork(
+          new JournalSqlStatusContext(
+            scope,
+            operation.input.operation.kind === 'state' || operation.input.operation.kind === 'input'
+              ? operation.input.operation.kind
+              : undefined
+          ),
+          operation.input
+        )
      }
    case 'observer':
      return {
@@ -1,5 +1,13 @@
 import type { JournalCurrentWorkAuthority } from './journal-observer-read'
-import { selectJournalEligibleQueueCard } from './journal-domain-input-eligibility'
+import {
+  selectJournalEligibleQueueCard,
+  journalDomainResolvedInputEligibility
+} from './journal-domain-input-eligibility'
+import { JournalSqlStatusContext } from './journal-sql-status-context'
+import { readJournalSqlWork, readJournalSqlWorkInput } from './journal-sql-work-read'
+import { sameJournalCursor } from './replay/journal-cursor'
+import { copyJournalCurrentWorkAuthority } from './journal-observer-read'
+import type { JournalSqlWorkResult, JournalWorkRead } from './journal-work-read'
 import type { JournalMailCustody } from './journal-writer-host-facts'
 import { readJournalDomainQueuePauseContext } from './journal-domain-queue-pauses'
 import type { AgentJournalCursor } from '../../../shared/agent-session/journal/agent-session-journal-identity'
@@ -29,6 +37,7 @@ export type JournalSqlQueueRead = { facts: JournalSqlQueueFacts } & (
      authority: JournalCurrentWorkAuthority
      allowSteer: boolean
      now: number
+      resolved?: JournalSqlWorkResult
    }
  | { kind: 'list' }
  | { kind: 'get'; messageId: string }
@@ -42,7 +51,7 @@ export type JournalSqlQueueRead = { facts: JournalSqlQueueFacts } & (
 )

 export type JournalSqlQueueReadResult =
-  | { kind: 'eligible-input-card'; card: QueuedMessageRow | null }
+  | { kind: 'eligible-input-card'; card: QueuedMessageRow | null; work?: JournalSqlWorkResult }
  | { kind: 'list'; cards: QueuedMessageRow[] }
  | { kind: 'get'; card: QueuedMessageRow | null }
  | { kind: 'revision'; revision: string }
@@ -64,6 +73,44 @@ export function readJournalSqlQueue(
  const { facts } = operation
  switch (operation.kind) {
    case 'eligible-input-card': {
+      if (operation.resolved) {
+        const context = new JournalSqlStatusContext(read, 'input')
+        const prior = operation.resolved
+        const cursor = { epoch: context.epoch, sequence: context.sequence }
+        const physicalCursor = { epoch: read.epoch, sequence: context.sequence }
+        const input: Extract<JournalWorkRead, { kind: 'input' }> = {
+          kind: 'input',
+          now: operation.now,
+          facts,
+          allowSteer: operation.allowSteer
+        }
+        const state = prior.result.kind === 'state' ? prior.result.state : undefined
+        const reusable =
+          state &&
+          sameJournalCursor(prior.cursor, cursor) &&
+          sameJournalCursor(prior.physicalCursor, physicalCursor) &&
+          JSON.stringify(copyJournalCurrentWorkAuthority(prior.authority)) ===
+            JSON.stringify(operation.authority)
+        const work =
+          reusable && state
+            ? {
+                cursor,
+                physicalCursor,
+                authority: operation.authority,
+                result: readJournalSqlWorkInput(context, operation.authority, input, state, () =>
+                  journalDomainResolvedInputEligibility(context, state)
+                )
+              }
+            : readJournalSqlWork(context, { authority: operation.authority, operation: input })
+        if (work.result.kind !== 'input') {
+          throw new Error('journal queue work answer changed kind')
+        }
+        return {
+          kind: operation.kind,
+          card: 'card' in work.result.queue ? work.result.queue.card : null,
+          work
+        }
+      }
      const selected = selectJournalEligibleQueueCard(
        read,
        operation.authority,
@@ -1,3 +1,14 @@
+import { readJournalSqlStatusHead } from './journal-sql-status-head'
+import type { SqliteRow } from '../../sqlite/sqlite-statement'
+import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session/requests/agent-session-queued-submission'
+import {
+  readJournalSqlWorkItems,
+  hydrateJournalSqlWorkDependencies
+} from './journal-sql-work-items'
+import {
+  journalDomainFirstQueuedInput,
+  journalDomainUnsettledInputs
+} from './journal-domain-input-populations'
 import type {
  AgentJournalRenderItem,
  AgentJournalSubmission
@@ -8,14 +19,10 @@ import { journalContextEpoch } from './context-epochs/journal-context-epoch'
 import type { JournalSqlScope } from './journal-sql-scope'
 import { AGENT_SESSION_ITEM_WIRE_VERSION } from '../agent-session-wire/publication/history/agent-session-item-presentation-contract'
 import type { JournalStatusItem, JournalStatusItemSelection } from './journal-status-items'
-import { sqlJournalStatusItems } from './journal-sql-status-items'
+import { sqlJournalStatusItems, sqlJournalStatusItem } from './journal-sql-status-items'
 import { sqlFailedStatusItems, sqlHasStatusRequestItem } from './journal-sql-status-request-items'
 import { sqlStatusRelatedInputIds } from './journal-sql-status-inputs'
-import {
-  readJournalDomainStopStates,
-  resolveJournalDomainDisplay,
-  journalDomainDisplayTurnState
-} from './journal-domain-item-display'
+import { readJournalDomainStopStates } from './journal-domain-item-display'
 import type { JournalDomainSubmissionStoredFacts } from './journal-domain-submission-records'
 import {
  journalDomainInputForItem,
@@ -32,7 +39,12 @@ export class JournalSqlStatusContext {
  private readonly selected = new Map<string, JournalStatusItem[]>()
  private readonly items = new Map<
    string,
-    | { item: AgentJournalRenderItem; displayJson: unknown; hidden: boolean; fence: number }
+    | {
+        item: AgentJournalRenderItem
+        displayed: AgentJournalRenderItem
+        hidden: boolean
+        fence: number
+      }
    | undefined
  >()
  private readonly submissions = new Map<string, AgentJournalSubmission | undefined>()
@@ -45,17 +57,26 @@ export class JournalSqlStatusContext {
  readonly submissionCount: number
  readonly stop: JournalLatestStop | null

-  constructor(readonly scope: JournalSqlScope) {
-    const row = scope.db
-      .prepare(`SELECT context_boundary,last_sequence,last_activity_at,queue_latest_stop_json,
-      (SELECT COALESCE(MAX(ordinal),-1)+1 FROM journal_submission_attempts
-       WHERE session_id=? AND epoch=?) AS submission_count
-      FROM journal_chat_state WHERE session_id=? AND epoch=?`)
-      .get(...scope.prefix, ...scope.prefix)
+  private unsettled: AgentJournalSubmission[] | undefined
+  private workFloor: number | undefined
+
+  readonly inputHead: SqliteRow | undefined
+  private rawWorkFloor: number | undefined
+
+  constructor(
+    readonly scope: JournalSqlScope,
+    work?: 'state' | 'input'
+  ) {
+    const row = readJournalSqlStatusHead(scope, work)
    if (!row) {
      throw new Error('journal status read targets a retired chat epoch')
    }
    scope.examined(row)
+    this.inputHead = work === 'input' ? row : undefined
+    if (work) {
+      this.workFloor = Number(row.work_floor ?? 0)
+      this.rawWorkFloor = Number(row.raw_work_floor ?? 0)
+    }
    this.epoch = journalContextEpoch(scope.epoch, Number(row.context_boundary))
    this.sequence = Number(row.last_sequence)
    this.lastActivityAt = Number(row.last_activity_at)
@@ -74,46 +95,122 @@ export class JournalSqlStatusContext {
        FROM journal_items WHERE session_id=? AND epoch=? AND item_id=?`)
        .get(...this.scope.prefix, id)
      if (row) {
-        this.scope.examined(row)
-        const source = this.scope.decode<AgentJournalRenderItem>(row.payload_json)
-        const states =
-          row.wire_version === AGENT_SESSION_ITEM_WIRE_VERSION
-            ? undefined
-            : readJournalDomainStopStates(this.scope.db, this.scope, [id])
-        const displayed = resolveJournalDomainDisplay(
-          source,
-          row.wire_version,
-          row.wire_json,
-          states && journalDomainDisplayTurnState(source, states)
-        )
-        row.display_json = displayed === source ? row.payload_json : JSON.stringify(displayed)
+        this.remember(row, id)
+      } else {
+        this.items.set(id, undefined)
      }
-
-      this.items.set(
-        id,
-        row
-          ? {
-              item: this.scope.decode<AgentJournalRenderItem>(row.payload_json),
-              displayJson: row.display_json === row.payload_json ? null : row.display_json,
-              hidden: Boolean(row.hidden),
-              fence: Number(row.fence)
-            }
-          : undefined
-      )
    }
    return this.items.get(id)
  }

+  private remember(
+    row: SqliteRow,
+    id: string,
+    states?: ReturnType<typeof readJournalDomainStopStates>
+  ) {
+    if (this.items.has(id)) {
+      return
+    }
+    this.items.set(id, sqlJournalStatusItem(this.scope, row, id, states))
+  }
+
+  private rememberAll(rows: SqliteRow[]) {
+    const stale = rows.filter(
+      (row) => row.payload_json != null && row.wire_version !== AGENT_SESSION_ITEM_WIRE_VERSION
+    )
+    const states = stale.length
+      ? readJournalDomainStopStates(
+          this.scope.db,
+          this.scope,
+          stale.map((row) => String(row.item_id))
+        )
+      : undefined
+    for (const row of rows) {
+      const id = String(row.item_id)
+      if (row.payload_json === null) {
+        this.items.set(id, undefined)
+      } else {
+        this.remember(row, id, states)
+      }
+    }
+  }
+
+  /** Work-only populations and their reciprocal turn/opener facts share this snapshot. */
+  prepareWork() {
+    const selections = new Map<string, JournalStatusItem[]>([
+      ['raw', []],
+      ['boundary', []],
+      ['pending-prompts', []]
+    ])
+    const rows = readJournalSqlWorkItems(this.scope)
+    this.rememberAll(rows)
+    for (const row of rows) {
+      const id = String(row.item_id)
+      selections.get(String(row.selection))?.push({
+        item: this.items.get(id)!.item,
+        position: {
+          sequence: Number(row.sequence),
+          sequenceIndex: Number(row.sequence_index),
+          ordinal: Number(row.ordinal)
+        }
+      })
+    }
+    for (const [key, items] of selections) {
+      this.selected.set(
+        JSON.stringify(key === 'raw' ? { kind: 'boundary', raw: true } : { kind: key }),
+        items
+      )
+    }
+    const sources = rows.flatMap((row) => {
+      const current = this.items.get(String(row.item_id))
+      return current && !current.hidden ? [current.displayed] : []
+    })
+    hydrateJournalSqlWorkDependencies(
+      this.scope,
+      sources,
+      (id) => {
+        const current = this.items.get(id)
+        return current && !current.hidden
+          ? current.displayed
+          : this.items.has(id)
+            ? null
+            : undefined
+      },
+      (rows) => this.rememberAll(rows)
+    )
+    this.unsettled = [...journalDomainUnsettledInputs(this.scope)]
+    this.recoverable = this.unsettled.filter(
+      (send) => send.dispatchState === 'pending' || !send.recovered
+    )
+    for (const send of this.unsettled) {
+      this.submissions.set(send.clientMessageId, send)
+    }
+  }
+
+  commandFloor = () =>
+    this.workFloor ??
+    this.select({ kind: 'clear' })[Symbol.iterator]().next().value?.item.sequence ??
+    0
+
+  enteredSubmissions = () =>
+    (this.unsettled ?? [...journalDomainUnsettledInputs(this.scope)]).filter(
+      (send) => !isQueuedAgentJournalSubmission(send)
+    )
+  sendingSubmissions = () =>
+    (this.unsettled ?? [...journalDomainUnsettledInputs(this.scope)]).filter(
+      (send) => send.dispatchState === 'pending'
+    )
+  firstQueuedInput = () =>
+    this.unsettled
+      ? this.sendingSubmissions().find(isQueuedAgentJournalSubmission)
+      : journalDomainFirstQueuedInput(this.scope)
+
  item = (id: string) => this.current(id)?.item

  /** Semantic display uses the writer's saved Stop rewrite without shortening ordinary items. */
  displayed = (id: string): AgentJournalRenderItem | null => {
    const row = this.current(id)
-    return row && !row.hidden
-      ? row.displayJson === null
-        ? row.item
-        : this.scope.decode<AgentJournalRenderItem>(row.displayJson)
-      : null
+    return row && !row.hidden ? row.displayed : null
  }

  submission = (id: string) => {
@@ -146,6 +243,10 @@ export class JournalSqlStatusContext {
  steerInput = (itemId: string) => journalDomainLatestSteer(this.scope, itemId)

  select = (selection: JournalStatusItemSelection): Iterable<JournalStatusItem> => {
+    const prepared = this.selected.get(JSON.stringify(selection))
+    if (prepared) {
+      return prepared
+    }
    if (!['clear', 'boundary', 'prompt', 'assistant-exists'].includes(selection.kind)) {
      return sqlJournalStatusItems(this.scope, this.item, selection)
    }
@@ -163,6 +264,9 @@ export class JournalSqlStatusContext {
    sqlStatusRelatedInputIds(this.scope, start, floor)

  contextFloor = (): number => {
+    if (this.rawWorkFloor !== undefined) {
+      return this.rawWorkFloor
+    }
    const row = this.scope.db
      .prepare(`SELECT sequence FROM journal_items
      WHERE session_id=? AND epoch=? AND context_clear=1 ORDER BY sequence DESC LIMIT 1`)
@@ -0,0 +1,30 @@
+import { JOURNAL_SQL_ACCEPTED_INPUT_COLUMNS } from './journal-sql-accepted-inputs'
+import type { JournalSqlScope } from './journal-sql-scope'
+
+export function readJournalSqlStatusHead(scope: JournalSqlScope, work?: 'state' | 'input') {
+  return scope.db
+    .prepare(`SELECT context_boundary,last_sequence,last_activity_at,queue_latest_stop_json,
+      (SELECT COALESCE(MAX(ordinal),-1)+1 FROM journal_submission_attempts
+       WHERE session_id=? AND epoch=?) AS submission_count
+      ${
+        work
+          ? `,(SELECT sequence FROM journal_items INDEXED BY journal_items_context_clear
+        WHERE session_id=head.session_id AND epoch=head.epoch AND hidden=0 AND context_clear=1
+        ORDER BY sequence DESC LIMIT 1) AS work_floor,
+        (SELECT sequence FROM journal_items INDEXED BY journal_items_raw_context_clear
+        WHERE session_id=head.session_id AND epoch=head.epoch AND context_clear=1
+        ORDER BY sequence DESC LIMIT 1) AS raw_work_floor`
+          : ''
+      }
+      ${
+        work === 'input'
+          ? `,(SELECT record_json FROM agent_session_records
+        WHERE session_id=head.session_id) AS record_json,
+        head.queue_manual_stop_json,head.queue_resumed_sequence,head.queue_reopened_sequence,
+        head.queue_clear_sequence,head.queue_clear_operation_id,head.queue_clear_lifted,
+        ${JOURNAL_SQL_ACCEPTED_INPUT_COLUMNS}`
+          : ''
+      }
+      FROM journal_chat_state head WHERE session_id=? AND epoch=?`)
+    .get(...scope.prefix, ...scope.prefix)
+}
@@ -1,3 +1,10 @@
+import type { SqliteRow } from '../../sqlite/sqlite-statement'
+import { AGENT_SESSION_ITEM_WIRE_VERSION } from '../agent-session-wire/publication/history/agent-session-item-presentation-contract'
+import {
+  readJournalDomainStopStates,
+  resolveJournalDomainDisplay,
+  journalDomainDisplayTurnState
+} from './journal-domain-item-display'
 import type { AgentJournalRenderItem } from '../../../shared/agent-session/journal/agent-session-journal-types'
 import type { JournalSqlScope } from './journal-sql-scope'
 import type { JournalStatusItem, JournalStatusItemSelection } from './journal-status-items'
@@ -114,3 +121,24 @@ export function* sqlJournalStatusItems(
    after = columns.map((column) => Number(last[column]))
  }
 }
+
+export function sqlJournalStatusItem(
+  scope: JournalSqlScope,
+  row: SqliteRow,
+  id: string,
+  states?: ReturnType<typeof readJournalDomainStopStates>
+) {
+  scope.examined(row)
+  const item = scope.decode<AgentJournalRenderItem>(row.payload_json)
+  const stopStates =
+    row.wire_version === AGENT_SESSION_ITEM_WIRE_VERSION
+      ? undefined
+      : (states ?? readJournalDomainStopStates(scope.db, scope, [id]))
+  const displayed = resolveJournalDomainDisplay(
+    item,
+    row.wire_version,
+    row.wire_json,
+    stopStates && journalDomainDisplayTurnState(item, stopStates)
+  )
+  return { item, displayed, hidden: Boolean(row.hidden), fence: Number(row.fence) }
+}
@@ -1,10 +1,9 @@
 import { AgentSessionSubscribers } from '../agent-session-wire/publication/subscribers/structured-agent-session-subscribers'
 import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session/agent-session-wire'
 import Database from '../../sqlite/sync-database'
-import { StructuredAgentSessionJournalProjections } from '../agent-session-wire/observations/structured-agent-session-status-journal-projection'
 import { StructuredAgentSessionStatusFeed } from '../agent-session-wire/observations/structured-agent-session-status-feed'
 import type { StatusFeedSession } from '../agent-session-wire/observations/structured-agent-session-status-feed-deps'
-import type { JournalWorkAuthority, JournalCurrentWorkAuthority } from './journal-observer-read'
+import type { JournalCurrentWorkAuthority } from './journal-observer-read'
 import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
@@ -300,85 +299,6 @@ it('preserves a corrupt database on worker startup refusal', async () => {
  expect(await readFile(path)).toEqual(bytes)
 })

-it('discards row-free authority changes before publishing an asynchronous observer answer', async () => {
-  const { journal } = await fixture()
-  const projections = new StructuredAgentSessionJournalProjections()
-  let revision = 0
-  const held: string[] = []
-  const capture = () => {
-    const authority: JournalWorkAuthority = {
-      handle: journal.observationHandle(),
-      record: {
-        currentFence: 1,
-        ownerObservation: { claimStatus: 'live', unreconciled: false, handoffStage: null }
-      },
-      child: { fence: 1, heldSends: held },
-      handingOver: [],
-      continuingUnadmitted: [],
-      finalizing: [],
-      openedThrough: journal.openingCursor(),
-      reopenFloor: null,
-      operationalRevision: revision,
-      stopRevision: 0,
-      stop: null
-    }
-    return { authority, contextId: null }
-  }
-  const reached = Promise.withResolvers<void>()
-  const release = Promise.withResolvers<void>()
-  const read = journal.reads.observer.bind(journal.reads)
-  const observer = vi.spyOn(journal.reads, 'observer').mockImplementation(async (input) => {
-    const result = await read(input)
-    reached.resolve()
-    await release.promise
-    return result
-  })
-  const invalidated: string[] = []
-  const observed: unknown[] = []
-  const pending = Promise.all(
-    ['first', 'second'].map((reader) =>
-      projections.readSql(
-        journal,
-        capture,
-        () => invalidated.push(reader),
-        (projection) => observed.push(projection)
-      )
-    )
-  )
-  await reached.promise
-  revision = 1
-  held.push('new-live-owner')
-  release.resolve()
-  expect(await pending).toEqual([null, null])
-  expect(invalidated).toEqual(['first', 'second'])
-  expect(observed).toEqual([])
-  observer.mockRestore()
-  const current = await projections.readSql(journal, capture, () => {})
-  expect(current).toMatchObject({
-    workRevision: 1,
-    authority: { child: { heldSends: ['new-live-owner'] } }
-  })
-  held.push('later-owner')
-  expect(current?.authority?.child?.heldSends).toEqual(['new-live-owner'])
-  expect(await projections.readSql(journal, capture, () => {})).toMatchObject({
-    authority: { child: { heldSends: ['later-owner', 'new-live-owner'] } }
-  })
-  revision = 2
-  const failed = Promise.withResolvers<Awaited<ReturnType<typeof read>>>()
-  const failure = new Error('observer read failed')
-  vi.spyOn(journal.reads, 'observer').mockImplementationOnce(() => failed.promise)
-  const failures = Promise.allSettled([
-    projections.readSql(journal, capture, () => {}),
-    projections.readSql(journal, capture, () => {})
-  ])
-  failed.reject(failure)
-  expect(await failures).toEqual([
-    { status: 'rejected', reason: failure },
-    { status: 'rejected', reason: failure }
-  ])
-  expect(await projections.readSql(journal, capture, () => {})).toMatchObject({ workRevision: 2 })
-})
-
 it('rechecks row-free owner death before answering an attempt-end action read', async () => {
  const { journal } = await fixture()
  await journal.appendItem(
@@ -43,7 +43,7 @@ export async function readJournalSqlWorkAnswer<K extends JournalWorkRead['kind']
  capture: () => JournalCurrentWorkAuthority,
  operation: Extract<JournalWorkRead, { kind: K }>
 ): Promise<JournalWorkAnswer<K>> {
-  const requested = structuredClone(operation)
+  const requested: JournalWorkRead = structuredClone(operation)
  for (;;) {
    if (journal.isClosed) {
      throw new AgentSessionJournalError(
@@ -52,7 +52,23 @@ export async function readJournalSqlWorkAnswer<K extends JournalWorkRead['kind']
      )
    }
    const authority = copyJournalCurrentWorkAuthority(capture())
-    const answer = await journal.reads.work({ authority, operation: requested })
+    let answer: JournalSqlWorkResult
+    if (requested.kind === 'input' && requested.resolved) {
+      const queue = await journal.reads.queue({
+        kind: 'eligible-input-card',
+        authority,
+        allowSteer: requested.allowSteer,
+        now: requested.now,
+        facts: requested.facts,
+        resolved: requested.resolved
+      })
+      if (queue.kind !== 'eligible-input-card' || !queue.work) {
+        throw new Error('journal queue reader omitted its resolved work answer')
+      }
+      answer = queue.work
+    } else {
+      answer = await journal.reads.work({ authority, operation: requested })
+    }
    if (journalSqlWorkAnswerIsCurrent(journal, capture, authority, answer)) {
      if (answer.result.kind !== operation.kind) {
        throw new Error(`journal work reader answered ${answer.result.kind} for ${operation.kind}`)
@@ -0,0 +1,79 @@
+import type { AgentJournalRenderItem } from '../../../shared/agent-session/journal/agent-session-journal-types'
+import { isHistoricalProviderTimelineItem } from '../agent-session-timeline/provider-timeline-history'
+import type { JournalSqlScope } from './journal-sql-scope'
+
+/** Independent indexed populations avoid multiplying prompts by unrelated obligations. */
+function journalSqlWorkItems(scope: JournalSqlScope) {
+  return scope.db
+    .prepare(`WITH raw_boundary AS (
+      SELECT 'raw' AS selection,item_id FROM journal_items INDEXED BY journal_items_raw_turn_marker
+      WHERE session_id=?1 AND epoch=?2 AND turn_marker=1
+      ORDER BY sequence DESC,ordinal DESC LIMIT 1),
+    visible_boundary AS (
+      SELECT 'boundary' AS selection,item_id FROM journal_items INDEXED BY journal_items_turn_marker
+      WHERE session_id=?1 AND epoch=?2 AND hidden=0 AND turn_marker=1
+      ORDER BY sequence DESC,sequence_index DESC,ordinal DESC LIMIT 1),
+    wanted AS (
+      SELECT selection,item_id FROM raw_boundary UNION ALL
+      SELECT selection,item_id FROM visible_boundary UNION ALL
+      SELECT 'pending-prompts',item_id FROM journal_items INDEXED BY journal_items_status_pending
+      WHERE session_id=?1 AND epoch=?2 AND hidden=0
+        AND kind IN ('approval','question') AND lifecycle_state='pending')
+    SELECT wanted.selection,item.item_id,item.sequence,item.sequence_index,item.ordinal,
+      item.payload_json,item.wire_version,item.wire_json,item.hidden,item.turn_item_id,item.opener_item_id,
+      COALESCE(item.execution_owner,item.owner_fence) AS fence
+    FROM wanted CROSS JOIN journal_items item
+    WHERE item.session_id=?1 AND item.epoch=?2 AND item.item_id=wanted.item_id`)
+    .iterate(...scope.prefix)
+}
+
+export function readJournalSqlWorkItems(scope: JournalSqlScope) {
+  return [...journalSqlWorkItems(scope)].sort(
+    (a, b) =>
+      Number(a.sequence) - Number(b.sequence) ||
+      Number(a.sequence_index) - Number(b.sequence_index) ||
+      Number(a.ordinal) - Number(b.ordinal)
+  )
+}
+
+export function readJournalSqlWorkRelatedItems(scope: JournalSqlScope, ids: readonly string[]) {
+  return scope.db
+    .prepare(`SELECT wanted.value AS item_id,item.payload_json,item.wire_version,item.wire_json,item.hidden,
+      COALESCE(item.execution_owner,item.owner_fence) AS fence
+    FROM json_each(?) wanted LEFT JOIN journal_items item
+    ON item.session_id=? AND item.epoch=? AND item.item_id=wanted.value`)
+    .all(JSON.stringify(ids), ...scope.prefix)
+}
+
+function journalSqlWorkDependencies(
+  items: AgentJournalRenderItem[],
+  read: (id: string) => AgentJournalRenderItem | null | undefined
+) {
+  const missing = new Set<string>()
+  for (const item of items) {
+    isHistoricalProviderTimelineItem(item, (id) => {
+      const value = read(id)
+      if (value === undefined) {
+        missing.add(id)
+      }
+      return value ?? null
+    })
+  }
+  return [...missing]
+}
+
+export function hydrateJournalSqlWorkDependencies(
+  scope: JournalSqlScope,
+  sources: AgentJournalRenderItem[],
+  read: (id: string) => AgentJournalRenderItem | null | undefined,
+  remember: (rows: ReturnType<typeof readJournalSqlWorkRelatedItems>) => void
+) {
+  // The shared policy asks for at most the owning turn and its reciprocal opener.
+  for (let depth = 0; depth < 2; depth++) {
+    const missing = journalSqlWorkDependencies(sources, read)
+    if (!missing.length) {
+      break
+    }
+    remember(readJournalSqlWorkRelatedItems(scope, missing))
+  }
+}
@@ -1,12 +1,14 @@
+import { readJournalSqlCurrentWork } from './journal-sql-current-work'
+export { readJournalSqlCurrentWork } from './journal-sql-current-work'
+import {
+  journalDomainInputEligibility,
+  selectJournalEligibleQueueCard
+} from './journal-domain-input-eligibility'
+import { journalSqlFailure } from './journal-sql-failure'
 import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session/requests/agent-session-queued-submission'
 import { agentJournalSubmissionKey } from '../../../shared/agent-session/journal/agent-session-journal-item-key'
 import { latestStructuredAgentSessionTurn } from '../../../shared/agent-session/view/structured-agent-session-live-turn'
-import {
-  StructuredAgentSessionWorkPolicy,
-  structuredAgentSessionLiveFence,
-  structuredAgentSessionOwnerUnverifiable
-} from '../agent-session-wire/observations/structured-agent-session-work-policy'
-import { journalWroteBeforeOpen } from './replay/journal-cursor'
+import type { StructuredAgentSessionWorkPolicy } from '../agent-session-wire/observations/structured-agent-session-work-policy'
 import type { JournalSqlStatusContext } from './journal-sql-status-context'
 import type { JournalCurrentWorkAuthority, JournalObserverResult } from './journal-observer-read'
 import type {
@@ -18,48 +20,6 @@ import type {
 import { journalInputAttemptEndedFromFacts } from './input-order/journal-input-attempt-ended'
 import { readJournalSqlInput } from './journal-sql-input-reads'

-export function readJournalSqlCurrentWork(
-  context: JournalSqlStatusContext,
-  authority: JournalCurrentWorkAuthority
-) {
-  const handingOver = new Set(authority.handingOver)
-  const continuing = new Set(authority.continuingUnadmitted)
-  const finalizing = new Set(authority.finalizing)
-  const held =
-    authority.child?.heldSends === undefined ? undefined : new Set(authority.child.heldSends)
-  const evidence = {
-    record: authority.record,
-    ended: authority.ended,
-    replaced: authority.replaced,
-    child:
-      authority.child === null
-        ? null
-        : authority.child === undefined
-          ? undefined
-          : { fence: authority.child.fence }
-  }
-  return new StructuredAgentSessionWorkPolicy(
-    {
-      item: context.displayed,
-      itemFence: context.itemFence,
-      runningTurn: context.runningTurn,
-      recoverableSubmissions: context.recoverableSubmissions,
-      wroteBeforeOpen: (sequence) =>
-        journalWroteBeforeOpen({ epoch: context.scope.epoch }, authority.openedThrough, sequence),
-      continuingInput: (id) =>
-        continuing.has(id) && context.submission(id)?.dispatchState === 'pending',
-      finalizingInput: (id) => finalizing.has(id),
-      handingOver: (id) => handingOver.has(id)
-    },
-    structuredAgentSessionLiveFence(evidence),
-    authority.operationalRevision,
-    authority.ended,
-    authority.replaced,
-    structuredAgentSessionOwnerUnverifiable(evidence),
-    () => held
-  )
-}
-
 /** The same finished work state serves observers and action readers. */
 export function readJournalSqlWorkState(
  context: JournalSqlStatusContext,
@@ -70,8 +30,7 @@ export function readJournalSqlWorkState(
    const displayed = work.isCurrentItem(item.itemId) ? context.displayed(item.itemId) : null
    return displayed ? [displayed] : []
  })
-  const floor =
-    context.select({ kind: 'clear' })[Symbol.iterator]().next().value?.item.sequence ?? 0
+  const floor = context.commandFloor()
  const owed = context
    .recoverableSubmissions()
    .filter((send) => (floor === 0 || (send.acceptedSequence ?? 0) > floor) && work.owesSend(send))
@@ -103,32 +62,40 @@ function owedPhaseSubmissions(
  work: StructuredAgentSessionWorkPolicy,
  kind: 'entered-submissions' | 'sending-submissions'
 ) {
-  const result = readJournalSqlInput(context.scope, { kind })
-  if (result.kind !== kind) {
-    throw new Error('journal phase read returned a different answer')
-  }
-  return result.submissions.filter((send) => work.owesSend(send))
+  const submissions =
+    kind === 'entered-submissions' ? context.enteredSubmissions() : context.sendingSubmissions()
+  return submissions.filter((send) => work.owesSend(send))
 }

 export function readJournalSqlWork(
  context: JournalSqlStatusContext,
  input: JournalSqlWorkRead
 ): JournalSqlWorkResult {
+  if (input.operation.kind === 'state' || input.operation.kind === 'input') {
+    context.prepareWork()
+  }
  const work = readJournalSqlCurrentWork(context, input.authority)
  return {
    cursor: { epoch: context.epoch, sequence: context.sequence },
    physicalCursor: { epoch: context.scope.epoch, sequence: context.sequence },
    authority: input.authority,
-    result: readWork(context, work, input.operation)
+    result: readWork(context, work, input.operation, input.authority)
  }
 }

 function readWork(
  context: JournalSqlStatusContext,
  work: StructuredAgentSessionWorkPolicy,
-  operation: JournalWorkRead
+  operation: JournalWorkRead,
+  authority: JournalCurrentWorkAuthority
 ): JournalWorkReadResult {
  switch (operation.kind) {
+    case 'input': {
+      const state = readJournalSqlWorkState(context, work)
+      return readJournalSqlWorkInput(context, authority, operation, state, () =>
+        journalDomainInputEligibility(context.scope, authority, context, work)
+      )
+    }
    case 'opening-send':
      return { kind: 'opening-send', opening: journalOpeningSend(context, work, operation.fence) }
    case 'state':
@@ -176,6 +143,29 @@ function readWork(
  }
 }

+/** Mandatory work survives the existing optional queue publication boundary. */
+export function readJournalSqlWorkInput(
+  context: JournalSqlStatusContext,
+  authority: JournalCurrentWorkAuthority,
+  operation: Extract<JournalWorkRead, { kind: 'input' }>,
+  state: JournalObserverResult['work'],
+  eligibility: () => Omit<ReturnType<typeof journalDomainInputEligibility>, 'work'>
+): Extract<JournalWorkReadResult, { kind: 'input' }> {
+  try {
+    const pick = selectJournalEligibleQueueCard(
+      context.scope,
+      authority,
+      operation.now,
+      operation.facts,
+      operation.allowSteer,
+      eligibility()
+    )
+    return { kind: 'input', state, queue: { card: pick.kind === 'selected' ? pick.card : null } }
+  } catch (error) {
+    return { kind: 'input', state, queue: { failure: journalSqlFailure(error) } }
+  }
+}
+
 /** Only pending handovers and the newest raw root turn can hold a subsequent send. */
 function journalOpeningSend(
  context: JournalSqlStatusContext,
@@ -1,9 +1,19 @@
+import type { JournalSqlQueueFacts } from './journal-sql-queue-read'
+import type { QueuedMessageRow } from './queued-messages/queued-message-table'
+import type { JournalSqlFailure } from './journal-sql-failure'
 import type { AgentJournalSubmission } from '../../../shared/agent-session/journal/agent-session-journal-types'
 import type { AgentJournalCursor } from '../../../shared/agent-session/journal/agent-session-journal-identity'
 import type { JournalCurrentWorkAuthority, JournalObserverResult } from './journal-observer-read'

 export type JournalWorkRead =
  | { kind: 'state' }
+  | {
+      kind: 'input'
+      now: number
+      facts: JournalSqlQueueFacts
+      allowSteer: boolean
+      resolved?: JournalSqlWorkResult
+    }
  | { kind: 'opening-send'; fence: number }
  | { kind: 'turn' }
  | { kind: 'child-work' }
@@ -14,6 +24,11 @@ export type JournalWorkRead =
 export type JournalWorkReadResult =
  | { kind: 'opening-send'; opening: boolean }
  | { kind: 'state'; state: JournalObserverResult['work'] }
+  | {
+      kind: 'input'
+      state: JournalObserverResult['work']
+      queue: { card: QueuedMessageRow | null } | { failure: JournalSqlFailure }
+    }
  | {
      kind: 'child-work'
      working: boolean
@@ -1,3 +1,4 @@
+import { restoreJournalSqlFailure } from '../../agent-session-journal/journal-sql-failure'
 import {
  QUEUED_MESSAGE_PAUSED_SEND_FAILED,
  type AgentSessionQueuedMessage,
@@ -13,16 +14,10 @@ import {
  type DerivedQueuePause
 } from '../../agent-session-journal/queued-messages/queued-message-pause'
 import { structuredQueuePauses } from './structured-agent-session-queued-pause'
-import {
-  nextStructuredQueuedMessage,
-  type StructuredQueueGateInput
-} from './structured-agent-session-queued-gate'
-import {
-  heldStructuredAgentSessionCurrentWork,
-  type StructuredAgentSessionCurrentWorkHost
-} from '../observations/structured-agent-session-host-current-work'
+import type { StructuredQueueGateInput } from './structured-agent-session-queued-gate'
+import type { StructuredAgentSessionCurrentWorkHost } from '../observations/structured-agent-session-host-current-work'
 import { captureHostStructuredAgentSessionWork } from '../observations/structured-agent-session-read-work'
-import type { StructuredAgentSessionObservedWork } from '../observations/structured-agent-session-observed-work'
+import { StructuredAgentSessionObservedWork } from '../observations/structured-agent-session-observed-work'
 import { journalSubmissionIsPersonal } from '../../agent-session-journal/submissions/journal-submission-fold'

 export type QueuePublication = {
@@ -33,9 +28,13 @@ export type QueuePublication = {
 }

 export type QueueSendGate = (
-  journal: AgentSessionJournal
+  journal: AgentSessionJournal,
+  cards: readonly QueuedMessageRow[]
 ) => Promise<
-  StructuredQueueGateInput & { state: ReturnType<StructuredAgentSessionObservedWork['queueState']> }
+  StructuredQueueGateInput & {
+    state: ReturnType<StructuredAgentSessionObservedWork['queueState']>
+    selected: QueuedMessageRow | null
+  }
 >

 export function structuredQueueSendGate(
@@ -43,22 +42,32 @@ export function structuredQueueSendGate(
    store: Pick<AgentSessionRecordStore, 'getRecord'>
  },
  sessionId: string,
-  capturedWork?: StructuredAgentSessionObservedWork
+  now: () => number,
+  onWork?: (work: StructuredAgentSessionObservedWork) => void,
+  resolved?: StructuredAgentSessionObservedWork
 ): QueueSendGate {
-  return async (journal) => {
-    const work =
-      capturedWork ??
-      (await heldStructuredAgentSessionCurrentWork(
-        host.store,
-        sessionId,
-        journal,
-        host.sessions.get(sessionId)
-      ))
+  return async (journal, cards) => {
+    const answer = await journal.readWork(
+      () => captureHostStructuredAgentSessionWork(host, sessionId, journal),
+      {
+        kind: 'input',
+        resolved: resolved?.readSource,
+        allowSteer: false,
+        now: now(),
+        facts: journal.queuedMessages.facts(journal.queuedMessages.answerMail(cards))
+      }
+    )
+    const work = new StructuredAgentSessionObservedWork(journal, answer.state, journal.cursor())
+    onWork?.(work)
+    if ('failure' in answer.queue) {
+      throw restoreJournalSqlFailure(answer.queue.failure)
+    }
    return {
      record: host.store.getRecord(sessionId),
      authority: captureHostStructuredAgentSessionWork(host, sessionId, journal),
      work,
-      state: work.queueState()
+      state: work.queueState(),
+      selected: answer.queue.card
    }
  }
 }
@@ -168,7 +177,7 @@ export async function readQueuePublication(
    const personal = { queue: { cards: new Map(cardRows.map((row) => [row.messageId, row])) } }
    const resumable = silent ? null : resumableQueuePause(pauses, cardRows)
    // The input order indexes pending sends, so a held queue never rescans history.
-    const input = await gate(journal)
+    const input = await gate(journal, cardRows)
    const { state } = input
    const sending = state.owedSendingSubmissions
    const pause =
@@ -186,8 +195,7 @@ export async function readQueuePublication(
      pause?.reason === 'stopped' || (pause?.reason === 'failed' && !working)
        ? { reason: pause.reason }
        : null
-    const nextQueuedMessageId =
-      (await nextStructuredQueuedMessage({ journal, ...input, cards: cardRows }))?.messageId ?? null
+    const nextQueuedMessageId = input.selected?.messageId ?? null
    const previous = publications.get(journal)
    if (
      previous &&
@@ -4,6 +4,7 @@ import type { AgentJournalTurnScope } from '../../../../shared/agent-session/jou
 import type { AgentJournalCursor } from '../../../../shared/agent-session/journal/agent-session-journal-identity'
 import type { AgentSessionLatestTurn } from '../../../../shared/agent-session/agent-session-wire'
 import type { JournalObserverResult } from '../../agent-session-journal/journal-observer-read'
+import type { JournalSqlWorkResult } from '../../agent-session-journal/journal-work-read'
 import type { SqlAgentSessionJournal } from '../../agent-session-journal/journal-sql-store'
 import { currentRequestsProjection } from './agent-session-current-requests'

@@ -14,7 +15,8 @@ export class StructuredAgentSessionObservedWork {
  constructor(
    private readonly journal: Pick<SqlAgentSessionJournal, 'cursor'>,
    private readonly work: JournalObserverResult['work'],
-    private readonly cursor: AgentJournalCursor
+    private readonly cursor: AgentJournalCursor,
+    readonly readSource?: JournalSqlWorkResult
  ) {
    this.restated = new Map(work.restatedSubmissions.map((send) => [send.clientMessageId, send]))
  }
@@ -1,4 +1,9 @@
-import { copyJournalWorkAuthority } from '../../agent-session-journal/journal-observer-read'
+import {
+  copyJournalWorkAuthority,
+  copyJournalCurrentWorkAuthority
+} from '../../agent-session-journal/journal-observer-read'
+import { journalSqlWorkAnswerIsCurrent } from '../../agent-session-journal/journal-sql-work-answer'
+import type { JournalSqlWorkResult } from '../../agent-session-journal/journal-work-read'
 import type { SqlAgentSessionJournal } from '../../agent-session-journal/journal-sql-store'
 import type {
  JournalWorkAuthority,
@@ -32,11 +37,14 @@ export type StructuredAgentSessionJournalProjection = {
  work?: JournalObserverResult['work']
 }

-type ProjectionEntry = {
-  projection: StructuredAgentSessionJournalProjection
-  physicalCursor: AgentJournalCursor
-  savedStop: JournalLatestStop | null
-}
+type ProjectionEntry =
+  | {
+      kind: 'full'
+      projection: StructuredAgentSessionJournalProjection
+      physicalCursor: AgentJournalCursor
+      savedStop: JournalLatestStop | null
+    }
+  | { kind: 'work-only'; answer: JournalSqlWorkResult; contextId: string | null }

 type PendingObservation = { key: string; result: Promise<JournalObserverResult> }
 type CapturedWork = { authority: JournalWorkAuthority; contextId: string | null }
@@ -51,17 +59,24 @@ export class StructuredAgentSessionJournalProjections {
    contextId: string | null
  ): { value: JournalLatestStop | null } | undefined {
    const entry = this.currentEntry(journal)
-    return entry?.projection.contextId === contextId ? { value: entry.savedStop } : undefined
+    return entry?.kind === 'full' && entry.projection.contextId === contextId
+      ? { value: entry.savedStop }
+      : undefined
  }

  private currentEntry(journal: SqlAgentSessionJournal): ProjectionEntry | undefined {
    const entry = this.byJournal.get(journal)
-    const authority = entry?.projection.authority
+    const projection = entry?.kind === 'full' ? entry.projection : entry?.answer.cursor
+    const authority = entry?.kind === 'full' ? entry.projection.authority : entry?.answer.authority
+    const physicalCursor =
+      entry?.kind === 'full' ? entry.physicalCursor : entry?.answer.physicalCursor
    return entry &&
      authority &&
+      projection &&
+      physicalCursor &&
      !journal.isClosed &&
-      sameJournalCursor(entry.projection, journal.cursor()) &&
-      sameJournalCursor(entry.physicalCursor, journal.physicalCursor()) &&
+      sameJournalCursor(projection, journal.cursor()) &&
+      sameJournalCursor(physicalCursor, journal.physicalCursor()) &&
      authority.handle === journal.observationHandle() &&
      sameJournalCursor(authority.openedThrough, journal.openingCursor())
      ? entry
@@ -72,7 +87,8 @@ export class StructuredAgentSessionJournalProjections {
    journal: SqlAgentSessionJournal,
    captured: CapturedWork | null
  ): StructuredAgentSessionJournalProjection | null {
-    const prior = this.currentEntry(journal)?.projection
+    const entry = this.currentEntry(journal)
+    const prior = entry?.kind === 'full' ? entry.projection : undefined
    return captured &&
      prior?.authority &&
      prior.contextId === captured.contextId &&
@@ -82,6 +98,60 @@ export class StructuredAgentSessionJournalProjections {
      : null
  }

+  /** Narrow work and full status occupy the same validated per-journal entry. */
+  async readWorkSql(journal: SqlAgentSessionJournal, capture: () => CapturedWork | null) {
+    for (;;) {
+      const captured = capture()
+      if (!captured) {
+        this.byJournal.delete(journal)
+        return null
+      }
+      const authority = copyJournalCurrentWorkAuthority(captured.authority)
+      const entry = this.currentEntry(journal)
+      const answer: JournalSqlWorkResult | undefined =
+        entry?.kind === 'work-only'
+          ? entry.contextId === captured.contextId
+            ? entry.answer
+            : undefined
+          : entry?.projection.work &&
+              entry.projection.contextId === captured.contextId &&
+              entry.projection.authority
+            ? {
+                cursor: { epoch: entry.projection.epoch, sequence: entry.projection.sequence },
+                physicalCursor: entry.physicalCursor,
+                authority: copyJournalCurrentWorkAuthority(entry.projection.authority),
+                result: { kind: 'state', state: entry.projection.work }
+              }
+            : undefined
+      if (
+        answer &&
+        journalSqlWorkAnswerIsCurrent(journal, () => captured.authority, authority, answer)
+      ) {
+        return answer
+      }
+      const result = await journal.reads.work({ authority, operation: { kind: 'state' } })
+      const recaptured = capture()
+      if (!recaptured) {
+        this.byJournal.delete(journal)
+        return null
+      }
+      if (
+        recaptured.contextId !== captured.contextId ||
+        !journalSqlWorkAnswerIsCurrent(journal, () => recaptured.authority, authority, result)
+      ) {
+        continue
+      }
+      if (this.currentEntry(journal)?.kind !== 'full') {
+        this.byJournal.set(journal, {
+          kind: 'work-only',
+          answer: result,
+          contextId: captured.contextId
+        })
+      }
+      return result
+    }
+  }
+
  /** Discard an invalidated asynchronous answer before touching the existing publication cache. */
  async readSql(
    journal: SqlAgentSessionJournal,
@@ -105,7 +175,8 @@ export class StructuredAgentSessionJournalProjections {
    const authority = copyJournalWorkAuthority(captured.authority)
    const signature = JSON.stringify(authority)
    const cursor = journal.cursor()
-    const prior = this.byJournal.get(journal)?.projection
+    const priorEntry = this.byJournal.get(journal)
+    const prior = priorEntry?.kind === 'full' ? priorEntry.projection : undefined
    const sameContext =
      prior?.epoch === cursor.epoch && prior.contextId === contextId ? prior : undefined
    const input = {
@@ -165,7 +236,12 @@ export class StructuredAgentSessionJournalProjections {
    }
    const stop = authority.stop
    const savedStop = stop && { sequence: stop.sequence, event: stop.event }
-    this.byJournal.set(journal, { projection, physicalCursor: result.physicalCursor, savedStop })
+    this.byJournal.set(journal, {
+      kind: 'full',
+      projection,
+      physicalCursor: result.physicalCursor,
+      savedStop
+    })
    observed?.(projection)
    return projection
  }
@@ -1,4 +1,5 @@
 import type { SqlAgentSessionJournal } from '../../agent-session-journal/journal-sql-store'
+import { StructuredAgentSessionObservedWork } from './structured-agent-session-observed-work'
 import { captureStructuredAgentSessionWorkAuthority } from './structured-agent-session-work-authority'
 import type { StructuredAgentSessionStatusFeedDeps } from './structured-agent-session-status-feed-deps'
 import {
@@ -40,23 +41,6 @@ export function createStructuredAgentSessionStatusReader(
    )
  }

-  async function current(sessionId: string, journal: SqlAgentSessionJournal) {
-    const saved = projections.savedStop(
-      journal,
-      deps.getConversation(sessionId)?.providerContextBoundary?.operationId ?? null
-    )
-    return saved
-      ? projections.currentSql(
-          journal,
-          captureStructuredAgentSessionWorkAuthority(journal, sessionId, deps, {
-            cursor: journal.cursor(),
-            revision: journal.stopMarks.revision(),
-            latest: journal.stopMarks.withCurrentSettlement(saved.value)
-          })
-        )
-      : null
-  }
-
  async function read(
    sessionId: string,
    journal?: SqlAgentSessionJournal,
@@ -67,5 +51,27 @@ export function createStructuredAgentSessionStatusReader(
    return source ? readSql(sessionId, source, refresh, observed) : null
  }

-  return Object.assign(read, { current })
+  const work = async (sessionId: string, journal: SqlAgentSessionJournal) => {
+    const answer = await projections.readWorkSql(journal, () =>
+      captureStructuredAgentSessionWorkAuthority(journal, sessionId, deps, {
+        cursor: journal.cursor(),
+        revision: journal.stopMarks.revision(),
+        latest: null
+      })
+    )
+    if (!answer) {
+      return null
+    }
+    if (answer.result.kind !== 'state') {
+      throw new Error('journal work reader returned another answer')
+    }
+    return new StructuredAgentSessionObservedWork(
+      journal,
+      answer.result.state,
+      answer.cursor,
+      answer
+    )
+  }
+
+  return Object.assign(read, { work })
 }
@@ -1,3 +1,5 @@
+import { StructuredAgentSessionJournalProjections } from '../structured-agent-session-status-journal-projection'
+import type { JournalWorkAuthority } from '../../../agent-session-journal/journal-observer-read'
 import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../../../shared/agent-session/journal/agent-session-journal-types'
 import { mkdtemp, rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
@@ -115,3 +117,110 @@ describe('structured status canonical owner re-entry', () => {
    ])
  })
 })
+
+it('discards row-free authority changes before publishing an asynchronous observer answer', async () => {
+  const journal = await openJournal()
+  const projections = new StructuredAgentSessionJournalProjections()
+  let revision = 0
+  let attached = true
+  const held: string[] = []
+  const capture = () => {
+    if (!attached) {
+      return null
+    }
+    const authority: JournalWorkAuthority = {
+      handle: journal.observationHandle(),
+      record: {
+        currentFence: 1,
+        ownerObservation: { claimStatus: 'live', unreconciled: false, handoffStage: null }
+      },
+      child: { fence: 1, heldSends: held },
+      handingOver: [],
+      continuingUnadmitted: [],
+      finalizing: [],
+      openedThrough: journal.openingCursor(),
+      reopenFloor: null,
+      operationalRevision: revision,
+      stopRevision: 0,
+      stop: null
+    }
+    return { authority, contextId: null }
+  }
+  const narrow = await projections.readWorkSql(journal, capture)
+  expect(narrow?.result.kind).toBe('state')
+  expect(projections.currentSql(journal, capture())).toBeNull()
+  expect(projections.savedStop(journal, null)).toBeUndefined()
+  await journal.appendItem(
+    { provider: 'orca', clientMessageId: 'work-turn' },
+    { kind: 'turn', turnId: 'work-turn', state: 'running' },
+    { fence: 1, turnScope: { kind: 'thread' } }
+  )
+  expect(await projections.readWorkSql(journal, capture)).toMatchObject({
+    cursor: journal.cursor(),
+    result: { kind: 'state', state: { latestTurn: { turn: { state: 'running' } } } }
+  })
+  expect(narrow?.cursor).not.toEqual(journal.cursor())
+  attached = false
+  expect(await projections.readWorkSql(journal, capture)).toBeNull()
+  attached = true
+  const reattachFailure = new Error('reattached work read failed')
+  vi.spyOn(journal.reads, 'work').mockRejectedValueOnce(reattachFailure)
+  await expect(projections.readWorkSql(journal, capture)).rejects.toBe(reattachFailure)
+  await projections.readWorkSql(journal, capture)
+  const reached = Promise.withResolvers<void>()
+  const release = Promise.withResolvers<void>()
+  const read = journal.reads.observer.bind(journal.reads)
+  const observer = vi.spyOn(journal.reads, 'observer').mockImplementation(async (input) => {
+    const result = await read(input)
+    reached.resolve()
+    await release.promise
+    return result
+  })
+  const invalidated: string[] = []
+  const observed: unknown[] = []
+  const pending = Promise.all(
+    ['first', 'second'].map((reader) =>
+      projections.readSql(
+        journal,
+        capture,
+        () => invalidated.push(reader),
+        (projection) => observed.push(projection)
+      )
+    )
+  )
+  await reached.promise
+  revision = 1
+  held.push('new-live-owner')
+  release.resolve()
+  expect(await pending).toEqual([null, null])
+  expect(invalidated).toEqual(['first', 'second'])
+  expect(observed).toEqual([])
+  observer.mockRestore()
+  const current = await projections.readSql(journal, capture, () => {})
+  expect(current).toMatchObject({
+    workRevision: 1,
+    authority: { child: { heldSends: ['new-live-owner'] } }
+  })
+  await projections.readWorkSql(journal, capture)
+  expect(projections.currentSql(journal, capture())).toBe(current)
+  expect(projections.savedStop(journal, null)).toEqual({ value: null })
+  held.push('later-owner')
+  expect(current?.authority?.child?.heldSends).toEqual(['new-live-owner'])
+  expect(await projections.readSql(journal, capture, () => {})).toMatchObject({
+    authority: { child: { heldSends: ['later-owner', 'new-live-owner'] } }
+  })
+  revision = 2
+  const failed = Promise.withResolvers<Awaited<ReturnType<typeof read>>>()
+  const failure = new Error('observer read failed')
+  vi.spyOn(journal.reads, 'observer').mockImplementationOnce(() => failed.promise)
+  const failures = Promise.allSettled([
+    projections.readSql(journal, capture, () => {}),
+    projections.readSql(journal, capture, () => {})
+  ])
+  failed.reject(failure)
+  expect(await failures).toEqual([
+    { status: 'rejected', reason: failure },
+    { status: 'rejected', reason: failure }
+  ])
+  expect(await projections.readSql(journal, capture, () => {})).toMatchObject({ workRevision: 2 })
+})
@@ -39,7 +39,8 @@ export type SubscriberFieldHooks = {
  ) => AgentSessionSlashCommand[] | undefined
  readQueuePublication?: (
    sessionId: string,
-    work?: StructuredAgentSessionObservedWork
+    work?: StructuredAgentSessionObservedWork,
+    onWork?: (work: StructuredAgentSessionObservedWork) => void
  ) => QueuePublication | undefined | Promise<QueuePublication | undefined>
  /** Read expensive child-derived rosters only when a frame owes them. */
  readBackgroundTasks?: (sessionId: string) => AgentSessionBackgroundTaskState | null
@@ -111,12 +111,11 @@ export class StructuredAgentSessionBackgroundTaskChannel {

  /** What a waiter decides from, from the sources history pages state, without building a page. */
  async workState(sessionId: string): Promise<AgentSessionWorkState> {
-    await this.conversation(sessionId)
-    const work = await this.currentWork(sessionId)
+    const { journal } = await this.conversation(sessionId)
+    const { work, queue } = await this.subscribers.readWorkAndQueue(sessionId, journal)
    if (!work) {
      throw new AgentSessionRefusalError(AGENT_SESSION_NOT_ATTACHED)
    }
-    const queue = await this.subscribers.readQueuePublication(sessionId)
    const fence = this.deps.store.getConversation(sessionId)?.currentFence
    return {
      working: work.working(),
@@ -28,7 +28,6 @@ import {
  sendSettlementReading
 } from '../../delivery/structured-agent-session-send-settlement'
 import { readHostStructuredAgentSessionWork } from '../../observations/structured-agent-session-read-work'
-import { projectedStructuredAgentSessionCurrentWork } from '../../observations/structured-agent-session-host-current-work'
 import { structuredAgentSessionConversationFence } from '../../host/structured-agent-session-provider-child'
 import type { StructuredAgentSessionStatusFeed } from '../../observations/structured-agent-session-status-feed'
 import type { StructuredAgentSessionStatusSubscriber as StatusSubscriber } from '../../observations/structured-agent-session-status-feed-deps'
@@ -112,17 +111,16 @@ export class StructuredAgentSessionClientDelivery {
      readForkMarker: (sessionId) => readForkMarker(this.deps().store, sessionId),
      readCommands: (sessionId, commandsOnRequest) =>
        this.readCommands(sessionId, commandsOnRequest),
-      readQueuePublication: (sessionId) =>
-        tryReadQueuePublication(sessions.get(sessionId)?.journal, async (journal) =>
+      readQueuePublication: (sessionId, work, onWork) =>
+        tryReadQueuePublication(
+          sessions.get(sessionId)?.journal,
          structuredQueueSendGate(
            { store: this.deps().store, sessions },
            sessionId,
-            (await this.currentWork(
-              sessionId,
-              journal,
-              this.statusFeed.journalProjection.current(sessionId, journal)
-            )) ?? undefined
-          )(journal)
+            now,
+            onWork,
+            work
+          )
        ),
      readBackgroundTasks,
      onJournalPublished: (sessionId, journal) => this.publishJournal(sessionId, journal),
@@ -315,13 +313,8 @@ export class StructuredAgentSessionClientDelivery {
  }

  // Called inside queue admission too: misses must use direct readers, never admit/readInOrder.
-  private currentWork(
-    sessionId: string,
-    journal: AgentSessionJournal,
-    projection = this.statusFeed.journalProjection(sessionId, journal)
-  ) {
-    const host = { store: this.deps().store, sessions: this.sessions }
-    return projectedStructuredAgentSessionCurrentWork(host, sessionId, journal, projection)
+  private currentWork(sessionId: string, journal: AgentSessionJournal) {
+    return this.statusFeed.journalProjection.work(sessionId, journal)
  }

  private settlementReading(sessionId: string, journal: AgentSessionJournal) {
@@ -1,3 +1,4 @@
+import type { StructuredAgentSessionObservedWork } from '../../observations/structured-agent-session-observed-work'
 import type { AgentJournalRenderItem } from '../../../../../shared/agent-session/journal/agent-session-journal-types'
 import { subscriberDeliveryPort } from './agent-session-subscriber-delivery-port'
 import type { QueuePublication } from '../../delivery/structured-agent-session-queued-publication'
@@ -157,6 +158,16 @@ export class AgentSessionSubscribers {
    return this.hooks.readQueuePublication?.(sessionId)
  }

+  async readWorkAndQueue(sessionId: string, journal: AgentSessionJournal) {
+    let work: StructuredAgentSessionObservedWork | null = null
+    const queue = await this.hooks.readQueuePublication?.(sessionId, undefined, (value) => {
+      work = value
+    })
+    // Discovery may fail before work was read; queue unavailability never erases valid work.
+    work ??= (await this.hooks.readCurrentWork?.(sessionId, journal)) ?? null
+    return { work, queue }
+  }
+
  deliverFrames(
    sessionId: string,
    journal: AgentSessionJournal,
@@ -1,6 +1,6 @@
+import { SqlJournalQueuedMessages } from '../../native-chat/agent-session-journal/journal-sql-queued-messages'
 import { afterEach, describe, expect, it, vi } from 'vitest'
 import type { RpcResponse } from '../rpc/core'
-import { StructuredAgentSessionJournalProjections } from '../../native-chat/agent-session-wire/observations/structured-agent-session-status-journal-projection'
 import { AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY } from '../../../shared/agent-session/agent-session-runtime-capabilities'
 import { RuntimeClient, RuntimeClientError } from '../../../cli/runtime-client'
 import { CHAT_HANDLERS } from '../../../cli/handlers/chat'
@@ -98,7 +98,6 @@ describe('chat send --wait over the real host', () => {
    harness = await openStructuredCodexRpcHarness([], {}, { localCli: true })
    await harness.ok('agentSession.create', harness.createIntentParams())
    await seedRunningTurn(1, 16)
-    vi.spyOn(StructuredAgentSessionJournalProjections.prototype, 'readSql').mockResolvedValue(null)
    const response = await harness.call('agentSession.workState', { sessionId: SESSION })
    expect(response).toMatchObject({
      ok: true,
@@ -119,6 +118,25 @@ describe('chat send --wait over the real host', () => {
    ).toMatchObject({ ok: true, result: { page: { working: true } } })
  })

+  it('keeps work available when optional queue discovery fails', async () => {
+    harness = await openStructuredCodexRpcHarness([], {}, { localCli: true })
+    await harness.ok('agentSession.create', harness.createIntentParams())
+    await seedRunningTurn(1, 16)
+    vi.spyOn(SqlJournalQueuedMessages.prototype, 'answerMail').mockImplementation(() => {
+      throw new Error('queue discovery unavailable')
+    })
+    const response = await harness.call('agentSession.workState', { sessionId: SESSION })
+    expect(response).toMatchObject({
+      ok: true,
+      result: { working: true, actionablePrompt: false, fence: expect.any(Number) }
+    })
+    if (!response.ok) {
+      throw new Error('workState failed')
+    }
+    expect(response.result).not.toHaveProperty('queuedMessageIds')
+    expect(response.result).not.toHaveProperty('nextQueuedMessageId')
+  })
+
  it('returns the completed turn when the host lacks agentSession.workState', async () => {
    harness = await openStructuredCodexRpcHarness(
      [],