Skip to content
Merged
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,10 @@ export type StructuredAgentSessionHostDeps = {
/** Whether an orchestration dispatch still owns this session's worker; absent answers no. */
hasOpenDispatch?: (record: AgentSessionRecord) => boolean
onEventSinkError?: (input: { sessionId: string; error: unknown }) => void
/** Lease bookkeeping run for startup or a read (the reconcile, or resolving a chat's recovery)
* that refused or threw, once per distinct failure. Startup and the read carry on: the next
* attach or send reconciles and resolves recovery again before it acts. */
onLeaseReconcileFailure?: (failure: unknown) => void
/** Every status projection this host publishes. `replay` marks a re-projection of state the host
* already knew (restore, an arriving subscriber) rather than a fresh journal edge. */
onSessionStatusChanged?: (
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ export class StructuredAgentSessionHost {
flushStreamedEvents: (sessionId) => this.flushStreamedEvents(sessionId)
})
this.restore = createStructuredAgentSessionHostRestore(deps, {
reconcile: this.reconcileLeases,
reconcileLeases: this.reconcileLeases,
resolveRecovery: (sessionId) => this.runtimeState.resolveRecovery(sessionId),
serialize: (sessionId, task) => this.serialize(sessionId, task),
hasSession: this.hasSession,
Expand Down Expand Up @@ -215,6 +215,7 @@ export class StructuredAgentSessionHost {
listSessionTabs = () => sessionTabs.listStructuredAgentSessionTabs(this.sessions)
getPersistedVisibleSessionTabIndex = () => this.deps.store.getVisibleSessionTabIndex()
getSessionTabId = (sessionId: string): string | null => this.deps.store.getSessionTabId(sessionId)
showSessionTabs = (sessionIds: readonly string[]) => this.deps.store.showSessionTabs(sessionIds)

setSessionTabVisibility = async (
sessionId: string,
Expand All @@ -228,12 +229,7 @@ export class StructuredAgentSessionHost {
}
}

reconcileRestartLeases = async (): Promise<void> => {
const refusal = await this.reconcileLeases('startup')
if (refusal) {
throw new Error(refusal.code)
}
}
reconcileRestartLeases = (): Promise<void> => this.restore.reconcileRestartLeases()

restoreReadableSessions = (sessionIds?: readonly string[]): Promise<void> =>
this.restore.restoreReadableSessions(sessionIds)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,8 @@ describe('StructuredAgentSessionReadableRestorer', () => {
adapter: {}
},
supportsRecord: () => true,
reconcile: async () => null,
resolveRecovery: async () => undefined,
reconcile: async () => true,
resolveRecovery: async () => true,
serialize: async (_sessionId, task) => task(),
hasSession: () => false,
onReadable: () => undefined
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,76 @@ export function createRestartReconciler(deps: {
}
}

/** Reports each distinct failure once, until `clear` says the bookkeeping settled again: the
* startup check, the restore pass and a restore retried after a failed journal open would each
* log the same store failure. */
export type ReaderBookkeepingFailures = {
report: (failure: unknown) => void
clear: () => void
}

export function reportEachFailureOnce(
onFailure: ((failure: unknown) => void) | undefined
): ReaderBookkeepingFailures {
let reported: string | null = null
return {
report: (failure) => {
const key = failureKey(failure)
if (key !== reported) {
reported = key
try {
onFailure?.(failure)
} catch (sinkError) {
// A throwing sink must not turn the reported failure back into a failed read.
console.warn('[structured-agent-session] reporting a lease bookkeeping failure failed', {
failure,
sinkError
})
}
}
},
clear: () => {
reported = null
}
}
}

/** The reconcile a reader runs, at startup and before each restored read: it never throws, since
* an unreconciled lease grants no writer and the next send reconciles again before it acts.
* Answers whether every lease is settled. */
export function createReaderReconcile(
reconcile: (sessionId: string) => Promise<AgentSessionWireRefusal | null>,
failures: ReaderBookkeepingFailures
): (sessionId: string) => Promise<boolean> {
return async (sessionId) => {
let failure: unknown
try {
const refusal = await reconcile(sessionId)
if (!refusal) {
failures.clear()
Comment thread
greptile-apps[bot] marked this conversation as resolved.
return true
}
failure = refusal
} catch (error) {
failure = error
}
failures.report(failure)
return false
}
}

function failureKey(failure: unknown): string {
if (typeof failure === 'object' && failure !== null) {
if ('code' in failure && failure.code) {
return String(failure.code)
}
if ('message' in failure) {
return String(failure.message)
}
}
return String(failure)
}

async function reconcileCurrentLeases(deps: {
store: AgentSessionRecordStore
probe: (record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -62,8 +62,8 @@ describe('restart journal restoration', () => {
const restoration = restoreStructuredAgentSessionsOnRestart({
openDeps: NO_OPEN_DEPS,
records,
reconcile: async () => null,
resolveRecovery: async () => undefined,
reconcile: async () => true,
resolveRecovery: async () => true,
serialize: async (_sessionId, task) => task(),
hasSession: () => false,
onReadable: () => undefined
Expand Down Expand Up @@ -103,8 +103,8 @@ describe('restart journal restoration', () => {
await restoreStructuredAgentSessionsOnRestart({
openDeps: NO_OPEN_DEPS,
records,
reconcile: async () => null,
resolveRecovery: async () => undefined,
reconcile: async () => true,
resolveRecovery: async () => true,
serialize: async (_sessionId, task) => task(),
hasSession: () => false,
onReadable: () => undefined
Expand Down Expand Up @@ -150,9 +150,10 @@ describe('restart journal restoration', () => {
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the restore reads only the record's session id here.
records: [{ sessionId: 'session-1' } as AgentSessionRecord],
openDeps: NO_OPEN_DEPS,
reconcile: async () => null,
reconcile: async () => true,
resolveRecovery: async () => {
calls.push('resolveRecovery')
return true
},
serialize: async (_sessionId, task) => task(),
hasSession: () => false,
Expand All @@ -164,6 +165,58 @@ describe('restart journal restoration', () => {
expect(calls).toEqual(['resolveRecovery', 'open', 'onReadable:restored'])
})

// Each failed bookkeeping call stands for one wait on a held store lock.
describe('once lease bookkeeping fails in a pass', () => {
const records = Array.from(
{ length: 8 },
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the restore reads only the record's session id here.
(_, index) => ({ sessionId: `session-${index}` }) as AgentSessionRecord
)
const restore = (
bookkeeping: Pick<
Parameters<typeof restoreStructuredAgentSessionsOnRestart>[0],
'reconcile' | 'resolveRecovery'
>
) =>
restoreStructuredAgentSessionsOnRestart({
openDeps: NO_OPEN_DEPS,
records,
...bookkeeping,
serialize: async (_sessionId, task) => task(),
hasSession: () => false,
onReadable: () => undefined
})

/** A failure that takes a while, as a lock wait does, so the chats open at once overlap it. */
const slowFailure = async (): Promise<boolean> => {
await new Promise((resolve) => setTimeout(resolve, 5))
return false
}

beforeEach(() => restoreRead.mockResolvedValue(null))

it('skips it for every chat when the pass check fails, and still opens them all', async () => {
const reconcile = vi.fn(slowFailure)
const resolveRecovery = vi.fn(async () => true)

await restore({ reconcile, resolveRecovery })

expect(reconcile).toHaveBeenCalledOnce()
expect(resolveRecovery).not.toHaveBeenCalled()
expect(restoreRead).toHaveBeenCalledTimes(records.length)
})

it('starts no more after the first failed recovery, and still opens every chat', async () => {
const resolveRecovery = vi.fn(slowFailure)

await restore({ reconcile: async () => true, resolveRecovery })

// Only those already started when the first failed: at most one per chat open at once.
expect(resolveRecovery.mock.calls.length).toBeLessThanOrEqual(4)
expect(restoreRead).toHaveBeenCalledTimes(records.length)
})
})

it('does not settle again when a second restore finds the session already open', async () => {
restoreRead.mockResolvedValue({
session: { journal: {}, params: {}, child: null },
Expand All @@ -174,8 +227,8 @@ describe('restart journal restoration', () => {
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the restore reads only the record's session id here.
records: [{ sessionId: 'session-1' } as AgentSessionRecord],
openDeps: NO_OPEN_DEPS,
reconcile: async () => null,
resolveRecovery: async () => undefined,
reconcile: async () => true,
resolveRecovery: async () => true,
serialize: async (_sessionId, task) => task(),
hasSession: () => true,
onReadable: () => undefined
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@

import { setImmediate as yieldToEventLoop } from 'node:timers/promises'
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire'
import { mapWithConcurrency } from '../../../shared/map-with-concurrency'
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import type {
Expand All @@ -28,8 +27,11 @@ export type StructuredAgentSessionReadRestoreDeps = {
openDeps: StructuredAgentSessionConversationOpenDeps & {
store: Pick<AgentSessionRecordStore, 'getRecord' | 'listRecords'>
}
reconcile: (sessionId: string) => Promise<AgentSessionWireRefusal | null>
resolveRecovery: (sessionId: string) => Promise<unknown>
// Lease bookkeeping. Neither throws: a read grants no writer, so bookkeeping must not block it.
/** Whether every lease is settled. */
reconcile: (sessionId: string) => Promise<boolean>
/** False when its store write failed; the next attach or send resolves it again. */
resolveRecovery: (sessionId: string) => Promise<boolean>
serialize: <T>(sessionId: string, task: () => Promise<T>) => Promise<T>
hasSession: (sessionId: string) => boolean
onReadable: (
Expand All @@ -41,13 +43,10 @@ export type StructuredAgentSessionReadRestoreDeps = {
/** One session's share of the restart restore. Startup maps this over every supported record. */
async function restoreOneStructuredAgentSessionRead(
input: StructuredAgentSessionReadRestoreDeps,
sessionId: string
sessionId: string,
settleLeases: (sessionId: string) => Promise<void>
): Promise<void> {
const unreconciled = await input.reconcile(sessionId)
if (!unreconciled) {
// A session latched in recovery exits here at startup, without waiting for a client.
await input.resolveRecovery(sessionId)
}
await settleLeases(sessionId)
await input.serialize(sessionId, () =>
restoreOneStructuredAgentSessionReadUnderSerialize(input, sessionId)
)
Expand All @@ -73,9 +72,26 @@ async function restoreOneStructuredAgentSessionReadUnderSerialize(
export async function restoreStructuredAgentSessionsOnRestart(
input: StructuredAgentSessionReadRestoreDeps & { records: AgentSessionRecord[] }
): Promise<void> {
const [first] = input.records
if (!first) {
return
}
// One check for the pass. Each chat checks again while it holds, since another writer can mark
// leases unreconciled mid-pass; after the first failure, retrying per chat only waits on the
// same store again, and the next attach or send settles those chats instead.
let settled = await input.reconcile(first.sessionId)
const settleLeases = async (sessionId: string): Promise<void> => {
// A session latched in recovery exits here at startup, without waiting for a client.
if (
settled &&
!((await input.reconcile(sessionId)) && (await input.resolveRecovery(sessionId)))
) {
settled = false
}
}
await mapWithConcurrency(input.records, JOURNAL_RESTORE_CONCURRENCY, async ({ sessionId }) => {
// A journal open is synchronous SQLite: without a macrotask per chat the restore is one long task.
await yieldToEventLoop()
await restoreOneStructuredAgentSessionRead(input, sessionId)
await restoreOneStructuredAgentSessionRead(input, sessionId, settleLeases)
})
}
Original file line number Diff line number Diff line change
Expand Up @@ -232,8 +232,8 @@ async function restore(sessionIds: readonly string[]) {
await restoreStructuredAgentSessionsOnRestart({
openDeps: deps,
records: sessionIds.map(recordFor),
reconcile: async () => null,
resolveRecovery: async () => undefined,
reconcile: async () => true,
resolveRecovery: async () => true,
serialize: async (_sessionId, task) => task(),
hasSession: (sessionId) => sessions.has(sessionId),
onReadable: (sessionId, opened) => {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,15 @@
// a send does. And a journal it cannot open is not a refusal — the chat shows that failure with a
// Retry, so the tab is worth publishing either way.

import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire'
import { agentSessionRefusalError } from '../../../shared/agent-session-wire-refusals'
import { adapterSupportsRecord } from './structured-agent-session-provider-support'
import { StructuredAgentSessionReadableRestorer } from './structured-agent-session-readable-restorer'
import { StructuredAgentSessionRestartRestoreGate } from './structured-agent-session-restart-restore-gate'
import {
createReaderReconcile,
reportEachFailureOnce
} from './structured-agent-session-restart-reconcile'
import type {
StructuredAgentSessionHostDeps,
StructuredAgentSessionReveal
Expand Down Expand Up @@ -51,23 +56,44 @@ export async function revealStructuredAgentSession(
}
}

/** The host's startup readable-restore sweep: reconcile, resolve, then open each chat's journal. */
/** The host's startup readable-restore sweep: reconcile, resolve, then open each chat's journal.
* Its lease bookkeeping is a reader's, which never fails a read or startup; startup shares it. */
export function createStructuredAgentSessionHostRestore(
deps: StructuredAgentSessionHostDeps,
wiring: Omit<
ConstructorParameters<typeof StructuredAgentSessionReadableRestorer>[0],
'openDeps' | 'supportsRecord'
>
'openDeps' | 'supportsRecord' | 'reconcile' | 'resolveRecovery'
> & {
reconcileLeases: (sessionId: string) => Promise<AgentSessionWireRefusal | null>
resolveRecovery: (sessionId: string) => Promise<unknown>
}
): {
reconcileRestartLeases: () => Promise<void>
restoreReadableSessions: (sessionIds?: readonly string[]) => Promise<void>
} {
const { reconcileLeases, resolveRecovery, ...rest } = wiring
const failures = reportEachFailureOnce(deps.onLeaseReconcileFailure)
const reconcile = createReaderReconcile(reconcileLeases, failures)
const restorer = new StructuredAgentSessionReadableRestorer({
openDeps: deps,
supportsRecord: (record) => adapterSupportsRecord(deps.adapter, record),
...wiring
reconcile,
// The next attach or send resolves recovery again, strictly, before it acts.
resolveRecovery: (sessionId) =>
resolveRecovery(sessionId).then(
() => true,
(error: unknown) => {
failures.report(error)
return false
}
),
...rest
})
const gate = new StructuredAgentSessionRestartRestoreGate()
return {
reconcileRestartLeases: async () => {
await reconcile('startup')
},
restoreReadableSessions: (sessionIds) => gate.run(() => restorer.restore(sessionIds))
}
}
Loading
Loading