Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -148,11 +148,13 @@ function comparable(merged: Drawn, fresh: Drawn): { merged: string[]; fresh: str
}

/** Every frame from each subscribe point in `starts`, `rowsPerFrame` rows at a time; with how many
* of those frames the merged state held only a window of the chat. */
* of those frames the merged state held only a window of the chat. `kept`: the never-opened send
* became a card, so from its rejection on neither list draws it nor its stop row. */
function compareFrames(
journal: StopJournal,
starts: readonly number[],
rowsPerFrame: number
rowsPerFrame: number,
kept = false
): { differing: string[]; windowed: number } {
const last = journal.rows.length
const fresh = new Map<number, Drawn>()
Expand All @@ -165,15 +167,19 @@ function compareFrames(
fresh.set(upTo, drawnThen)
return drawnThen
}
// Whether a list draws the never-opened send as the Stop left it: its stop row, or for a kept
// send, nothing of it at all.
const asStopped = (ids: readonly string[]): boolean =>
kept ? !ids.includes(STOP_ROW) && !ids.includes(`orca:${NEVER_OPENED}`) : ids.includes(STOP_ROW)
const freshListing = new Map<number, boolean>()
const freshListed = (upTo: number): boolean => {
const cached = freshListing.get(upTo)
if (cached !== undefined) {
return cached
}
const shown = listed(
apply(EMPTY_STRUCTURED_AGENT_SESSION, [journal.snapshotAt(upTo)])
).includes(STOP_ROW)
const shown = asStopped(
listed(apply(EMPTY_STRUCTURED_AGENT_SESSION, [journal.snapshotAt(upTo)]))
)
freshListing.set(upTo, shown)
return shown
}
Expand All @@ -191,7 +197,7 @@ function compareFrames(
if (JSON.stringify(compared.merged) !== JSON.stringify(compared.fresh)) {
differing.push(`subscribed at ${start}, frame through ${upTo}`)
}
if (upTo >= journal.takenBack && !listed(merged).includes(STOP_ROW)) {
if (upTo >= journal.takenBack && !asStopped(listed(merged))) {
differing.push(`subscribed at ${start}, no stop row in the list through ${upTo}`)
}
if (upTo >= journal.takenBack && !freshListed(upTo)) {
Expand Down Expand Up @@ -231,6 +237,29 @@ describe.each(['after-end', 'during-turn'] as const)(
}
)

// The same chat, but the never-opened send was still held by the host when the Stop came: the
// card holds its words, so the phone draws no bubble and no stop row for it, from any frame, and the
// desktop's transcript (which draws other refused sends in place) draws none either.
describe('a short chat whose last send a Stop kept as a card: merged frames draw as a fresh snapshot does', () => {
const journal = stopJournal(8, 'after-end', 'kept')

it('draws nothing of the kept send on the phone or the desktop', () => {
const state = apply(EMPTY_STRUCTURED_AGENT_SESSION, [journal.snapshotAt(journal.rows.length)])
expect(drawn(state).rows.map((row) => row.id)).not.toContain(`orca:${NEVER_OPENED}`)
const desktop = projectStructuredAgentSessionMessages(state.items, [], state.submissions, {
rejectedInPlace: true
})
expect(desktop.map((message) => message.id)).not.toContain(`orca:${NEVER_OPENED}`)
expect(desktop.some((message) => message.stoppedBeforeStart)).toBe(false)
})

it.each([1, 2, 3])('from every subscribe point, %i row(s) per frame', (rowsPerFrame) => {
expect(
compareFrames(journal, range(2, journal.rows.length), rowsPerFrame, true).differing
).toEqual([])
})
})

describe('a chat longer than the phone first loads', () => {
const journal = stopJournal(70)
const last = journal.rows.length
Expand Down
24 changes: 19 additions & 5 deletions mobile/src/session/mobile-native-chat-stop-journal.test-fixture.ts
Original file line number Diff line number Diff line change
Expand Up @@ -42,9 +42,14 @@ export type StopJournal = HostJournalFrames & {
* that holds sends while Stopping, as the QA journal shows), or while the turn still runs. */
export type SendWhileStoppingWritten = 'after-end' | 'during-turn'

/** The last send, never opened: handed over and taken back at the child's exit, or still held by
* the host when the Stop came, which kept it as a card. */
export type NeverOpenedSend = 'taken-back' | 'kept'

export function stopJournal(
historyTurns: number,
sendWhileStopping: SendWhileStoppingWritten = 'after-end'
sendWhileStopping: SendWhileStoppingWritten = 'after-end',
neverOpened: NeverOpenedSend = 'taken-back'
): StopJournal {
const rows: JournalRow[] = []
const stopping: StopJournalStopping[] = []
Expand Down Expand Up @@ -228,19 +233,28 @@ export function stopJournal(
openTurn('sent-while-stopping', 't-sent-while-stopping')
endTurn('sent-while-stopping', 't-sent-while-stopping', 'completed')
// A send the provider never opened; a turn-less Stop ends the child, whose exit takes it back.
// One the host still held is kept as a card ahead of the Stop's event.
const neverOpenedSent = submit(NEVER_OPENED)
handOver(NEVER_OPENED)
const stopNeverOpened = stopEvent()
const kept = neverOpened === 'kept'
if (!kept) {
handOver(NEVER_OPENED)
}
const stopNeverOpened = kept ? 0 : stopEvent()
const takenBack = add({
kind: 'dispatch',
clientMessageId: NEVER_OPENED,
state: 'rejected',
providerItemId: null,
reason: DISPATCH_REJECTED_CANCELLED,
rejection: { kind: 'cancelled' },
recovered: true
recovered: true,
...(kept ? { keptAsQueuedMessageId: NEVER_OPENED } : {})
})
stopping.push({ from: stopNeverOpened, until: takenBack - 1 })
if (kept) {
stopEvent()
} else {
stopping.push({ from: stopNeverOpened, until: takenBack - 1 })
}
// The next send starts a new child under a new fence.
fence = 2
submit('recovery')
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -63,47 +63,6 @@ export async function markJournalPendingSubmissionsUnknown(
return unresolved.map((entry) => entry.clientMessageId)
}

/** Settles as `rejected` every submission a child handed over and never echoed, when that child
* ended in its start: one that never answered initialize ran nothing, so each is safe to send
* again. A queued submission was never handed to that child; the delivery loop settles it. */
export async function rejectJournalPendingSubmissions(
journal: AgentSessionJournal,
fence: number,
rejection: AgentJournalDispatchRejection
): Promise<string[]> {
const unwritten = journalPendingSubmissionResolutions(journal.submissions(), fence, { rejection })
for (const resolution of unwritten) {
await journal.resolveDispatch(resolution)
}
return unwritten.map((entry) => entry.clientMessageId)
}

/** Rejects queued submissions — accepted, never handed over, so provably unwritten. */
export async function rejectJournalQueuedSubmissions(
journal: AgentSessionJournal,
fence: number,
rejection: AgentJournalDispatchRejection,
which: (submission: AgentJournalSubmission) => boolean = () => true
): Promise<string[]> {
const queued = journal
.submissions()
.filter((entry) => isQueuedAgentJournalSubmission(entry) && which(entry))
// Issued together, so the fold shows none of them queued once this call returns: a Stop decides
// whether anything is working from it without awaiting the withdrawal.
await Promise.all(
queued.map((entry) =>
journal.resolveDispatch({
clientMessageId: entry.clientMessageId,
state: 'rejected',
...rejection,
fence,
recovered: true
})
)
)
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(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import {
type AgentSessionJournalIdentity
} from '../../../shared/agent-session-journal-types'
import type { AgentSessionJournal } from './journal-store'
import { holdUnsentSends } from './journal-unsent-send-hold'
import {
closeTestJournalHostDatabases,
createTrackedJournalOpener
Expand Down Expand Up @@ -114,12 +115,8 @@ it('writes neither when the row cannot be written', async () => {
// 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),
holdUnsentSends(journal, { fence: 0, hostInstance: 'proc-1', hold: { cause: 'cancelled' } }),
journal.appendLifecycleBatch(startFailureBatch())
])

Expand Down
24 changes: 1 addition & 23 deletions src/main/native-chat/agent-session-journal/journal-store.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
// Append-only journal store for one agent session. It owns the chat's fold and write queue, and no
// connection: every statement goes through the host's one journal database.

import type { AgentJournalDispatchRejection } from '../../../shared/agent-session-failure-words'
import { randomUUID } from 'node:crypto'
import type {
AgentJournalAcceptanceReceipt,
Expand Down Expand Up @@ -29,11 +28,7 @@ import type { JournalReplacementItem } from './journal-epoch-replacement'
import { readJournalSince } from './journal-cursor'
import type { JournalHostDatabase } from './journal-host-database'
import { journalRowsAfterReader, type JournalLoad } from './journal-open'
import {
markJournalPendingSubmissionsUnknown,
rejectJournalPendingSubmissions,
rejectJournalQueuedSubmissions
} from './journal-pending-submission-recovery'
import { markJournalPendingSubmissionsUnknown } from './journal-pending-submission-recovery'
import {
applyJournalRow,
createJournalReducerState,
Expand Down Expand Up @@ -355,23 +350,6 @@ export class AgentSessionJournal {
return markJournalPendingSubmissionsUnknown(this, fence, reason)
}

/** Reject sends a child that ended in its start was handed and never echoed: none ran. */
async rejectPendingSubmissions(
fence: number,
rejection: AgentJournalDispatchRejection
): Promise<string[]> {
return rejectJournalPendingSubmissions(this, fence, rejection)
}

/** Reject sends accepted but never handed over, optionally only those `which` names. */
async rejectQueuedSubmissions(
fence: number,
rejection: AgentJournalDispatchRejection,
which?: (submission: AgentJournalSubmission) => boolean
): Promise<string[]> {
return rejectJournalQueuedSubmissions(this, fence, rejection, which)
}

/** The escape hatch for a forked handle and an unreadable schema. It invalidates every cursor;
* clients reload. */
async rollEpoch(reason: AgentJournalEpochReason, fence: number): Promise<AgentJournalCursor> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -445,6 +445,68 @@ describe('a close of the chat', () => {
})
})

describe("a person's Stop", () => {
const CANCELLED = agentSessionFailureWords(agentSessionFailureFact('cancelled'), {
surface: 'rejection'
})

it('keeps every queued text of a person by the same rule, ahead of the cards already queued', async () => {
const image: AgentJournalMessageItem = {
kind: 'message',
role: 'user',
blocks: [
{ type: 'text', text: 'look' },
{ type: 'image-ref', path: '/tmp/attachment.png' }
]
}
const journal = await open()
await journal.queuedMessages.insert({
messageId: 'queued-card',
body: message('queued card'),
fingerprint: fingerprint(message('queued card')),
hostInstance: HOST
})
await accept(journal, 'first', { origin: 'client', source: USER_MESSAGE_SOURCE })
await accept(journal, 'mail', { origin: 'host', source: MAIL_SOURCE })
await accept(journal, 'compact', {
origin: 'client',
source: USER_MESSAGE_SOURCE,
body: structuredAgentSessionCompactBody()
})
await accept(journal, 'image', { origin: 'client', source: USER_MESSAGE_SOURCE, body: image })
await accept(journal, 'second', { origin: 'client', source: USER_MESSAGE_SOURCE })
await hold(journal, { cause: 'cancelled' })

expect(cardOrder(journal)).toEqual(['first', 'second', 'queued-card'])
for (const id of ['first', 'second']) {
expect(journal.submission(id)).toMatchObject({
dispatchState: 'rejected',
...CANCELLED,
keptAsQueuedMessageId: id
})
}
for (const id of ['mail', 'compact', 'image']) {
expect(journal.submission(id)).toMatchObject({ dispatchState: 'rejected', ...CANCELLED })
expect(journal.submission(id)).not.toHaveProperty('keptAsQueuedMessageId')
}
})

// The Stop decides whether the agent still works from the fold, without awaiting the hold.
it('leaves none of them queued in the fold by the time the call returns', async () => {
const journal = await open()
await accept(journal, 'first', { origin: 'client', source: USER_MESSAGE_SOURCE })
await accept(journal, 'second', { origin: 'client', source: USER_MESSAGE_SOURCE })
const settled = hold(journal, { cause: 'cancelled' })

expect(journal.submissions().map((entry) => entry.dispatchState)).toEqual([
'rejected',
'rejected'
])
await settled
expect(cardOrder(journal)).toEqual(['first', 'second'])
})
})

describe('the order kept cards stand in', () => {
// Sequences restart with each epoch, so a card kept before a rewind has no sequence to compare.
it('a card kept before the epoch rolled stays ahead of one kept after it', async () => {
Expand Down
Loading
Loading