Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
0b9642d
fix(native-chat): keep a message the host accepted then rejected in t…
brennanb2025 Oct 2, 2026
4fd1c77
test(native-chat): pin the same-id rule apart from the body match
brennanb2025 Oct 2, 2026
2003f48
test(native-chat): type the rejected-in-place fixture body as a text …
brennanb2025 Oct 2, 2026
13a2ef8
fix(native-chat): let the host's row own a message it recorded and th…
brennanb2025 Oct 2, 2026
39d7cf2
test(native-chat): keep one row when copies of a rejected message sha…
brennanb2025 Oct 2, 2026
cfa2644
refactor(native-chat): let the host's notice replace the outbox's und…
brennanb2025 Oct 2, 2026
fd4d30e
Merge remote-tracking branch 'origin/main' into brennanb2025/rejected…
brennanb2025 Oct 2, 2026
2252ced
test(native-chat): pass the queued card ids in the tool-stream cost t…
brennanb2025 Oct 2, 2026
f5d29b5
fix(native-chat): keep the host's record as what lets a rejected mess…
brennanb2025 Oct 2, 2026
2783ccc
fix(native-chat): draw a queued message where the host rejected it
brennanb2025 Oct 2, 2026
8711607
test(native-chat): pin that rows written together roll back together
brennanb2025 Oct 2, 2026
31ae1c1
fix(native-chat): draw every rejected message where it was rejected
brennanb2025 Oct 2, 2026
a664eb2
fix(native-chat): decide a rejected message's Retry from the host's s…
brennanb2025 Oct 2, 2026
9681cbb
fix(native-chat): write no start-failure row when a Stop withdrew eve…
brennanb2025 Oct 2, 2026
c4b31ab
test(native-chat): pass the Dismiss action in the delivery-notice hoo…
brennanb2025 Oct 2, 2026
7f7636e
Merge remote-tracking branch 'origin/main' into brennanb2025/rejected…
brennanb2025 Oct 2, 2026
33fc577
fix(native-chat): keep a rejected message's outbox copy until its row…
brennanb2025 Oct 2, 2026
50f8525
test(native-chat): type the outbox hook props that carry loaded rows
brennanb2025 Oct 2, 2026
bc82e42
fix(native-chat): write nothing when a journal batch settles nothing …
brennanb2025 Oct 2, 2026
6ef010a
test(native-chat): count storage writes on the outbox's own storage o…
brennanb2025 Oct 2, 2026
831e8eb
Merge remote-tracking branch 'origin/main' into brennanb2025/rejected…
brennanb2025 Oct 2, 2026
804ca8c
fix(native-chat): let a recorded rejected message's outbox copy leave…
brennanb2025 Oct 2, 2026
50c957b
Merge remote-tracking branch 'origin/main' into brennanb2025/rejected…
brennanb2025 Oct 2, 2026
7c7a4c6
Merge origin/main into brennanb2025/rejected-message-shows-in-chat
brennanb2025 Oct 4, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion mobile/src/session/use-mobile-structured-agent-session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,11 @@ export function useMobileStructuredAgentSession(args: {
})

