mirror of
https://github.com/stablyai/orca.git
synced 2026-10-24 00:04:01 +00:00
fix(native-chat): harden round-three reliability
This commit is contained in:
@@ -95,10 +95,13 @@ const CROSS_VERSION_WIRE_PREFIXES = [
|
||||
'src/shared/browser-client-host-placement',
|
||||
'src/shared/agent-session-wire',
|
||||
'src/shared/agent-session-mutation-envelope',
|
||||
'src/shared/agent-session-journal-',
|
||||
'src/main/native-chat/agent-session-journal/',
|
||||
'src/main/native-chat/agent-session-wire/',
|
||||
'src/main/runtime/agent-session-record-store',
|
||||
'src/main/runtime/rpc/dispatcher',
|
||||
'src/main/runtime/rpc/methods/browser-tab-create-schema',
|
||||
'src/main/runtime/rpc/methods/structured-agent-session',
|
||||
'src/main/runtime/rpc/methods/terminal',
|
||||
'src/renderer/src/runtime/remote-runtime-terminal-multiplexer'
|
||||
]
|
||||
|
||||
@@ -183,10 +183,19 @@ describe('per-job path classification', () => {
|
||||
'src/shared/terminal-stream-protocol.ts',
|
||||
'src/shared/agent-session-wire.ts',
|
||||
'src/shared/agent-session-mutation-envelope.ts',
|
||||
'src/shared/agent-session-journal-item-key.ts',
|
||||
'src/shared/agent-session-journal-types.ts',
|
||||
'src/main/native-chat/agent-session-journal/journal-cursor.ts',
|
||||
'src/main/native-chat/agent-session-journal/journal-reducer.ts',
|
||||
'src/main/native-chat/agent-session-journal/journal-row-schema.ts',
|
||||
'src/main/native-chat/agent-session-wire/structured-agent-session-host.ts',
|
||||
'src/main/runtime/agent-session-record-store.ts',
|
||||
'src/main/runtime/rpc/dispatcher.ts',
|
||||
'src/main/runtime/rpc/methods/browser-tab-create-schema.ts',
|
||||
'src/main/runtime/rpc/methods/structured-agent-session.ts',
|
||||
'src/main/runtime/rpc/methods/structured-agent-session-gate.ts',
|
||||
'src/main/runtime/rpc/methods/structured-agent-session-hold.ts',
|
||||
'src/main/runtime/rpc/methods/structured-agent-session-schemas.ts',
|
||||
'src/main/runtime/rpc/methods/terminal.ts',
|
||||
'src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts'
|
||||
]) {
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
agentJournalItemKey,
|
||||
agentJournalSubmissionKey
|
||||
agentJournalSubmissionKey,
|
||||
boundJournalKeyComponent,
|
||||
MAX_JOURNAL_KEY_COMPONENT_CHARS,
|
||||
parseAgentJournalItemKey
|
||||
} from '../../../shared/agent-session-journal-item-key'
|
||||
import type { AgentJournalItemIdentity } from '../../../shared/agent-session-journal-types'
|
||||
|
||||
@@ -143,3 +146,51 @@ describe('key encoding', () => {
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
describe('oversized identity bounding on Unicode boundaries', () => {
|
||||
// 39 UTF-16 units of ASCII put the astral character's surrogate pair across
|
||||
// the 40-unit diagnostic-head cut. Pre-fix the head ended in a lone high
|
||||
// surrogate and `encodeURIComponent` threw `URIError: URI malformed`.
|
||||
const STRADDLING = `${'a'.repeat(39)}😀${'x'.repeat(1100)}`
|
||||
const straddlingIdentity: AgentJournalItemIdentity = {
|
||||
provider: 'codex',
|
||||
threadId: THREAD,
|
||||
turnId: STRADDLING,
|
||||
ordinal: 0
|
||||
}
|
||||
|
||||
it('keys a valid astral id whose character straddles the head cut', () => {
|
||||
expect(() => agentJournalItemKey(straddlingIdentity)).not.toThrow()
|
||||
expect(boundJournalKeyComponent(STRADDLING).length).toBeLessThan(
|
||||
MAX_JOURNAL_KEY_COMPONENT_CHARS
|
||||
)
|
||||
})
|
||||
|
||||
it('stays deterministic and collision-resistant for straddling ids', () => {
|
||||
expect(agentJournalItemKey(straddlingIdentity)).toBe(agentJournalItemKey(straddlingIdentity))
|
||||
// A different oversized value sharing the same head still gets its own key.
|
||||
expect(agentJournalItemKey({ ...straddlingIdentity, turnId: `${STRADDLING}y` })).not.toBe(
|
||||
agentJournalItemKey(straddlingIdentity)
|
||||
)
|
||||
})
|
||||
|
||||
it('re-deriving from the parsed bounded key is a fixed point', () => {
|
||||
const key = agentJournalItemKey(straddlingIdentity)
|
||||
const parsed = parseAgentJournalItemKey(key)
|
||||
expect(parsed).not.toBeNull()
|
||||
expect(agentJournalItemKey(parsed as AgentJournalItemIdentity)).toBe(key)
|
||||
})
|
||||
|
||||
it('keeps an astral character that lands entirely inside the head', () => {
|
||||
const inside = `${'a'.repeat(38)}😀${'x'.repeat(1100)}`
|
||||
const bounded = boundJournalKeyComponent(inside)
|
||||
expect(bounded.startsWith(`${'a'.repeat(38)}😀~orca-oversized~`)).toBe(true)
|
||||
expect(() => encodeURIComponent(bounded)).not.toThrow()
|
||||
})
|
||||
|
||||
it('drops only the split surrogate from the straddling head', () => {
|
||||
const bounded = boundJournalKeyComponent(STRADDLING)
|
||||
expect(bounded.startsWith(`${'a'.repeat(39)}~orca-oversized~`)).toBe(true)
|
||||
expect(() => encodeURIComponent(bounded)).not.toThrow()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -7,6 +7,10 @@ import { appendJournalRows, JOURNAL_SNAPSHOT_FILE, readJournalSnapshot } from '.
|
||||
import type { JournalSnapshotFile } from './journal-log-file'
|
||||
import type { JournalRow } from './journal-row-schema'
|
||||
import { openAgentSessionJournal } from './journal-store'
|
||||
import {
|
||||
projectStructuredAgentSessionStatus,
|
||||
projectStructuredItemsToNativeChat
|
||||
} from '../../../shared/structured-agent-session-projection'
|
||||
|
||||
type FakeDirectoryHandle = { sync: ReturnType<typeof vi.fn>; close: ReturnType<typeof vi.fn> }
|
||||
|
||||
@@ -73,6 +77,64 @@ describe('readJournalSnapshot validation', () => {
|
||||
expect((await readJournalSnapshot(root)).status).toBe('valid')
|
||||
})
|
||||
|
||||
it('accepts every canonical item kind and a fully-formed submission', async () => {
|
||||
const snapshot = validSnapshot()
|
||||
const payload = { head: 'x', byteLength: 4, digest: 'd'.repeat(64), truncated: true }
|
||||
snapshot.items = [
|
||||
{ kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] },
|
||||
{
|
||||
kind: 'tool-call',
|
||||
name: 'Read',
|
||||
input: { path: 'a' },
|
||||
state: 'completed',
|
||||
output: payload
|
||||
},
|
||||
{ kind: 'diff', path: 'a.ts', patch: payload },
|
||||
{
|
||||
kind: 'approval',
|
||||
title: 'Run?',
|
||||
detail: null,
|
||||
options: [{ id: 'a', label: 'Yes' }],
|
||||
resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null }
|
||||
},
|
||||
{
|
||||
kind: 'question',
|
||||
question: 'Deploy?',
|
||||
options: [{ id: 'a', label: 'Yes' }],
|
||||
freeTextQuestionId: 'q-free',
|
||||
resolution: { state: 'resolved', selectedOptionId: 'a', resolvedBy: 'c', resolvedAt: 5 }
|
||||
},
|
||||
{
|
||||
kind: 'status',
|
||||
text: 'working',
|
||||
turnLifecycle: { turnId: 'turn-1', state: 'running' },
|
||||
providerFrame: { provider: 'codex', kind: 'raw', payload }
|
||||
}
|
||||
].map((body, index) => ({
|
||||
itemId: `codex:thread-1:turn-1:${index + 1}`,
|
||||
revision: 1,
|
||||
body: body as JournalSnapshotFile['items'][number]['body'],
|
||||
sequence: index + 1,
|
||||
observedAt: 1_000,
|
||||
...(index === 0 ? { recovered: true as const } : {})
|
||||
}))
|
||||
snapshot.compactedThrough = snapshot.items.length
|
||||
snapshot.submissions = [
|
||||
{
|
||||
clientMessageId: 'm-1',
|
||||
fence: 1,
|
||||
payloadFingerprint: 'a'.repeat(64),
|
||||
dispatchState: 'accepted',
|
||||
providerItemId: 'codex:thread-1:turn-1:1',
|
||||
reason: null,
|
||||
submittedAt: 1_000,
|
||||
resolvedAt: 1_001
|
||||
}
|
||||
]
|
||||
await writeSnapshot(snapshot)
|
||||
expect((await readJournalSnapshot(root)).status).toBe('valid')
|
||||
})
|
||||
|
||||
it('classifies a JSON-valid non-array tombstones collection as invalid instead of valid', async () => {
|
||||
await writeSnapshot({ ...validSnapshot(), tombstones: {} })
|
||||
expect((await readJournalSnapshot(root)).status).toBe('invalid')
|
||||
@@ -90,6 +152,75 @@ describe('readJournalSnapshot validation', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects JSON-valid nested item corruption instead of admitting it', async () => {
|
||||
// A resolved question with `options: null` used to pass shallow admission and
|
||||
// then throw `TypeError` in the shared projection's `options.map`.
|
||||
const poisonedQuestion = validSnapshot()
|
||||
poisonedQuestion.items = [
|
||||
{
|
||||
itemId: 'codex:thread-1:turn-1:1',
|
||||
revision: 1,
|
||||
body: {
|
||||
kind: 'question',
|
||||
question: 'Deploy?',
|
||||
options: null,
|
||||
resolution: { state: 'resolved', selectedOptionId: 'a', resolvedBy: 'c', resolvedAt: 1 }
|
||||
},
|
||||
sequence: 2,
|
||||
observedAt: 1_000
|
||||
}
|
||||
] as unknown as JournalSnapshotFile['items']
|
||||
await writeSnapshot(poisonedQuestion)
|
||||
expect((await readJournalSnapshot(root)).status).toBe('invalid')
|
||||
|
||||
// Pending-prompt surfaces read `resolution.state` before anything else.
|
||||
const nullResolution = validSnapshot()
|
||||
nullResolution.items = [
|
||||
{
|
||||
itemId: 'codex:thread-1:turn-1:1',
|
||||
revision: 1,
|
||||
body: { kind: 'question', question: 'Deploy?', options: [], resolution: null },
|
||||
sequence: 2,
|
||||
observedAt: 1_000
|
||||
}
|
||||
] as unknown as JournalSnapshotFile['items']
|
||||
await writeSnapshot(nullResolution)
|
||||
expect((await readJournalSnapshot(root)).status).toBe('invalid')
|
||||
})
|
||||
|
||||
it('rejects a JSON-valid nested corruption in the retained tail', async () => {
|
||||
const poisonedTail = validSnapshot()
|
||||
poisonedTail.tail = [
|
||||
{
|
||||
v: 1,
|
||||
epoch: 'epoch-A',
|
||||
seq: 3,
|
||||
fence: 1,
|
||||
ts: 1_000,
|
||||
kind: 'item',
|
||||
itemId: 'codex:thread-1:turn-1:3',
|
||||
revision: 1,
|
||||
body: {
|
||||
kind: 'question',
|
||||
question: 'Deploy?',
|
||||
options: null,
|
||||
resolution: { state: 'resolved', selectedOptionId: 'a', resolvedBy: 'c', resolvedAt: 1 }
|
||||
}
|
||||
}
|
||||
] as unknown as JournalSnapshotFile['tail']
|
||||
await writeSnapshot(poisonedTail)
|
||||
expect((await readJournalSnapshot(root)).status).toBe('invalid')
|
||||
})
|
||||
|
||||
it('rejects a submission that only carries a client message id', async () => {
|
||||
const shallowSubmission = validSnapshot()
|
||||
shallowSubmission.submissions = [
|
||||
{ clientMessageId: 'm-1' }
|
||||
] as unknown as JournalSnapshotFile['submissions']
|
||||
await writeSnapshot(shallowSubmission)
|
||||
expect((await readJournalSnapshot(root)).status).toBe('invalid')
|
||||
})
|
||||
|
||||
it('rejects items and counters that only look shallowly plausible', async () => {
|
||||
const missingSequence = validSnapshot()
|
||||
missingSequence.items = [
|
||||
@@ -127,6 +258,48 @@ describe('journal startup isolation from a malformed snapshot', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('reopen after a persisted JSON-valid poisoned question', () => {
|
||||
it('quarantines the snapshot so reopen-to-render cannot throw in projection', async () => {
|
||||
const poisoned = validSnapshot()
|
||||
poisoned.items = [
|
||||
{
|
||||
itemId: 'codex:thread-1:turn-1:1',
|
||||
revision: 1,
|
||||
body: {
|
||||
kind: 'question',
|
||||
question: 'Deploy?',
|
||||
options: null,
|
||||
resolution: { state: 'resolved', selectedOptionId: 'a', resolvedBy: 'c', resolvedAt: 1 }
|
||||
},
|
||||
sequence: 2,
|
||||
observedAt: 1_000
|
||||
}
|
||||
] as unknown as JournalSnapshotFile['items']
|
||||
await writeSnapshot(poisoned)
|
||||
|
||||
const journal = await openAgentSessionJournal({
|
||||
identity: {
|
||||
sessionId: 'session-1',
|
||||
workspaceId: 'ws-1',
|
||||
hostId: 'host-1',
|
||||
agent: 'codex',
|
||||
providerHandle: { kind: 'codex', threadId: 'thread-1' }
|
||||
},
|
||||
journalDir: root
|
||||
})
|
||||
|
||||
// The poisoned item must land in quarantine, not in the reopened state:
|
||||
// pre-fix it was admitted and the render path below threw
|
||||
// `TypeError: Cannot read properties of null (reading 'map')`.
|
||||
const entries = await readdir(root)
|
||||
expect(entries.some((entry) => entry.startsWith('quarantine-snapshot-'))).toBe(true)
|
||||
const items = journal.snapshot().items
|
||||
expect(() => projectStructuredItemsToNativeChat(items)).not.toThrow()
|
||||
expect(() => projectStructuredAgentSessionStatus(items)).not.toThrow()
|
||||
expect(items).toEqual([])
|
||||
})
|
||||
})
|
||||
|
||||
describe('appendJournalRows directory fsync', () => {
|
||||
const ROW: JournalRow = {
|
||||
kind: 'epoch',
|
||||
|
||||
@@ -16,6 +16,10 @@ import type {
|
||||
AgentJournalRenderItem,
|
||||
AgentJournalSubmission
|
||||
} from '../../../shared/agent-session-journal-types'
|
||||
import {
|
||||
isAdmissibleAgentJournalRenderItem,
|
||||
isAdmissibleAgentJournalSubmission
|
||||
} from '../../../shared/agent-session-journal-schemas'
|
||||
import { parseJournalRow, serializeJournalRow, type JournalRow } from './journal-row-schema'
|
||||
|
||||
export const JOURNAL_LOG_FILE = 'log.jsonl'
|
||||
@@ -159,8 +163,11 @@ function isJournalSnapshotFile(value: unknown): value is JournalSnapshotFile {
|
||||
Number.isInteger(snapshot.compactedThrough) &&
|
||||
(snapshot.compactedThrough as number) >= 0 &&
|
||||
Number.isInteger(snapshot.highestFence) &&
|
||||
arrayOf(snapshot.items, isRenderItem) &&
|
||||
arrayOf(snapshot.submissions, isSubmission) &&
|
||||
// Deep discriminated admission: a JSON-valid item with a corrupt nested
|
||||
// shape (e.g. a question whose options are null) must land this snapshot
|
||||
// in quarantine rather than throw later in projection or prompt render.
|
||||
arrayOf(snapshot.items, isAdmissibleAgentJournalRenderItem) &&
|
||||
arrayOf(snapshot.submissions, isAdmissibleAgentJournalSubmission) &&
|
||||
arrayOf(snapshot.receipts, isReceipt) &&
|
||||
arrayOf(snapshot.aliases, isAlias) &&
|
||||
// Older snapshots predate tombstones; absence is fine, a non-array is not —
|
||||
@@ -181,17 +188,6 @@ function recordOf(value: unknown): Record<string, unknown> | null {
|
||||
: null
|
||||
}
|
||||
|
||||
function isRenderItem(value: unknown): boolean {
|
||||
const item = recordOf(value)
|
||||
return Boolean(
|
||||
item &&
|
||||
typeof item.itemId === 'string' &&
|
||||
Number.isInteger(item.revision) &&
|
||||
Number.isInteger(item.sequence) &&
|
||||
recordOf(item.body)
|
||||
)
|
||||
}
|
||||
|
||||
function isTombstone(value: unknown): boolean {
|
||||
const tombstone = recordOf(value)
|
||||
return Boolean(
|
||||
@@ -199,11 +195,6 @@ function isTombstone(value: unknown): boolean {
|
||||
)
|
||||
}
|
||||
|
||||
function isSubmission(value: unknown): boolean {
|
||||
const submission = recordOf(value)
|
||||
return Boolean(submission && typeof submission.clientMessageId === 'string')
|
||||
}
|
||||
|
||||
function isReceipt(value: unknown): boolean {
|
||||
const receipt = recordOf(value)
|
||||
return Boolean(
|
||||
|
||||
@@ -107,6 +107,76 @@ describe('journal row validation', () => {
|
||||
expect(parse({ ...BASE, kind: 'epoch', reason: 'session_created' })).toBe(false)
|
||||
})
|
||||
|
||||
it('rejects JSON-valid nested body corruption that would throw during render', () => {
|
||||
const item = (body: unknown) => ({ ...BASE, kind: 'item', itemId: 'i-1', revision: 1, body })
|
||||
// A resolved question's options are mapped by the projection; null throws there.
|
||||
expect(
|
||||
parse(
|
||||
item({
|
||||
kind: 'question',
|
||||
question: 'Deploy?',
|
||||
options: null,
|
||||
resolution: { state: 'resolved', selectedOptionId: 'a', resolvedBy: 'c', resolvedAt: 1 }
|
||||
})
|
||||
)
|
||||
).toBe(false)
|
||||
// Prompt surfaces read `resolution.state` before anything else.
|
||||
expect(
|
||||
parse(item({ kind: 'question', question: 'Deploy?', options: [], resolution: null }))
|
||||
).toBe(false)
|
||||
expect(parse(item({ kind: 'message', role: 'user', blocks: 'not-blocks' }))).toBe(false)
|
||||
expect(parse(item({ kind: 'diff', path: 'a.ts', patch: { head: 'x' } }))).toBe(false)
|
||||
expect(
|
||||
parse(
|
||||
item({ kind: 'approval', title: 't', detail: null, options: [{ id: 1 }], resolution: null })
|
||||
)
|
||||
).toBe(false)
|
||||
// `turnLifecycle.turnId` is read whenever the value is truthy.
|
||||
expect(parse(item({ kind: 'status', text: 'x', turnLifecycle: true }))).toBe(false)
|
||||
})
|
||||
|
||||
it('rejects a submission row whose body is not a message item', () => {
|
||||
expect(
|
||||
parse({
|
||||
...BASE,
|
||||
kind: 'submission',
|
||||
clientMessageId: 'm-1',
|
||||
payloadFingerprint: 'a'.repeat(64),
|
||||
providerHandle: { kind: 'codex', threadId: 't' },
|
||||
body: { kind: 'status', text: 'not a message' }
|
||||
})
|
||||
).toBe(false)
|
||||
})
|
||||
|
||||
it('keeps forward compatibility for open string fields and unknown block types', () => {
|
||||
const item = (body: unknown) => ({ ...BASE, kind: 'item', itemId: 'i-1', revision: 1, body })
|
||||
// Renderers select known block types by equality and skip the rest.
|
||||
expect(
|
||||
parse(item({ kind: 'message', role: 'user', blocks: [{ type: 'future-block', data: 1 }] }))
|
||||
).toBe(true)
|
||||
// Role and tool-call state are type-checked, never enum-checked.
|
||||
expect(
|
||||
parse(item({ kind: 'message', role: 'narrator', blocks: [{ type: 'text', text: 'hi' }] }))
|
||||
).toBe(true)
|
||||
expect(parse(item({ kind: 'tool-call', name: 'Read', input: {}, state: 'paused' }))).toBe(true)
|
||||
expect(
|
||||
parse(
|
||||
item({
|
||||
kind: 'question',
|
||||
question: 'Deploy?',
|
||||
options: [{ id: 'a', label: 'Yes' }],
|
||||
resolution: {
|
||||
state: 'deferred',
|
||||
selectedOptionId: null,
|
||||
resolvedBy: null,
|
||||
resolvedAt: null
|
||||
},
|
||||
futureField: 'ignored'
|
||||
})
|
||||
)
|
||||
).toBe(true)
|
||||
})
|
||||
|
||||
it('keeps forward compatibility for new dispatch states without a version bump', () => {
|
||||
expect(
|
||||
parse({
|
||||
|
||||
@@ -12,6 +12,10 @@ import {
|
||||
type AgentJournalMessageItem,
|
||||
type AgentSessionProviderHandle
|
||||
} from '../../../shared/agent-session-journal-types'
|
||||
import {
|
||||
isAdmissibleAgentJournalItemBody,
|
||||
isAdmissibleAgentJournalMessageBody
|
||||
} from '../../../shared/agent-session-journal-schemas'
|
||||
|
||||
type JournalRowBase = {
|
||||
/** Schema version of THIS row. */
|
||||
@@ -138,14 +142,12 @@ function isPlainObject(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value)
|
||||
}
|
||||
|
||||
/** Every render body is a kinded object; anything else would poison the reducer. */
|
||||
function isKindedBody(value: unknown): boolean {
|
||||
return isPlainObject(value) && typeof value.kind === 'string' && value.kind.length > 0
|
||||
}
|
||||
|
||||
/** Field values are type-checked, never enum-checked: a future build adding a
|
||||
* dispatch state or handle kind must bump the row version, but this build
|
||||
* should not misread a same-version row as malformed over a wider enum. */
|
||||
/** Open field values are type-checked, never enum-checked: a future build
|
||||
* adding a dispatch state or handle kind must bump the row version, but this
|
||||
* build should not misread a same-version row as malformed over a wider enum.
|
||||
* Render BODIES are the exception and validate against the canonical deep
|
||||
* schema — their nested shapes are dereferenced unguarded all the way to the
|
||||
* rendered surface, so a JSON-valid corruption must fail here, not there. */
|
||||
function isJournalRow(record: Record<string, unknown>): record is JournalRow {
|
||||
if (typeof record.kind !== 'string' || !ROW_KINDS.has(record.kind)) {
|
||||
return false
|
||||
@@ -164,7 +166,7 @@ function isJournalRow(record: Record<string, unknown>): record is JournalRow {
|
||||
return (
|
||||
typeof record.itemId === 'string' &&
|
||||
Number.isInteger(record.revision) &&
|
||||
isKindedBody(record.body)
|
||||
isAdmissibleAgentJournalItemBody(record.body)
|
||||
)
|
||||
}
|
||||
if (record.kind === 'tombstone') {
|
||||
@@ -176,7 +178,7 @@ function isJournalRow(record: Record<string, unknown>): record is JournalRow {
|
||||
record.clientMessageId.length > 0 &&
|
||||
typeof record.payloadFingerprint === 'string' &&
|
||||
isPlainObject(record.providerHandle) &&
|
||||
isKindedBody(record.body)
|
||||
isAdmissibleAgentJournalMessageBody(record.body)
|
||||
)
|
||||
}
|
||||
if (record.kind === 'dispatch') {
|
||||
|
||||
@@ -80,13 +80,13 @@ function createReadOwner(
|
||||
setSnapshot({ ...snapshot, providerSession })
|
||||
}
|
||||
}
|
||||
const refreshTail = async (isStopped: () => boolean): Promise<void> => {
|
||||
const refreshTail = async (shouldStop: () => boolean): Promise<void> => {
|
||||
const result = await callStructuredAgentSession<AgentSessionHistoryResult>(
|
||||
target,
|
||||
'agentSession.history',
|
||||
{ sessionId, direction: 'tail', limit: AGENT_SESSION_HISTORY_MAX_LIMIT }
|
||||
)
|
||||
if (isStopped()) {
|
||||
if (shouldStop()) {
|
||||
return
|
||||
}
|
||||
setProviderSession(result.providerSession)
|
||||
@@ -107,7 +107,7 @@ function createReadOwner(
|
||||
let restored = snapshot.state.items.filter(countsTowardInitialHistory).length
|
||||
while (snapshot.state.hasOlder && restored < NATIVE_CHAT_INITIAL_LIMIT) {
|
||||
const oldest = oldestStructuredAgentSessionCursor(snapshot.state)
|
||||
if (!oldest || isStopped()) {
|
||||
if (!oldest || shouldStop()) {
|
||||
break
|
||||
}
|
||||
const missing = NATIVE_CHAT_INITIAL_LIMIT - restored
|
||||
@@ -121,6 +121,9 @@ function createReadOwner(
|
||||
limit: Math.min(AGENT_SESSION_HISTORY_MAX_LIMIT, missing)
|
||||
}
|
||||
)
|
||||
if (shouldStop()) {
|
||||
return
|
||||
}
|
||||
if (!older.ok || older.page.window.oldest?.sequence === oldest.sequence) {
|
||||
break
|
||||
}
|
||||
|
||||
@@ -32,7 +32,7 @@ export function startStructuredAgentSessionReadTransport(args: {
|
||||
applyEvent: (event: AgentSessionSubscribeEvent) => void
|
||||
applyError: (message: string) => void
|
||||
getCursor: () => AgentJournalCursor | null
|
||||
refreshTail: (isStopped: () => boolean) => Promise<void>
|
||||
refreshTail: (shouldStop: () => boolean) => Promise<void>
|
||||
sessionId: string
|
||||
target: RuntimeClientTarget
|
||||
}): { dispose: () => void; refresh: () => void } {
|
||||
@@ -127,7 +127,7 @@ export function startStructuredAgentSessionReadTransport(args: {
|
||||
const refresh = (): void => {
|
||||
const refreshGeneration = generation
|
||||
void args
|
||||
.refreshTail(() => stopped)
|
||||
.refreshTail(() => !isCurrentGeneration(refreshGeneration))
|
||||
.then(() => {
|
||||
if (!isCurrentGeneration(refreshGeneration)) {
|
||||
return
|
||||
@@ -145,7 +145,7 @@ export function startStructuredAgentSessionReadTransport(args: {
|
||||
}
|
||||
const initialGeneration = generation
|
||||
void args
|
||||
.refreshTail(() => stopped)
|
||||
.refreshTail(() => !isCurrentGeneration(initialGeneration))
|
||||
.then(() => {
|
||||
if (!isCurrentGeneration(initialGeneration)) {
|
||||
return
|
||||
|
||||
@@ -196,6 +196,65 @@ describe('useStructuredAgentSessionRead history window', () => {
|
||||
hasFocus.mockRestore()
|
||||
})
|
||||
|
||||
it('drops a delayed refresh after reconnect without mutating state or provider session', async () => {
|
||||
const hasFocus = vi.spyOn(document, 'hasFocus').mockReturnValue(true)
|
||||
const delayedRefresh = Promise.withResolvers<{
|
||||
ok: true
|
||||
page: AgentSessionHistoryPage
|
||||
providerSession: { key: 'session_id'; id: string }
|
||||
}>()
|
||||
const closes: (() => void)[] = []
|
||||
const initialProviderSession = { key: 'session_id', id: 'provider-initial' } as const
|
||||
mocks.call
|
||||
.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
page: page('tail', [message('initial', 1, 'assistant')], false),
|
||||
providerSession: initialProviderSession
|
||||
})
|
||||
.mockReturnValueOnce(delayedRefresh.promise)
|
||||
mocks.subscribe.mockImplementation((_target, _params, _onEvent, _onError, onClose) => {
|
||||
closes.push(onClose)
|
||||
return Promise.resolve({ unsubscribe: vi.fn() })
|
||||
})
|
||||
|
||||
const view = renderHook(() =>
|
||||
useStructuredAgentSessionRead({ sessionId: 'session-a', target: LOCAL_TARGET })
|
||||
)
|
||||
|
||||
try {
|
||||
await waitFor(() => expect(mocks.subscribe).toHaveBeenCalledOnce())
|
||||
expect(view.result.current.state.items[0]?.itemId).toBe('initial')
|
||||
expect(view.result.current.providerSession).toBe(initialProviderSession)
|
||||
const stateBeforeRefresh = view.result.current.state
|
||||
|
||||
act(() => window.dispatchEvent(new Event('focus')))
|
||||
await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(2))
|
||||
|
||||
vi.useFakeTimers()
|
||||
act(() => closes[0]?.())
|
||||
await act(async () => vi.advanceTimersByTimeAsync(750))
|
||||
expect(mocks.subscribe).toHaveBeenCalledTimes(2)
|
||||
|
||||
await act(async () => {
|
||||
delayedRefresh.resolve({
|
||||
ok: true,
|
||||
page: page('tail', [message('stale', 2, 'assistant')], false),
|
||||
providerSession: { key: 'session_id', id: 'provider-stale' }
|
||||
})
|
||||
await delayedRefresh.promise
|
||||
await Promise.resolve()
|
||||
})
|
||||
|
||||
expect(view.result.current.state).toBe(stateBeforeRefresh)
|
||||
expect(view.result.current.state.items[0]?.itemId).toBe('initial')
|
||||
expect(view.result.current.providerSession).toBe(initialProviderSession)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
view.unmount()
|
||||
hasFocus.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it('does no host work for retained inactive sessions', async () => {
|
||||
const first = renderHook(() =>
|
||||
useStructuredAgentSessionRead({
|
||||
|
||||
@@ -27,7 +27,16 @@ export function boundJournalKeyComponent(value: string): string {
|
||||
}
|
||||
const h1 = fnv1a32(value, 0x811c9dc5).toString(16).padStart(8, '0')
|
||||
const h2 = fnv1a32(value, 0x0100_0193).toString(16).padStart(8, '0')
|
||||
return `${value.slice(0, 40)}~orca-oversized~${value.length}~${h1}${h2}`
|
||||
return `${codePointBoundedHead(value, 40)}~orca-oversized~${value.length}~${h1}${h2}`
|
||||
}
|
||||
|
||||
/** `slice` cuts UTF-16 code units, so a head ending mid-astral-character would
|
||||
* carry a lone high surrogate and make `encodeURIComponent` throw on a valid
|
||||
* id. Dropping that unit keeps the head valid Unicode and deterministic. */
|
||||
function codePointBoundedHead(value: string, maxUnits: number): string {
|
||||
const head = value.slice(0, maxUnits)
|
||||
const last = head.charCodeAt(head.length - 1)
|
||||
return last >= 0xd800 && last <= 0xdbff ? head.slice(0, -1) : head
|
||||
}
|
||||
|
||||
function fnv1a32(value: string, seed: number): number {
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
isAdmissibleAgentJournalItemBody,
|
||||
isAdmissibleAgentJournalMessageBody,
|
||||
isAdmissibleAgentJournalRenderItem,
|
||||
isAdmissibleAgentJournalSubmission
|
||||
} from './agent-session-journal-schemas'
|
||||
import type {
|
||||
AgentJournalItemBody,
|
||||
AgentJournalRenderItem,
|
||||
AgentJournalSubmission
|
||||
} from './agent-session-journal-types'
|
||||
|
||||
const PAYLOAD = { head: 'x', byteLength: 4, digest: 'd'.repeat(64), truncated: true }
|
||||
const RESOLUTION = {
|
||||
state: 'pending',
|
||||
selectedOptionId: null,
|
||||
resolvedBy: null,
|
||||
resolvedAt: null
|
||||
} as const
|
||||
|
||||
// Canonical fixtures are typed: if a shape here stops compiling, the schema
|
||||
// audit below is validating the wrong model.
|
||||
const CANONICAL_BODIES: AgentJournalItemBody[] = [
|
||||
{
|
||||
kind: 'message',
|
||||
role: 'user',
|
||||
blocks: [
|
||||
{
|
||||
type: 'text',
|
||||
text: 'hi',
|
||||
providerFrame: { provider: 'codex', kind: 'raw', payload: PAYLOAD }
|
||||
},
|
||||
{ type: 'tool-call', name: 'Read', input: { path: 'a' } },
|
||||
{ type: 'tool-result', output: 'ok', isError: false },
|
||||
{ type: 'image-ref', path: '/tmp/a.png', alt: 'screenshot' }
|
||||
]
|
||||
},
|
||||
{ kind: 'tool-call', name: 'Read', input: undefined, state: 'running' },
|
||||
{ kind: 'tool-call', name: 'Read', input: {}, state: 'failed', output: PAYLOAD },
|
||||
{ kind: 'diff', path: 'a.ts', patch: PAYLOAD },
|
||||
{
|
||||
kind: 'approval',
|
||||
title: 'Run?',
|
||||
detail: null,
|
||||
options: [{ id: 'a', label: 'Yes' }],
|
||||
resolution: RESOLUTION
|
||||
},
|
||||
{
|
||||
kind: 'question',
|
||||
question: 'Deploy?',
|
||||
options: [{ id: 'a', label: 'Yes' }],
|
||||
freeTextQuestionId: 'q-free',
|
||||
resolution: { state: 'resolved', selectedOptionId: 'a', resolvedBy: 'client', resolvedAt: 5 }
|
||||
},
|
||||
{ kind: 'status', text: 'working' },
|
||||
{
|
||||
kind: 'status',
|
||||
text: 'turn',
|
||||
turnLifecycle: { turnId: 'turn-1', state: 'running' },
|
||||
providerFrame: { provider: 'codex', kind: 'raw', payload: PAYLOAD }
|
||||
}
|
||||
]
|
||||
|
||||
describe('canonical admission', () => {
|
||||
it('admits every body shape this build writes', () => {
|
||||
for (const body of CANONICAL_BODIES) {
|
||||
expect(isAdmissibleAgentJournalItemBody(body)).toBe(true)
|
||||
}
|
||||
})
|
||||
|
||||
it('admits a canonical render item and submission', () => {
|
||||
const item: AgentJournalRenderItem = {
|
||||
itemId: 'codex:t:turn:0',
|
||||
revision: 1,
|
||||
body: CANONICAL_BODIES[0] as AgentJournalItemBody,
|
||||
sequence: 1,
|
||||
observedAt: 1_000,
|
||||
recovered: true
|
||||
}
|
||||
expect(isAdmissibleAgentJournalRenderItem(item)).toBe(true)
|
||||
const submission: AgentJournalSubmission = {
|
||||
clientMessageId: 'm-1',
|
||||
fence: 1,
|
||||
payloadFingerprint: 'a'.repeat(64),
|
||||
dispatchState: 'unknown',
|
||||
providerItemId: null,
|
||||
reason: null,
|
||||
submittedAt: 1_000,
|
||||
resolvedAt: null
|
||||
}
|
||||
expect(isAdmissibleAgentJournalSubmission(submission)).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('nested corruption is rejected', () => {
|
||||
it('rejects prompt bodies whose options or resolution cannot be rendered', () => {
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({
|
||||
kind: 'question',
|
||||
question: 'Deploy?',
|
||||
options: null,
|
||||
resolution: { state: 'resolved', selectedOptionId: 'a', resolvedBy: 'c', resolvedAt: 1 }
|
||||
})
|
||||
).toBe(false)
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({
|
||||
kind: 'question',
|
||||
question: 'Deploy?',
|
||||
options: [],
|
||||
resolution: null
|
||||
})
|
||||
).toBe(false)
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({
|
||||
kind: 'approval',
|
||||
title: 'Run?',
|
||||
detail: null,
|
||||
options: [{ id: 'a' }],
|
||||
resolution: RESOLUTION
|
||||
})
|
||||
).toBe(false)
|
||||
})
|
||||
|
||||
it('rejects broken payload, lifecycle, and block shapes', () => {
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({ kind: 'diff', path: 'a.ts', patch: { head: 'x' } })
|
||||
).toBe(false)
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({ kind: 'status', text: 'x', turnLifecycle: true })
|
||||
).toBe(false)
|
||||
// A KNOWN block type with a broken payload must not slip through as a
|
||||
// "future" block.
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({
|
||||
kind: 'message',
|
||||
role: 'user',
|
||||
blocks: [{ type: 'text', text: null }]
|
||||
})
|
||||
).toBe(false)
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({ kind: 'message', role: 'user', blocks: 'not-blocks' })
|
||||
).toBe(false)
|
||||
})
|
||||
|
||||
it('rejects shallow render items and submissions', () => {
|
||||
expect(
|
||||
isAdmissibleAgentJournalRenderItem({
|
||||
itemId: 'i-1',
|
||||
revision: 1,
|
||||
body: { kind: 'status', text: 'x' }
|
||||
})
|
||||
).toBe(false)
|
||||
expect(isAdmissibleAgentJournalSubmission({ clientMessageId: 'm-1' })).toBe(false)
|
||||
})
|
||||
|
||||
it('only admits message bodies for submissions', () => {
|
||||
expect(isAdmissibleAgentJournalMessageBody({ kind: 'status', text: 'x' })).toBe(false)
|
||||
expect(isAdmissibleAgentJournalMessageBody({ kind: 'message', role: 'user', blocks: [] })).toBe(
|
||||
true
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
describe('forward tolerance', () => {
|
||||
it('keeps unknown block types, wider state strings, and extra keys admissible', () => {
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({
|
||||
kind: 'message',
|
||||
role: 'narrator',
|
||||
blocks: [{ type: 'future-block', data: 1 }],
|
||||
futureField: 'ignored'
|
||||
})
|
||||
).toBe(true)
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({ kind: 'tool-call', name: 'Read', state: 'paused' })
|
||||
).toBe(true)
|
||||
expect(
|
||||
isAdmissibleAgentJournalSubmission({
|
||||
clientMessageId: 'm-1',
|
||||
fence: 1,
|
||||
payloadFingerprint: 'a'.repeat(64),
|
||||
dispatchState: 'some-future-state',
|
||||
providerItemId: null,
|
||||
reason: null,
|
||||
submittedAt: 1_000,
|
||||
resolvedAt: null
|
||||
})
|
||||
).toBe(true)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,168 @@
|
||||
// ─── Canonical runtime schemas for the journal render model ─────────────────
|
||||
// The journal admits JSON it did not just write — snapshot files and log rows
|
||||
// re-enter from disk and are republished to clients — while the reducer, the
|
||||
// shared projection, and the prompt surfaces dereference nested fields without
|
||||
// guards. These schemas are the single deep validators for that render model:
|
||||
// admission must reject a JSON-valid but structurally wrong item (a question
|
||||
// whose `options` are null, a prompt without its `resolution`) so corruption
|
||||
// lands in quarantine instead of throwing mid-render.
|
||||
//
|
||||
// Discriminants (`kind`, known block `type`s) are validated deeply. Open string
|
||||
// fields (roles, dispatch/tool states) stay type-checked, never enum-checked,
|
||||
// and unknown object keys pass — a same-version row written by a slightly
|
||||
// newer build must not be misread as malformed (see journal-row-schema.ts).
|
||||
|
||||
import { z } from 'zod'
|
||||
import type {
|
||||
AgentJournalItemBody,
|
||||
AgentJournalMessageItem,
|
||||
AgentJournalRenderItem,
|
||||
AgentJournalSubmission
|
||||
} from './agent-session-journal-types'
|
||||
|
||||
const BoundedPayload = z.object({
|
||||
head: z.string(),
|
||||
byteLength: z.number(),
|
||||
digest: z.string(),
|
||||
truncated: z.boolean()
|
||||
})
|
||||
|
||||
const ProviderFrame = z.object({
|
||||
provider: z.string(),
|
||||
kind: z.string(),
|
||||
payload: BoundedPayload
|
||||
})
|
||||
|
||||
const KNOWN_BLOCK_TYPES = new Set(['text', 'tool-call', 'tool-result', 'image-ref'])
|
||||
|
||||
/** Renderers select blocks by `type` equality and skip what they cannot draw,
|
||||
* so an unknown block type stays admissible; a known type with a broken
|
||||
* payload does not. */
|
||||
const Block = z.union([
|
||||
z.discriminatedUnion('type', [
|
||||
z.object({
|
||||
type: z.literal('text'),
|
||||
text: z.string(),
|
||||
providerFrame: ProviderFrame.optional()
|
||||
}),
|
||||
// `input: undefined` loses its key under JSON.stringify, so a persisted
|
||||
// canonical tool call may lack it entirely.
|
||||
z.object({ type: z.literal('tool-call'), name: z.string(), input: z.unknown().optional() }),
|
||||
z.object({
|
||||
type: z.literal('tool-result'),
|
||||
output: z.string(),
|
||||
isError: z.boolean().optional()
|
||||
}),
|
||||
z.object({
|
||||
type: z.literal('image-ref'),
|
||||
path: z.string().optional(),
|
||||
url: z.string().optional(),
|
||||
alt: z.string().optional()
|
||||
})
|
||||
]),
|
||||
z.object({ type: z.string() }).refine((block) => !KNOWN_BLOCK_TYPES.has(block.type))
|
||||
])
|
||||
|
||||
const PromptOption = z.object({ id: z.string(), label: z.string() })
|
||||
|
||||
const Resolution = z.object({
|
||||
state: z.string().min(1),
|
||||
selectedOptionId: z.string().nullable(),
|
||||
resolvedBy: z.string().nullable(),
|
||||
resolvedAt: z.number().nullable()
|
||||
})
|
||||
|
||||
const MessageBody = z.object({
|
||||
kind: z.literal('message'),
|
||||
role: z.string().min(1),
|
||||
blocks: z.array(Block)
|
||||
})
|
||||
|
||||
export const AgentJournalItemBodySchema = z.discriminatedUnion('kind', [
|
||||
MessageBody,
|
||||
z.object({
|
||||
kind: z.literal('tool-call'),
|
||||
name: z.string(),
|
||||
// See the tool-call block: the key itself is lost when `input` is undefined.
|
||||
input: z.unknown().optional(),
|
||||
state: z.string().min(1),
|
||||
output: BoundedPayload.optional()
|
||||
}),
|
||||
z.object({ kind: z.literal('diff'), path: z.string(), patch: BoundedPayload }),
|
||||
z.object({
|
||||
kind: z.literal('approval'),
|
||||
title: z.string(),
|
||||
detail: z.string().nullable(),
|
||||
options: z.array(PromptOption),
|
||||
resolution: Resolution
|
||||
}),
|
||||
z.object({
|
||||
kind: z.literal('question'),
|
||||
question: z.string(),
|
||||
options: z.array(PromptOption),
|
||||
freeTextQuestionId: z.string().optional(),
|
||||
resolution: Resolution
|
||||
}),
|
||||
z.object({
|
||||
kind: z.literal('status'),
|
||||
text: z.string(),
|
||||
turnLifecycle: z.object({ turnId: z.string(), state: z.string().min(1) }).optional(),
|
||||
providerFrame: ProviderFrame.optional()
|
||||
})
|
||||
])
|
||||
|
||||
export const AgentJournalRenderItemSchema = z.object({
|
||||
itemId: z.string().min(1),
|
||||
revision: z.number().int(),
|
||||
body: AgentJournalItemBodySchema,
|
||||
sequence: z.number().int(),
|
||||
observedAt: z.number(),
|
||||
recovered: z.literal(true).optional()
|
||||
})
|
||||
|
||||
export const AgentJournalSubmissionSchema = z.object({
|
||||
clientMessageId: z.string().min(1),
|
||||
fence: z.number().int(),
|
||||
payloadFingerprint: z.string(),
|
||||
dispatchState: z.string().min(1),
|
||||
providerItemId: z.string().nullable(),
|
||||
reason: z.string().nullable(),
|
||||
submittedAt: z.number(),
|
||||
resolvedAt: z.number().nullable()
|
||||
})
|
||||
|
||||
export function isAdmissibleAgentJournalItemBody(value: unknown): value is AgentJournalItemBody {
|
||||
return AgentJournalItemBodySchema.safeParse(value).success
|
||||
}
|
||||
|
||||
/** Submission rows may only carry a user-authored message body. */
|
||||
export function isAdmissibleAgentJournalMessageBody(
|
||||
value: unknown
|
||||
): value is AgentJournalMessageItem {
|
||||
return MessageBody.safeParse(value).success
|
||||
}
|
||||
|
||||
export function isAdmissibleAgentJournalRenderItem(
|
||||
value: unknown
|
||||
): value is AgentJournalRenderItem {
|
||||
return AgentJournalRenderItemSchema.safeParse(value).success
|
||||
}
|
||||
|
||||
export function isAdmissibleAgentJournalSubmission(
|
||||
value: unknown
|
||||
): value is AgentJournalSubmission {
|
||||
return AgentJournalSubmissionSchema.safeParse(value).success
|
||||
}
|
||||
|
||||
/** Compile-time proof that every canonical value is admissible, so admission
|
||||
* can never quarantine a row a writer in this build produced. The schemas are
|
||||
* deliberately wider on open string fields, so only this direction holds. */
|
||||
type Admits<T extends true> = T
|
||||
export type CanonicalJournalShapesAreAdmissible = [
|
||||
Admits<AgentJournalItemBody extends z.input<typeof AgentJournalItemBodySchema> ? true : false>,
|
||||
Admits<AgentJournalMessageItem extends z.input<typeof MessageBody> ? true : false>,
|
||||
Admits<
|
||||
AgentJournalRenderItem extends z.input<typeof AgentJournalRenderItemSchema> ? true : false
|
||||
>,
|
||||
Admits<AgentJournalSubmission extends z.input<typeof AgentJournalSubmissionSchema> ? true : false>
|
||||
]
|
||||
Reference in New Issue
Block a user