Skip to content
Merged
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
1 change: 1 addition & 0 deletions config/scripts/vitest-sqlite-runtime-files.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,7 @@ export const SQLITE_RUNTIME_INCLUDE = [
'src/main/native-chat/agent-session-wire/structured-agent-session-wedged-profile-migration.test.ts',
'src/main/native-chat/agent-session-wire/structured-agent-session-wire-admission.test.ts',
'src/main/native-chat/agent-session-wire/structured-conversation-command.test.ts',
'src/main/native-chat/agent-session-wire/structured-conversation-compaction.test.ts',
'src/main/opencode/opencode-acp-stored-messages.test.ts',
'src/main/orcad/orcad-automations.test.ts',
'src/main/orcad/orcad-local-serve-selection.test.ts',
Expand Down
1 change: 0 additions & 1 deletion src/main/native-chat/agent-session-record-test-fixture.ts
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@ export function agentSessionRecordFixture(
leaseDeadlineAt: 0,
lastRenewedAt: 0,
handoffOperationId: null,
journalCheckpoint: null,
claimKeyId: 'test-key',
claimStatus: 'released',
unreconciled: false,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -870,7 +870,7 @@ describe('a close that stops the child and then a wind-down step fails', () => {
)
})

describe('a compaction or rewind an earlier child left prepared', () => {
describe('a rewind an earlier child left prepared', () => {
async function leftPrepared(prepare: (fence: number) => Promise<unknown>): Promise<void> {
await host.close(SESSION, 'evict')
await prepare(store.getRecord(SESSION)!.lease.runtimeFence)
Expand All @@ -879,25 +879,6 @@ describe('a compaction or rewind an earlier child left prepared', () => {
await startHost()
}

it("ignores an older build's interrupted compaction, so a send is accepted and delivered (R16)", async () => {
await leftPrepared((fence) =>
store.setConversationCommand(SESSION, fence, {
command: 'compact',
runtimeFence: fence,
operationId: 'compact-op',
callerKey: 'client-1',
phase: 'prepared',
state: 'unknown'
})
)

const id = await accept('after the compaction')

await eventually(async () => expect((await submission(id))?.dispatchState).toBe('accepted'))
// Nothing settles the record: it belongs to a child this host no longer runs.
expect(store.getRecord(SESSION)?.conversationCommand).toMatchObject({ phase: 'prepared' })
})

it('completes a rewind the provider already applied at open, so a send is accepted (R16)', async () => {
await leftPrepared((fence) =>
persistRewindRecord(store, SESSION, fence, {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,6 @@ function crashedClaudeRecord(): AgentSessionRecord {
leaseDeadlineAt: LAST_RENEWED_AT + 30_000,
lastRenewedAt: LAST_RENEWED_AT,
handoffOperationId: null,
journalCheckpoint: null,
claimKeyId: 'key-1',
claimStatus: 'live',
unreconciled: false,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,6 @@ function ownerRecord(): AgentSessionRecord {
leaseDeadlineAt: 100,
lastRenewedAt: 1,
handoffOperationId: null,
journalCheckpoint: null,
claimKeyId: 'fixture-key',
claimStatus: 'live',
unreconciled: false,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,6 @@ function reservedRecord(): AgentSessionRecord {
leaseDeadlineAt: NOW + 30_000,
lastRenewedAt: NOW,
handoffOperationId: 'op-1',
journalCheckpoint: null,
claimKeyId: 'key-1',
claimStatus: 'reserved',
unreconciled: false,
Expand Down
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
// The one queue gate: admission, the drain step and Send-now consume a single
// typed hold decision, so the lists cannot drift — pinned here with a clear in
// doubt, Send-now's override set, and the replay-preference rule for a refused
// draft.
// typed hold decision, so the lists cannot drift — pinned here with Send-now's
// override set and the replay-preference rule for a refused draft.

import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types'
Expand Down Expand Up @@ -43,26 +42,6 @@ const settleRejected: QueuedMessageTestRig['settleRejected'] = (...args) =>
rig.settleRejected(...args)

describe('the one queue gate', () => {
it('a clear an older build left prepared holds nothing: Send-now sends the draft', async () => {
await workingSend()
const queued = await send('queued behind the clear', 'queue-if-active').result
if (!queued.ok || !('queued' in queued.value)) {
throw new Error('expected a queued receipt')
}
const draftId = queued.value.queued.messageId
// An unfinished /clear changed nothing the chat reads, so it refuses no send.
await store.setConversationCommand(SESSION, 1, {
command: 'clear',
runtimeFence: 1,
operationId: hostTestOperationId(),
callerKey: CALLER.callerKey,
phase: 'prepared',
state: 'unknown'
})
expect(await sendNow(draftId)).toMatchObject({ ok: true })
await eventually(async () => expect(await rig.handoff(draftId)).toBeDefined())
})

it('Send-now refuses on a pending prompt and overrides a running turn', async () => {
const working = await workingSend()
const queued = await send('queued mid-turn', 'queue-if-active').result
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -41,15 +41,14 @@ let currentRig: QueuedMessageTestRig | undefined
const observationClock = createStructuredSessionObservationClock(() => rig.host, SESSION)
beforeEach(() => observationClock.start())
let host: QueuedMessageTestRig['host']
let store: QueuedMessageTestRig['store']
let dispatch: QueuedMessageTestRig['dispatch']

beforeEach(async () => {
currentRig = undefined
// A Stop ends the agent's session, so the next send starts a child that resumes the thread.
rig = await createQueuedMessageTestRig({ restartable: true })
currentRig = rig
;({ host, store, dispatch } = rig)
;({ host, dispatch } = rig)
})

afterEach(async () => {
Expand Down Expand Up @@ -738,25 +737,6 @@ describe('/clear', () => {
expect(await rig.queuePause(replacementId)).toBeNull()
})

it('a clear an older build left prepared holds no draft: it drains when the turn settles', async () => {
const working = await workingSend()
const queued = await send('behind the clear', 'queue-if-active').result
if (!queued.ok || !('queued' in queued.value)) {
throw new Error('expected a queued receipt')
}
const draftId = queued.value.queued.messageId
await store.setConversationCommand(SESSION, 1, {
command: 'clear',
runtimeFence: 1,
operationId: hostTestOperationId(),
callerKey: CALLER.callerKey,
phase: 'prepared',
state: 'unknown'
})
await settleAccepted(working, 'a')
await eventually(async () => expect(await rig.handoff(draftId)).toBeDefined())
})

it('a clear with no drafts keeps the existing conversation open', async () => {
const cleared = await clear(hostTestOperationId())
expect(cleared).toMatchObject({ ok: true, value: { command: 'clear', state: 'completed' } })
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,6 @@ export function record(overrides: { chain?: AgentSessionRecord['providerHandleCh
leaseDeadlineAt: NOW,
lastRenewedAt: NOW,
handoffOperationId: null,
journalCheckpoint: null,
claimKeyId: 'key-1',
claimStatus: 'released',
unreconciled: false,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,6 @@ function wedgedRecord(overrides: WedgeOverrides): PersistedAgentSessionRecord {
leaseDeadlineAt: NOW - 9_000,
lastRenewedAt: NOW - 10_000,
handoffOperationId: overrides.handoffOperationId ?? null,
journalCheckpoint: null,
claimKeyId: 'key-1',
claimStatus: overrides.claimStatus,
unreconciled: false,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,11 @@ import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-ses
import type { AgentSessionConversationCommand } from '../../../shared/agent-session-conversation-command'
import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication'
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness'
import {
editPersistedTestAgentSessionStore,
openTestAgentSessionRecordStore,
type PersistedTestAgentSessionStore
} from '../../runtime/agent-session-record-store-test-harness'
import { StructuredAgentSessionHost } from './structured-agent-session-host'
import { claudeAndCodexDeclared } from './structured-agent-session-adapter-router-test-support'
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
Expand Down Expand Up @@ -79,15 +83,21 @@ async function openHost(): Promise<void> {
hosts.push(host)
}

/** A crash and relaunch: the next host opens what the dying one had written, and nothing after. */
async function restartHost(): Promise<void> {
/** A crash and relaunch: the next host opens what the dying one had written, and nothing after.
* `olderBuild` edits that state first, as an older build would have left it. */
async function restartHost(
olderBuild?: (persisted: PersistedTestAgentSessionStore) => void
): Promise<void> {
await store.renewLeases([])
const dying = generationRoot()
generation++
await cp(dying, generationRoot(), {
recursive: true,
filter: (source) => !source.endsWith('.tmp') && !source.includes('.lock')
})
if (olderBuild) {
await editPersistedTestAgentSessionStore(generationRoot(), olderBuild)
}
await openHost()
}

Expand Down Expand Up @@ -347,15 +357,17 @@ describe('same-conversation clear', () => {
})

it('allows independently sending a record with an old-build replacement pointer', async () => {
const source = store.getRecord(HOST_TEST_SESSION)!
await store.setConversationCommand(HOST_TEST_SESSION, source.lease.runtimeFence, {
command: 'clear',
phase: 'committed',
state: 'completed',
operationId: hostTestOperationId(),
callerKey: caller.callerKey,
replacementSessionId: 'old-clear-destination'
await restartHost((persisted) => {
persisted.records[HOST_TEST_SESSION].conversationCommand = {
command: 'clear',
phase: 'committed',
state: 'completed',
operationId: hostTestOperationId(),
callerKey: caller.callerKey,
replacementSessionId: 'old-clear-destination'
}
})
await host.restoreReadableSessions(store.listVisibleSessionIds())
expect(await host.send(caller, sendParams('still this conversation'))).toMatchObject({
ok: true
})
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,10 @@ import {
replaceHostTestState
} from './structured-agent-session-host-test-harness'
import { StructuredAgentSessionHost } from './structured-agent-session-host'
import {
editPersistedTestAgentSessionStore,
openTestAgentSessionRecordStore
} from '../../runtime/agent-session-record-store-test-harness'
import { StructuredAgentRegistry } from './structured-agent-registry'
import { CODEX_STRUCTURED_AGENT } from '../../codex/codex-structured-agent-definition'
import {
Expand Down Expand Up @@ -750,18 +754,26 @@ it('ignores a ledger-only compaction record with no command receipt (B15)', asyn
hostFingerprint: older.envelope.payloadFingerprint,
now: HOST_TEST_NOW
})
await state.store.setConversationCommand(
SESSION,
state.store.getRecord(SESSION)!.lease.runtimeFence,
{
const fence = state.store.getRecord(SESSION)!.lease.runtimeFence
await state.host.close(SESSION, 'evict')
await state.host.flushAllStreamedEvents()
await editPersistedTestAgentSessionStore(state.root, (persisted) => {
persisted.records[SESSION].conversationCommand = {
command: 'compact',
runtimeFence: state.store.getRecord(SESSION)!.lease.runtimeFence,
runtimeFence: fence,
operationId: older.envelope.clientOperationId,
callerKey: CALLER.callerKey,
phase: 'prepared',
state: 'unknown'
}
)
})
const store = await openTestAgentSessionRecordStore(state.root)
replaceHostTestState({
store,
host: new StructuredAgentSessionHost({ ...state.host.deps, store })
})
state = hostTestState()
await state.host.restoreReadableSessions(store.listVisibleSessionIds())

await expect(state.host.send(CALLER, sendParams('still works'))).resolves.toMatchObject({
ok: true
Expand Down
13 changes: 0 additions & 13 deletions src/main/runtime/agent-session-conversation-command-record.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,19 +4,6 @@ import { agentSessionOperationKey } from '../../shared/agent-session-operation-l
import type { AgentSessionStoreState } from './agent-session-store-state'
import type { AgentSessionConversationCommandRecord } from '../../shared/agent-session-conversation-command'

export function commitConversationCommandRecord(
state: AgentSessionStoreState,
sessionId: string,
fence: number,
command: AgentSessionConversationCommandRecord
): void {
const record = state.records.get(sessionId)
if (!record || record.lease.runtimeFence !== fence) {
throw agentSessionRefusalError('agent_session_checkpoint_stale', { reason: 'leaseMoved' })
}
state.records.set(sessionId, { ...record, conversationCommand: command })
}

export type AgentSessionConversationClear = {
sessionId: string
fence: number
Expand Down
1 change: 0 additions & 1 deletion src/main/runtime/agent-session-record-founding.ts
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,6 @@ export function foundAgentSessionRecord(
leaseDeadlineAt: lease.now,
lastRenewedAt: lease.now,
handoffOperationId: null,
journalCheckpoint: null,
claimKeyId: lease.claimKeyId,
claimStatus: 'released',
unreconciled: false,
Expand Down
5 changes: 3 additions & 2 deletions src/main/runtime/agent-session-record-rows.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,19 +27,20 @@ import { AgentSessionTabTable, type PersistedAgentSessionTab } from './agent-ses
* no row count can say which. */
const SESSION_TABS_RECORDED = 'session_tabs_recorded'

/** Latch fields older builds wrote. Nothing reads them, and dropping them keeps a lease this build
* writes back from carrying a stale latch to an older build after a downgrade. */
/** Lease fields older builds wrote and nothing reads, dropped at load so no write carries them on. */
type RetiredAgentSessionLeaseFields = {
processlessAt?: unknown
settlementRetryRequired?: unknown
settlementRetryId?: unknown
journalCheckpoint?: unknown
}

export function withoutRetiredLeaseLatches(lease: AgentSessionLease): AgentSessionLease {
const {
processlessAt: _processlessAt,
settlementRetryRequired: _settlementRetryRequired,
settlementRetryId: _settlementRetryId,
journalCheckpoint: _journalCheckpoint,
Comment thread
brennanb2025 marked this conversation as resolved.
Resolved
...current
}: AgentSessionLease & RetiredAgentSessionLeaseFields = lease
return current
Expand Down
37 changes: 23 additions & 14 deletions src/main/runtime/agent-session-record-store.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -738,27 +738,36 @@ describe('claim keys and unreadable rows', () => {
expect(store.isClaimKeyVerifiable('key-unknown', NOW)).toBe(true)
})

it.each([
[
'invalid checkpoint',
(record: PersistedAgentSessionRecord) =>
Object.assign(record.lease, { journalCheckpoint: { epoch: 'bad', sequence: 1 } })
],
[
// No proof yet is a live owner whose provider has not answered; a proof naming no link is not.
'live proof naming no handle',
(record: PersistedAgentSessionRecord) =>
Object.assign(record.lease, { provenHandleLinkId: 'link-never-minted' })
]
])('quarantines a record with %s', async (_name, corrupt) => {
// No proof yet is a live owner whose provider has not answered; a proof naming no link is not.
it('quarantines a record with a live proof naming no handle', async () => {
const first = await open()
await establishOwner(first)
await editPersistedTestAgentSessionStore(directory, (persisted) => {
corrupt(persisted.records['session-alpha'])
Object.assign(persisted.records['session-alpha'].lease, {
provenHandleLinkId: 'link-never-minted'
})
})
expect((await open()).isSessionUnreadable('session-alpha')).toBe(true)
})

// Nothing reads the checkpoint any more, so even one this build could never have parsed loads.
it.each([null, { epoch: 3, sequence: 0 }, { epoch: 'bad', sequence: 1 }])(
"loads a record carrying an older build's journal checkpoint %o, and drops it on the next write",
async (journalCheckpoint) => {
const first = await open()
await establishOwner(first)
await editPersistedTestAgentSessionStore(directory, (persisted) => {
Object.assign(persisted.records['session-alpha'].lease, { journalCheckpoint })
})
const reopened = await open()
expect(reopened.isSessionUnreadable('session-alpha')).toBe(false)
await reopened.setConversationName('session-alpha', 'renamed')
const stored = (await readPersistedTestAgentSessionStore(directory)).records['session-alpha']
expect(stored.conversationName).toBe('renamed')
expect(stored.lease).not.toHaveProperty('journalCheckpoint')
}
)

it('sets aside a row whose handle is in both stored forms and never rewrites it', async () => {
const first = await open()
await establishOwner(first)
Expand Down
Loading
Loading