const messages = useMemo(
() => projectStructuredAgentSessionMessages(state.items, [], state.submissions),
// Off: the phone hands a rejected message back to its composer, so a row would show it twice.
() =>
projectStructuredAgentSessionMessages(state.items, [], state.submissions, {
rejectedInPlace: false
}),
[state.items, state.submissions]
)
const turnId = activeStructuredAgentSessionTurnId(state.items)
Expand Down
4 changes: 3 additions & 1 deletion src/main/codex/codex-collab-call-delegation.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,9 @@ import { THREAD_ID } from './codex-structured-session-adapter-fixture'
/** The delegation the parent's newest tool run reads as, the row a running chat's frontier judges. */
async function newestRunDelegation(frames: Frame[]): Promise<NativeChatSubagentDelegation | null> {
const { conversation } = projectNativeChatTranscript(
projectStructuredAgentSessionMessages(await publishedRows(frames), [], [])
projectStructuredAgentSessionMessages(await publishedRows(frames), [], [], {
rejectedInPlace: true
})
)
const runs = conversation.filter((message: NativeChatMessage) =>
message.blocks.some((block) => block.type === 'tool-call')
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,9 @@ function positionalRuns(rows: AgentJournalRenderItem[]): {
})
})
const transcript = projectNativeChatTranscript(
projectStructuredAgentSessionMessages(rows, [], []).map(withoutCallIds)
projectStructuredAgentSessionMessages(rows, [], [], { rejectedInPlace: true }).map(
withoutCallIds
)
)
// The conversation's runs, then each helper's section's.
const runs = [
Expand Down
7 changes: 4 additions & 3 deletions src/main/codex/codex-structured-question-order.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -206,6 +206,7 @@ function drawnPromptRows(): string[][] {
client.items,
[],
client.submissions,
{ rejectedInPlace: true },
projectStructuredQuestionMessages
)
)
Expand Down Expand Up @@ -243,9 +244,9 @@ describe('a Codex ask with several questions', () => {
])
// Mobile draws the shared projection in journal order, one row per question.
expect(
projectStructuredAgentSessionMessages(client.items, [], client.submissions).map(
({ blocks }) => (blocks[0]?.type === 'text' ? blocks[0].text.split('\n')[0] : null)
)
projectStructuredAgentSessionMessages(client.items, [], client.submissions, {
rejectedInPlace: false
}).map(({ blocks }) => (blocks[0]?.type === 'text' ? blocks[0].text.split('\n')[0] : null))
).toEqual(ASKED.map(({ question }) => question))
})

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,11 @@ import {
import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key'
import { journalDispatchRowApplies } from './journal-dispatch-settlement'
import type { JournalReducerState } from './journal-reducer'
import { notePersonTurnAccepted, placeHandedOverMessage } from './journal-submission-fold'
import {
notePersonTurnAccepted,
placeHandedOverMessage,
placeRejectedMessage
} from './journal-submission-fold'
import type { JournalRow } from './journal-row-schema'

export function applyJournalDispatchRow(
Expand All @@ -34,6 +38,8 @@ export function applyJournalDispatchRow(
if (row.state === 'pending') {
submission.handedOverAt = row.ts
placeHandedOverMessage(state, submission, row)
} else if (row.state === 'rejected') {
placeRejectedMessage(state, submission, row)
}
if (row.recovered) {
submission.recovered = row.recovered
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import type { JournalReducerState } from './journal-reducer'
import { journalLifecycleBatchRowBuilder } from './journal-row-builders'
import type { JournalLifecycleBatchInput } from './journal-store-contracts'
import type { JournalRow } from './journal-row-schema'
import { journalQueuedRejectionRowBuilders } from './journal-pending-submission-recovery'

const SETTLEMENT_ALREADY_APPLIED = new Error('journal_settlement_already_applied')

Expand All @@ -12,10 +13,38 @@ export class JournalLifecycleBatchAppender {
state: () => JournalReducerState
cursor: () => AgentJournalCursor
enqueue: (build: (seq: number, ts: number) => JournalRow) => Promise<JournalRow>
enqueueRows: (
plan: () => readonly ((seq: number, ts: number) => JournalRow)[]
) => Promise<JournalRow[]>
}
) {}

append(input: JournalLifecycleBatchInput): Promise<AgentJournalCursor> {
const { rejectsQueued } = input
if (rejectsQueued) {
// Planned on the lane: the sends queued then, and this batch unless it already landed. With
// none left (a Stop withdrew them first) it failed no one, so nothing is written.
return this.deps
.enqueueRows(() => {
const rejections = journalQueuedRejectionRowBuilders(
this.deps.state,
input.fence,
rejectsQueued
)
return rejections.length === 0 || this.wasApplied(input.settlementId)
? rejections
: [
...rejections,
journalLifecycleBatchRowBuilder(
this.deps.state,
input.settlementId,
input.mutations,
input
)
]
})
.then(() => this.deps.cursor())
}
if (this.wasApplied(input.settlementId)) {
return Promise.resolve(this.deps.cursor())
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,9 @@ import type { AgentJournalDispatchRejection } from '../../../shared/agent-sessio
import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types'
import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission'
import { DISPATCH_DOUBT_HOST_RESTARTED } from './journal-dispatch-doubt-reasons'
import type { JournalReducerState } from './journal-reducer'
import { journalDispatchRowBuilder } from './journal-row-builders'
import type { JournalRow } from './journal-row-schema'
import type { AgentSessionJournal } from './journal-store'

/** Settles every submission a process fact left unanswerable. Doubt is never
Expand Down Expand Up @@ -88,3 +91,21 @@ export async function rejectJournalQueuedSubmissions(
)
return queued.map((entry) => entry.clientMessageId)
}

/** Rows rejecting every submission still queued, read from `state` when called: for an append
* that must carry them with what follows, in one transaction. */
export function journalQueuedRejectionRowBuilders(
state: () => JournalReducerState,
fence: number,
rejection: AgentJournalDispatchRejection
): ((seq: number, ts: number) => JournalRow)[] {
return [...state().submissions.values()].filter(isQueuedAgentJournalSubmission).map((entry) =>
journalDispatchRowBuilder(state, {
clientMessageId: entry.clientMessageId,
state: 'rejected',
...rejection,
fence,
recovered: true
})
)
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
// A failed start's row and the queued messages it failed land in ONE append, the messages first:
// no reader meets one without the other, and the messages sit above the row that says why.

import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, expect, it } from 'vitest'
import { agentSessionFailureFact } from '../../../shared/agent-session-failure'
import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words'
import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key'
import {
AGENT_JOURNAL_THREAD_SCOPE,
type AgentSessionJournalIdentity
} from '../../../shared/agent-session-journal-types'
import type { AgentSessionJournal } from './journal-store'
import {
closeTestJournalHostDatabases,
createTrackedJournalOpener
} from './journal-host-database-test-support'

const IDENTITY: AgentSessionJournalIdentity = {
sessionId: 'session-start',
workspaceId: 'ws-1',
hostId: 'host-1',
agent: 'claude',
providerHandle: { kind: 'claude', sessionId: 'native-1', leafUuid: null }
}

const START_FAILED = agentSessionFailureWords(agentSessionFailureFact('providerStartFailed'), {
surface: 'rejection'
})
const ERROR_ROW = { provider: 'orca', clientMessageId: 'start-failure:gen-1' } as const

let root: string
let clock = 1_000
const journals = createTrackedJournalOpener()

beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'orca-queued-rejection-batch-'))
})

afterEach(async () => {
await closeTestJournalHostDatabases()
await rm(root, { recursive: true, force: true })
})

async function openWithQueued(...ids: string[]): Promise<AgentSessionJournal> {
const journal = await journals.open({
identity: IDENTITY,
stateDirectory: root,
now: () => (clock += 1),
mintEpoch: () => 'epoch-1'
})
for (const id of ids) {
await journal.appendSubmission({
clientMessageId: id,
payloadFingerprint: `fp-${id}`,
body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: id }] },
fence: 0,
handoverRecorded: true
})
}
return journal
}

