Skip to content
Open
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
4 changes: 2 additions & 2 deletions src/main/agent-hooks/first-work-branch-rename.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@ describe('maybeAutoRenameBranchOnFirstWork', () => {
lastActivityAt: () => 1,
cursor: () => ({ epoch: 1, sequence: (sequence += 1) }),
itemFence: () => undefined,
wroteBeforeOpen: () => false,
acceptedBeforeOpen: () => false,
runningTurn: () => null
} as unknown as AgentSessionJournal
const pending: Promise<void>[] = []
Expand Down Expand Up @@ -213,7 +213,7 @@ describe('maybeAutoRenameBranchOnFirstWork', () => {
submissions: []
}),
itemFence: () => undefined,
wroteBeforeOpen: () => false,
acceptedBeforeOpen: () => false,
runningTurn: () => null
} as unknown as AgentSessionJournal
const location = {
Expand Down
2 changes: 1 addition & 1 deletion src/main/claude/claude-background-stop-records.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ async function droppedFromRoster() {
runningTurn: () => null,
itemFence: () => undefined,
visitItems: () => undefined,
wroteBeforeOpen: () => false
acceptedBeforeOpen: () => false
},
adapter: run.adapter
},
Expand Down
2 changes: 1 addition & 1 deletion src/main/claude/claude-background-wake-retention.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ async function wiredSession() {
stopMarks: { latest: () => null, revision: () => 0 },
// One generation produced everything here, and every send was accepted by this process.
itemFence: () => undefined,
wroteBeforeOpen: () => false,
acceptedBeforeOpen: () => false,
// The live generation's own turns are never a died turn, so this is never consulted for one.
runningTurn: () => null,
snapshot: () => ({
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,7 @@ function expectLifecycleBatchBounds(batches: readonly LifecycleBatch[]): void {
const appender = new JournalLifecycleBatchAppender({
state: () => state,
cursor: () => ({ epoch: state.epoch, sequence: state.lastSequence }),
acceptedBeforeOpen: () => false,
enqueueRows: async () => []
})
for (const batch of batches) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,7 @@ export function registerJournalLifecycleBatchAppenderTests(): void {
const appender = new JournalLifecycleBatchAppender({
state: () => state,
cursor: () => ({ epoch: state.epoch, sequence: state.lastSequence }),
acceptedBeforeOpen: () => false,
enqueueRows: async (plan) => {
const built = plan().map((build, index) => build(index + 1, 1_000))
rows.push(...built)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ export class JournalLifecycleBatchAppender {
private readonly deps: {
state: () => JournalReducerState
cursor: () => AgentJournalCursor
acceptedBeforeOpen: (clientMessageId: string) => boolean
enqueueRows: (
plan: () => readonly ((seq: number, ts: number) => JournalRow)[]
) => Promise<JournalRow[]>
Expand All @@ -40,7 +41,8 @@ export class JournalLifecycleBatchAppender {
const rejections = journalQueuedRejectionRowBuilders(
this.deps.state,
input.fence,
input.rejectsQueued
input.rejectsQueued,
this.deps.acceptedBeforeOpen
)
if (rejections.length > 0 && input.mutations.length === 0) {
throw new Error('journal_lifecycle_batch_mutation_bound_exceeded')
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -104,14 +104,19 @@ 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. */
/** Rows rejecting every submission this handle still has queued, read from `state` when called:
* for an append that must carry them with what follows, in one transaction. A send an earlier
* handle accepted is the restart hold's to settle, never rejected with this handle's cause. */
export function journalQueuedRejectionRowBuilders(
state: () => JournalReducerState,
fence: number,
rejection: AgentJournalDispatchRejection
rejection: AgentJournalDispatchRejection,
acceptedBeforeOpen: (clientMessageId: string) => boolean
): ((seq: number, ts: number) => JournalRow)[] {
return [...state().submissions.values()].filter(isQueuedAgentJournalSubmission).map((entry) =>
const queued = [...state().submissions.values()].filter(
(entry) => isQueuedAgentJournalSubmission(entry) && !acceptedBeforeOpen(entry.clientMessageId)
)
return queued.map((entry) =>
journalDispatchRowBuilder(state, {
clientMessageId: entry.clientMessageId,
state: 'rejected',
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ export class JournalStopAcceptor {
private readonly deps: {
writer: Pick<JournalRowWriter, 'enqueueRows'>
state: () => JournalReducerState
acceptedBeforeOpen: (clientMessageId: string) => boolean
}
) {}

Expand All @@ -60,7 +61,12 @@ export class JournalStopAcceptor {
turnId = input.event.turnId ?? input.liveTurnId()
const event = { ...input.event, ...(turnId !== null ? { turnId } : {}) }
return [
...journalQueuedRejectionRowBuilders(state, input.fence, STOP_WITHDRAWAL),
...journalQueuedRejectionRowBuilders(
state,
input.fence,
STOP_WITHDRAWAL,
this.deps.acceptedBeforeOpen
),
journalStopEventRowBuilder(state, event, input.fence)
]
},
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -121,9 +121,12 @@ export function createJournalStoreCollaborators(host: JournalStoreHost): Journal
rolledBack: () => queuedMessages.invalidate(),
streamedTextRevision: (row) => isJournalStreamedTextRevision(host.state(), row)
})
const acceptedBeforeOpen = (clientMessageId: string) =>
host.journal().acceptedBeforeOpen(clientMessageId)
const lifecycleBatchAppender = new JournalLifecycleBatchAppender({
state: host.state,
cursor: host.cursor,
acceptedBeforeOpen,
enqueueRows: (plan) => rowWriter.enqueueRows(plan)
})
return {
Expand All @@ -135,7 +138,7 @@ export function createJournalStoreCollaborators(host: JournalStoreHost): Journal
writer: rowWriter,
cards: queuedMessages
}),
stops: new JournalStopAcceptor({ writer: rowWriter, state: host.state }),
stops: new JournalStopAcceptor({ writer: rowWriter, state: host.state, acceptedBeforeOpen }),
// Behind the stored fact: settles drafts whose consumed submission the loaded journal shows
// refused (a downgrade wrote no hook), then prunes. Bookkeeping, never failing the open.
restore: () =>
Expand Down
20 changes: 6 additions & 14 deletions src/main/native-chat/agent-session-journal/journal-store.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -483,7 +483,7 @@ export function registerJournalStoreTests(): void {
handoverRecorded: true as const
})

it('names rows an earlier handle wrote, and never a row of a later epoch', async () => {
it('names a send an earlier handle queued, still after a history replacement carries it', async () => {
const earlier = await open()
await earlier.appendItem(item(1), body('one'), {
fence: 1,
Expand All @@ -493,21 +493,13 @@ export function registerJournalStoreTests(): void {
await earlier.close()

const journal = await open()
const leftover = journal.submissions().find((entry) => entry.clientMessageId === 'earlier')
expect(journal.wroteBeforeOpen(leftover?.acceptedSequence)).toBe(true)

await journal.resolveDispatch({
clientMessageId: 'earlier',
state: 'unknown',
reason: 'unconfirmed from an earlier host',
fence: 1
})
// Sequences restart with an epoch, so a row accepted after it can sit below the open cursor.
expect(journal.acceptedBeforeOpen('earlier')).toBe(true)
// The replacement re-numbers the carried send in a new epoch; its id is what still names it.
await journal.replaceEpochItems('handle_forked', 1, [])
await journal.appendSubmission(submission('later'))
const later = journal.submissions().find((entry) => entry.clientMessageId === 'later')
expect(later?.acceptedSequence).toBeLessThanOrEqual(2)
expect(journal.wroteBeforeOpen(later?.acceptedSequence)).toBe(false)
expect(journal.submission('earlier')).toMatchObject({ dispatchState: 'pending' })
expect(journal.acceptedBeforeOpen('earlier')).toBe(true)
expect(journal.acceptedBeforeOpen('later')).toBe(false)
})
})
}
21 changes: 11 additions & 10 deletions src/main/native-chat/agent-session-journal/journal-store.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import type {
AgentSessionJournalIdentity
} from '../../../shared/agent-session-journal-types'
import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key'
import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission'
import type { AgentSessionContextUsage } from '../../../shared/agent-session-context-usage'
import { latestStructuredAgentContextFacts } from '../../../shared/structured-agent-session-context-usage'
import {
Expand Down Expand Up @@ -86,7 +87,7 @@ export class AgentSessionJournal {
private readonly mintEpoch: () => string

private state: JournalReducerState
private openedThrough: AgentJournalCursor = { epoch: '', sequence: 0 }
private queuedAtOpen: ReadonlySet<string> = new Set()
private reopenUnmarked: AgentJournalCursor | null = null
private onCommitted: (() => void) | null = null
private readonly queue: JournalWriteQueue
Expand Down Expand Up @@ -157,14 +158,10 @@ export class AgentSessionJournal {
return this.identity.agent
}

/** Whether a row at this sequence was on disk when this handle opened, so an earlier handle
* wrote it. Sequences restart with each epoch, so a row of a later epoch never was. */
wroteBeforeOpen(sequence: number | undefined): boolean {
return (
sequence !== undefined &&
this.state.epoch === this.openedThrough.epoch &&
sequence <= this.openedThrough.sequence
)
/** Whether this send was already queued when this handle opened, so an earlier handle accepted
* it. Keyed by id: a history replacement carries a queued send under a new sequence. */
acceptedBeforeOpen(clientMessageId: string): boolean {
return this.queuedAtOpen.has(clientMessageId)
}

/** Where the reopen's pause begins when this handle could not write its mark: where the mark
Expand All @@ -175,7 +172,11 @@ export class AgentSessionJournal {

async open(): Promise<void> {
await this.restore()
this.openedThrough = this.cursor()
this.queuedAtOpen = new Set(
this.submissions()
.filter(isQueuedAgentJournalSubmission)
.map((entry) => entry.clientMessageId)
)
}

/** Refuses every later write and resolves once the admitted ones have landed. Holds no
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,10 @@ export type UnsentSendHold =
/** A close of the chat; `which` narrows it to what a close that did not complete closed. */
| { cause: 'chatClosed'; which?: (submission: AgentJournalSubmission) => boolean }

export function unsentSendRejection(cause: UnsentSendHold['cause']) {
return agentSessionFailureWords(agentSessionFailureFact(cause), { surface: 'rejection' })
}

/** The body an unsent send is kept with, or null when it is rejected instead:
* - a card's own hand-off: its rejection already returns the card (`rejectedDraftSettlement`);
* - `/compact` and other commands: a command in flight is not resumed, the person re-runs it;
Expand Down Expand Up @@ -97,7 +101,7 @@ export async function holdUnsentSends(
? isUnansweredHandedOverSubmission(entry)
: isQueuedAgentJournalSubmission(entry) &&
(hold.cause === 'hostRestarted'
? journal.wroteBeforeOpen(entry.acceptedSequence)
? journal.acceptedBeforeOpen(entry.clientMessageId)
: (hold.which?.(entry) ?? true))
)
.sort((a, b) => (a.acceptedSequence ?? 0) - (b.acceptedSequence ?? 0))
Expand All @@ -113,9 +117,7 @@ export async function holdUnsentSends(
)
}))
const positions = headOfQueuePositions(journal, kept)
const rejection = agentSessionFailureWords(agentSessionFailureFact(hold.cause), {
surface: 'rejection'
})
const rejection = unsentSendRejection(hold.cause)
const failures: unknown[] = []
for (const [index, { submission, body }] of kept.entries()) {
const { clientMessageId } = submission
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -454,7 +454,7 @@ export function registerQueuedMessageStoreTests(): void {
journal = await open()
expect(journal.queuedMessages.get('draft-1')?.state).toBe('dispatched')
await journal.rejectQueuedSubmissions(0, HOST_RESTARTED, (submission) =>
journal.wroteBeforeOpen(submission.acceptedSequence)
journal.acceptedBeforeOpen(submission.clientMessageId)
)
// Whoever sent it, it waits with no hold of its own, under the reopen's pause.
expect(journal.queuedMessages.get('draft-1')).toMatchObject({
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ describe('the stop-note refresh frame', () => {
}
},
submissions: () => [],
wroteBeforeOpen: () => false
acceptedBeforeOpen: () => false
},
2
)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,13 @@
// acceptance commits, the Stop acts only while that target still stands, read again from state,
// so a newer turn or a newer child is never this Stop's.

import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission'
import type { StructuredAgentSessionProviderChild } from './structured-agent-session-host-types'
import { stopReachesUnrecordedWork } from './structured-agent-session-queued-stop'
import { isMainAgentWorking } from './structured-agent-session-turns-cancel'
import { contextStructuredAgentSessionCurrentWork } from './structured-agent-session-current-work'
import {
contextStructuredAgentSessionCurrentWork,
structuredAgentSessionHandsOver
} from './structured-agent-session-current-work'
import {
structuredAgentSessionStopNamesEndedTurn,
structuredAgentSessionStoppedTurnId
Expand Down Expand Up @@ -52,7 +54,9 @@ export function captureChatStopTarget(
if (reached === 'late') {
return null
}
const queued = ctx.journal.submissions().some(isQueuedAgentJournalSubmission)
const queued = ctx.journal
.submissions()
.some((submission) => structuredAgentSessionHandsOver(ctx.journal, submission))
const target = {
child,
turnId: structuredAgentSessionStoppedTurnId(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ import {
} from '../../../shared/agent-session-journal-types'
import type { AgentSessionConversationCommand } from '../../../shared/agent-session-conversation-command'
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission'
import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view'
import { agentSessionRefusalReference } from '../../../shared/agent-session-wire-refusals'
import {
Expand All @@ -46,6 +47,7 @@ import { structuredAgentSessionCommandTurn } from '../../../shared/structured-ag
import { refuseQueuedCommand } from './structured-agent-session-queued-command-refusal'
import {
contextStructuredAgentSessionCurrentWork,
structuredAgentSessionHandsOver,
type StructuredAgentSessionCurrentWork
} from './structured-agent-session-current-work'

Expand All @@ -65,13 +67,14 @@ export function structuredAgentSessionCompactBody(): AgentJournalMessageItem {

type AwaitedSubmission = Pick<
AgentJournalSubmission,
'clientMessageId' | 'dispatchState' | 'acceptedSequence'
'clientMessageId' | 'dispatchState' | 'acceptedSequence' | 'handoverRecorded' | 'handedOverAt'
>

/** Optional `submissions` as the dead-generation journal reads it. */
export type StructuredAgentSessionAwaitedCommandJournal = {
submissions?: () => readonly AwaitedSubmission[]
itemBody: AgentSessionJournal['itemBody']
acceptedBeforeOpen: AgentSessionJournal['acceptedBeforeOpen']
}

/** The command the oldest message still waiting on the provider names: a start that fails now
Expand All @@ -83,6 +86,8 @@ export function structuredAgentSessionAwaitedCommand(
for (const submission of journal.submissions?.() ?? []) {
if (
submission.dispatchState === 'pending' &&
(!isQueuedAgentJournalSubmission(submission) ||
structuredAgentSessionHandsOver(journal, submission)) &&
(oldest === undefined || (submission.acceptedSequence ?? 0) < (oldest.acceptedSequence ?? 0))
) {
oldest = submission
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -124,8 +124,8 @@ export async function openStructuredAgentSessionConversationJournal(
hold: { cause: 'hostRestarted' }
})
} catch (error) {
// The row stays queued. The delivery loop's first step tries again before it hands anything
// over; if that fails too, the loop fails and rejects every queued send.
// The row stays queued, never handed over (`handsOver`); the delivery loop's next step tries
// again.
deps.logger.warn('settling sends an earlier process left queued failed on open', {
scope: 'open-leftover-sends',
sessionId,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ import { structuredAgentSessionHandingOver } from './structured-agent-session-ha

export type StructuredAgentSessionCurrentWorkJournal = Pick<
AgentSessionJournal,
'runningTurn' | 'itemFence' | 'visitItems' | 'submissions' | 'wroteBeforeOpen' | 'item' | 'cursor'
'runningTurn' | 'itemFence' | 'visitItems' | 'submissions' | 'acceptedBeforeOpen' | 'item' | 'cursor'
>

/** What the host knows of who runs the chat now. */
Expand Down Expand Up @@ -104,17 +104,18 @@ export function structuredAgentSessionReplacedRuntimeEnded(
)
}

/** `StructuredAgentSessionCurrentWork.handsOver`, for a reader that needs no other answer. */
function structuredAgentSessionHandsOver(
journal: Pick<AgentSessionJournal, 'wroteBeforeOpen'>,
/** `StructuredAgentSessionCurrentWork.handsOver`, for a reader that needs no other answer: the one
* rule for whether a queued send is this handle's to deliver, withdraw or wait on. */
export function structuredAgentSessionHandsOver(
journal: Pick<AgentSessionJournal, 'acceptedBeforeOpen'>,
submission: Pick<
AgentJournalSubmission,
'handoverRecorded' | 'dispatchState' | 'handedOverAt' | 'acceptedSequence'
'handoverRecorded' | 'dispatchState' | 'handedOverAt' | 'clientMessageId'
>
): boolean {
return (
isQueuedAgentJournalSubmission(submission) &&
!journal.wroteBeforeOpen(submission.acceptedSequence)
!journal.acceptedBeforeOpen(submission.clientMessageId)
)
}

Expand Down Expand Up @@ -217,7 +218,7 @@ export class StructuredAgentSessionCurrentWork {

/** Whether this process may hand this send over: accepted here, never by an earlier host process.
* Derived from where the row sits, so nothing has to be written first; keeping an earlier
* process's send as a card is the reconciliation pass's bookkeeping (`holdUnsentSends`). */
* process's send as a card is bookkeeping (`holdUnsentSends`) whose failure gates nothing. */
handsOver(submission: AgentJournalSubmission): boolean {
return structuredAgentSessionHandsOver(this.journal, submission)
}
Expand Down
Loading
Loading