function startFailureBatch(mutations = 1) {
return {
settlementId: 'start-failure:gen-1',
fence: 0,
recovered: true as const,
mutations: Array.from({ length: mutations }, () => ({
kind: 'item' as const,
identity: ERROR_ROW,
body: { kind: 'status' as const, tone: 'error' as const, text: 'Claude did not start.' },
turnScope: AGENT_JOURNAL_THREAD_SCOPE
})),
rejectsQueued: START_FAILED
}
}

it('writes the rejections first and the row after them, and draws the messages above it', async () => {
const journal = await openWithQueued('first', 'second')
const before = journal.cursor().sequence

await journal.appendLifecycleBatch(startFailureBatch())

expect(journal.cursor().sequence).toBe(before + 3)
expect(journal.submissions().map((entry) => entry.dispatchState)).toEqual([
'rejected',
'rejected'
])
const order = journal.snapshot().items.map((item) => item.itemId)
expect(order).toEqual([
agentJournalSubmissionKey('first'),
agentJournalSubmissionKey('second'),
'orca:start-failure%3Agen-1'
])
})

it('writes neither when the row cannot be written', async () => {
const journal = await openWithQueued('first')
const before = journal.cursor().sequence

// An empty batch breaks the row's bound, so the transaction rolls back as a whole.
await expect(journal.appendLifecycleBatch(startFailureBatch(0))).rejects.toThrow(
'journal_lifecycle_batch_mutation_bound_exceeded'
)

expect(journal.cursor().sequence).toBe(before)
expect(journal.submissions().map((entry) => entry.dispatchState)).toEqual(['pending'])
})

// A Stop that reaches the lane first takes the message back; the failed start then failed no one.
it('writes nothing when a Stop withdrew every queued message first', async () => {
const journal = await openWithQueued('first')
const withdrawal = agentSessionFailureWords(agentSessionFailureFact('cancelled'), {
surface: 'rejection'
})

await Promise.all([
journal.rejectQueuedSubmissions(0, withdrawal),
journal.appendLifecycleBatch(startFailureBatch())
])

expect(journal.submissions()[0]?.rejection).toEqual({ kind: 'cancelled' })
expect(journal.snapshot().items.map((item) => item.itemId)).toEqual([
agentJournalSubmissionKey('first')
])
})
Original file line number Diff line number Diff line change
Expand Up @@ -551,7 +551,7 @@ describe('submission and dispatch state machine', () => {
expect(state.receipts.get('cm_1')).toBeTruthy()
})

it('keeps a refused write rejected and leaves its bubble where it was', () => {
it('keeps a refused write rejected, at its rejection, whatever comes after', () => {
const state = fold([
submission,
{
Expand Down Expand Up @@ -579,7 +579,8 @@ describe('submission and dispatch state machine', () => {
submittedAt: submission.ts,
reason: 'provider_write_failed: closed before enqueue'
})
expect(renderJournalState(state).items[0]?.sequence).toBe(submission.seq)
// It sits where it was rejected; the late `pending` moves nothing.
expect(renderJournalState(state).items[0]?.sequence).toBe(2)
})

it('ignores a dispatch for a submission this epoch never saw', () => {
Expand Down
6 changes: 3 additions & 3 deletions src/main/native-chat/agent-session-journal/journal-reducer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,9 @@
// dropped rather than resurrecting stale content, and ordering is by the
// position (sequence, then place in the row) of the write that CREATED an item
// (a later revision updates the body, it does not move the bubble) — except a
// queued message, which sits where its handover put it. Producer linkage is
// likewise the creating write's: a revision naming no producer keeps it, one
// naming any replaces it.
// queued message, which sits where its handover put it, and a rejected one, which
// sits where it was rejected. Producer linkage is likewise the creating write's: a
// revision naming no producer keeps it, one naming any replaces it.

import type {
AgentJournalAcceptanceReceipt,
Expand Down
Loading
Loading