starred/orca
1
0
Fork 0
mirror of https://github.com/stablyai/orca.git synced 2026-10-23 16:03:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(agent-hooks): reconcile authoritative remote Codex state

Browse Source
This commit is contained in:
Brennan Benson
2026-08-27 16:01:30 -07:00
parent 4f3e1f43e7
commit dd750dc6b3
14 changed files with 360 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/agent-hooks/server-ingest-remote.test.ts
+148
View File
@@ -5,6 +5,7 @@ import {
parseAgentStatusPayload
} from '../../shared/agent-status-types'
import { PANE } from './server.test-fixtures'
import { createShedSubagentsField } from '../../shared/agent-hook-relay'
const { getCohortAtEmitMock, trackMock } = vi.hoisted(() => ({
getCohortAtEmitMock: vi.fn(),
@@ -193,6 +194,153 @@ describe('AgentHookServer ingestRemote', () => {
)
})
it('clears a stale Codex roster only when the relay marks its transcript snapshot authoritative', () => {
const server = new AgentHookServer()
const child = {
id: '019fa65f-3144-7151-9c02-cff7a28f3171',
description: '/root/ssh_restart_qa',
state: 'working' as const,
startedAt: 1_000
}
server.ingestRemote(
{
paneKey: PANE,
tabId: 'tab-1',
worktreeId: 'wt-1',
hookEventName: 'PostToolUse',
payload: { state: 'working', prompt: '', agentType: 'codex', subagents: [child] }
},
'conn-1'
)
server.ingestRemote(
{
paneKey: PANE,
tabId: 'tab-1',
worktreeId: 'wt-1',
hookEventName: 'PostToolUse',
isReplay: true,
payload: { state: 'working', prompt: '', agentType: 'codex' }
},
'conn-1'
)
expect(server.getStatusSnapshot()[0]?.subagents).toEqual([child])
server.ingestRemote(
{
paneKey: PANE,
tabId: 'tab-1',
worktreeId: 'wt-1',
hookEventName: 'PostToolUse',
isReplay: true,
codexSubagentsAuthoritative: true,
payload: { state: 'working', prompt: '', agentType: 'codex' }
},
'conn-1'
)
expect(server.getStatusSnapshot()[0]?.subagents).toBeUndefined()
})
for (const hookEventName of ['PostToolUse', 'SessionStart', 'Stop']) {
it(`does not clear a Codex roster when relay shedding removed an authoritative ${hookEventName} snapshot`, () => {
const server = new AgentHookServer()
const child = {
id: '019fa65f-3144-7151-9c02-cff7a28f3171',
state: 'working' as const,
startedAt: 1_000
}
server.ingestRemote(
{
paneKey: PANE,
tabId: 'tab-1',
worktreeId: 'wt-1',
hookEventName: 'PostToolUse',
payload: { state: 'working', prompt: '', agentType: 'codex', subagents: [child] }
},
'conn-1'
)
server.ingestRemote(
{
paneKey: PANE,
tabId: 'tab-1',
worktreeId: 'wt-1',
hookEventName,
isReplay: true,
codexSubagentsAuthoritative: true,
shedFields: [createShedSubagentsField([{ ...child, id: `${child.id}-different` }])],
payload: {
state: hookEventName === 'Stop' ? 'done' : 'working',
prompt: '',
agentType: 'codex'
}
},
'conn-1'
)
expect(server.getStatusSnapshot()[0]?.subagents).toEqual([child])
})
}
it('accepts a transcript-proved parent completion over the historical replay hook name', () => {
const server = new AgentHookServer()
server.ingestRemote(
{
paneKey: PANE,
tabId: 'tab-1',
worktreeId: 'wt-1',
hookEventName: 'PostToolUse',
payload: { state: 'working', prompt: '', agentType: 'codex' }
},
'conn-1'
)
server.ingestRemote(
{
paneKey: PANE,
tabId: 'tab-1',
worktreeId: 'wt-1',
hookEventName: 'PostToolUse',
isReplay: true,
codexSubagentsAuthoritative: true,
codexAuthoritativeParentState: 'done',
payload: { state: 'done', prompt: '', agentType: 'codex' }
},
'conn-1'
)
expect(server.getStatusSnapshot()[0]?.state).toBe('done')
})
it('keeps transcript-proved children discovered from a cached SessionStart', () => {
const server = new AgentHookServer()
const child = {
id: '019fa65f-3144-7151-9c02-cff7a28f3171',
state: 'working' as const,
startedAt: 1_000
}
server.ingestRemote(
{
paneKey: PANE,
tabId: 'tab-1',
worktreeId: 'wt-1',
hookEventName: 'SessionStart',
isReplay: true,
codexSubagentsAuthoritative: true,
payload: {
state: 'working',
prompt: '',
agentType: 'codex',
subagents: [child]
}
},
'conn-1'
)
expect(server.getStatusSnapshot()[0]?.subagents).toEqual([child])
})
it('preserves active pane identity when a nested remote hook reports another agent', () => {
vi.useFakeTimers()
vi.setSystemTime(1_000)
src/main/agent-hooks/server.ts
+20 -1
View File
@@ -80,6 +80,7 @@ import { codexRosterToSnapshots } from '../../shared/codex-subagent-roster'
import { reconcileCodexSubagentTranscript } from '../../shared/codex-subagent-transcript'
import { seedCodexSubagentTranscriptFromSnapshot } from '../../shared/codex-subagent-transcript-seeding'
import {
hasShedSubagentsField,
isAgentHookSource,
restoreShedStatusFields,
type AgentHookSource
@@ -1506,7 +1507,12 @@ export class AgentHookServer {
diagnosticAwarePayload.hookEventName,
diagnosticAwarePayload.toolAgentId,
diagnosticAwarePayload.payload,
previous?.payload
previous?.payload,
{
subagentsAuthoritative: diagnosticAwarePayload.codexSubagentsAuthoritative === true,
subagentsShed: diagnosticAwarePayload.codexSubagentsShed === true,
authoritativeParentState: diagnosticAwarePayload.codexAuthoritativeParentState
}
)
}
: diagnosticAwarePayload
@@ -2486,6 +2492,8 @@ export class AgentHookServer {
providerSessionOnly?: unknown
reconcileDiagnostic?: unknown
isReplay?: boolean
codexSubagentsAuthoritative?: unknown
codexAuthoritativeParentState?: unknown
/** Payload fields the relay dropped to fit an oversized frame; validated below. */
shedFields?: unknown
claudeRunningNonAgentTask?: unknown
@@ -2708,6 +2716,17 @@ export class AgentHookServer {
providerSessionOnly: envelope.providerSessionOnly === true ? true : undefined,
...(reconcileDiagnostic !== undefined ? { reconcileDiagnostic } : {}),
isReplay: envelope.isReplay === true ? true : undefined,
codexSubagentsAuthoritative:
envelope.codexSubagentsAuthoritative === true && !hasShedSubagentsField(envelope.shedFields)
? true
: undefined,
codexSubagentsShed: hasShedSubagentsField(envelope.shedFields) ? true : undefined,
codexAuthoritativeParentState:
envelope.codexAuthoritativeParentState === 'working' ||
envelope.codexAuthoritativeParentState === 'waiting' ||
envelope.codexAuthoritativeParentState === 'done'
? envelope.codexAuthoritativeParentState
: undefined,
claudeRunningNonAgentTask:
typeof envelope.claudeRunningNonAgentTask === 'boolean'
? envelope.claudeRunningNonAgentTask
src/main/ssh/ssh-relay-session.ts
+5
View File
@@ -1505,6 +1505,8 @@ export class SshRelaySession {
teammateName?: unknown
toolAgentType?: unknown
isReplay?: unknown
codexSubagentsAuthoritative?: unknown
codexAuthoritativeParentState?: unknown
providerSession?: unknown
providerSessionOnly?: unknown
reconcileDiagnostic?: unknown
@@ -1541,6 +1543,9 @@ export class SshRelaySession {
toolAgentType:
typeof envelope.toolAgentType === 'string' ? envelope.toolAgentType : undefined,
isReplay: envelope.isReplay === true ? true : undefined,
codexSubagentsAuthoritative:
envelope.codexSubagentsAuthoritative === true ? true : undefined,
codexAuthoritativeParentState: envelope.codexAuthoritativeParentState,
providerSession: envelope.providerSession,
providerSessionOnly: envelope.providerSessionOnly === true ? true : undefined,
reconcileDiagnostic: envelope.reconcileDiagnostic,
src/relay/agent-hook-codex-reconciliation.ts
+30 -5
View File
@@ -45,8 +45,13 @@ export function reconcileRelayCodexEvent(
}
seedCodexStateFromSnapshot(state, event.paneKey, event.payload)
const transcript = getOrCreateCodexSubagentTranscriptState(state, event.paneKey)
if (event.payload.subagents?.length) {
seedCodexSubagentTranscriptFromSnapshot(transcript, event.payload.subagents, transcriptPath)
if (event.payload.subagents?.length || event.codexSubagentsAuthoritative === true) {
seedCodexSubagentTranscriptFromSnapshot(
transcript,
event.payload.subagents ?? [],
transcriptPath,
{ authoritative: event.codexSubagentsAuthoritative === true }
)
}
const roster = getOrCreateCodexSubagentRoster(state, event.paneKey)
reconcileCodexSubagentTranscript(transcript, roster, transcriptPath)
@@ -64,12 +69,32 @@ export function reconcileRelayCodexEvent(
: {}
: {})
}
const codexAuthoritativeParentState =
options.reconcileParentState &&
transcript.parentReadable === true &&
transcript.parentTerminalObserved !== undefined
? transcript.parentTerminalObserved
? ('done' as const)
: payload.state === 'waiting' &&
!payload.subagents?.some((subagent) => subagent.state === 'waiting')
? ('waiting' as const)
: ('working' as const)
: undefined
const transcriptUnreadable =
transcript.parentReadable === false ||
[...transcript.subagents.values()].some((child) => child.unresolvedSince)
const reconciledEvent = {
...event,
payload,
codexSubagentsAuthoritative:
transcript.parentReadable === true && transcript.parent.coverageAuthoritative
? true
: undefined,
codexAuthoritativeParentState
}
return transcriptUnreadable
? { ...event, payload }
? reconciledEvent
: event.reconcileDiagnostic?.reason === 'transcript-unreadable'
? { ...event, payload, reconcileDiagnostic: null }
: { ...event, payload }
? { ...reconciledEvent, reconcileDiagnostic: null }
: reconciledEvent
}
src/relay/agent-hook-envelope-build.ts
+2
View File
@@ -34,6 +34,8 @@ export function buildRelayHookEnvelope(
...(event.providerSession ? { providerSession: event.providerSession } : {}),
...(event.providerSessionOnly ? { providerSessionOnly: true } : {}),
isReplay: options.isReplay === true ? true : undefined,
codexSubagentsAuthoritative: event.codexSubagentsAuthoritative === true ? true : undefined,
codexAuthoritativeParentState: event.codexAuthoritativeParentState,
...(event.reconcileDiagnostic !== undefined
? { reconcileDiagnostic: event.reconcileDiagnostic }
: {}),
src/relay/agent-hook-envelope-publication.test.ts
+21
View File
@@ -476,6 +476,27 @@ describe('publishAgentHookEnvelope shed marker', () => {
}
})
it('drops roster authority when the authoritative roster itself is shed', () => {
const primary = makeBoundedClient(16_384)
const dispatcher = new RelayDispatcher(primary.write, primary.options)
try {
const envelope = makeEnvelope({
lastAssistantMessage: 8_000,
interactivePrompt: 6_000,
subagents: 40
})
envelope.codexSubagentsAuthoritative = true
publishAgentHookEnvelope(dispatcher, envelope)
const published = decodeEnvelopes(primary)[0]
expect(published.payload.subagents).toBeUndefined()
expect(published.codexSubagentsAuthoritative).toBeUndefined()
expect(envelope.codexSubagentsAuthoritative).toBe(true)
} finally {
dispatcher.dispose()
}
})
it('omits the marker when nothing was shed and never stamps the caller envelope', () => {
const primary = makeBoundedClient(65536)
const dispatcher = new RelayDispatcher(primary.write, primary.options)
src/relay/agent-hook-envelope-publication.ts
+3
View File
@@ -220,6 +220,9 @@ export function publishAgentHookEnvelope(
const shedField =
field === 'subagents' ? createShedSubagentsField(candidate.payload.subagents ?? []) : field
delete candidate.payload[field]
if (field === 'subagents') {
candidate.codexSubagentsAuthoritative = undefined
}
shedFields.push(shedField)
}
}
src/relay/agent-hook-server-codex-subagent-transcript.test.ts
+2
View File
@@ -69,6 +69,7 @@ describe('RelayAgentHookServer Codex subagent transcript polling', () => {
expect(response.status).toBe(204)
expect(forward.mock.calls[0]?.[0].payload.subagents).toHaveLength(1)
expect(forward.mock.calls[0]?.[0].codexSubagentsAuthoritative).toBe(true)
appendFileSync(childPath, line({ type: 'event_msg', payload: { type: 'task_complete' } }))
await vi.waitFor(
@@ -136,6 +137,7 @@ describe('RelayAgentHookServer Codex subagent transcript polling', () => {
expect(secondForward.mock.calls.at(-1)?.[0].payload.subagents).toBeUndefined()
expect(secondForward.mock.calls.at(-1)?.[0].payload.state).toBe('working')
expect(secondForward.mock.calls.at(-1)?.[0].isReplay).toBe(true)
expect(secondForward.mock.calls.at(-1)?.[0].codexSubagentsAuthoritative).toBe(true)
expect(secondForward).toHaveBeenCalledTimes(2)
},
{ timeout: 2_000, interval: 50 }
src/relay/agent-hook-status-cache.test.ts
+65 -3
View File
@@ -126,9 +126,71 @@ describe('relay agent-hook status cache', () => {
}
expect(reconcileRelayCodexEvent(state, live).payload.state).toBe('working')
expect(
reconcileRelayCodexEvent(state, live, { reconcileParentState: true }).payload.state
).toBe('done')
const restart = reconcileRelayCodexEvent(state, live, { reconcileParentState: true })
expect(restart.payload.state).toBe('done')
expect(restart.codexAuthoritativeParentState).toBe('done')
} finally {
rmSync(dir, { recursive: true, force: true })
}
})
it('does not claim an authoritative roster when an initial read skips old parent history', () => {
const dir = mkdtempSync(join(tmpdir(), 'relay-hook-status-cache-long-parent-'))
try {
const transcriptPath = join(dir, 'rollout-parent.jsonl')
writeFileSync(
transcriptPath,
`${JSON.stringify({ type: 'ignored', payload: 'x'.repeat(1024 * 1024) })}\n${JSON.stringify(
{ type: 'event_msg', payload: { type: 'task_started' } }
)}\n`
)
const state = createHookListenerState()
const reconciled = reconcileRelayCodexEvent(
state,
{
...event('PostToolUse'),
providerSession: { key: 'session_id', id: 'session-1', transcriptPath }
},
{ reconcileParentState: true }
)
expect(reconciled.codexSubagentsAuthoritative).toBeUndefined()
expect(reconciled.codexAuthoritativeParentState).toBe('working')
} finally {
rmSync(dir, { recursive: true, force: true })
}
})
it('drops seeded roster authority when more than one read window arrives between polls', () => {
const dir = mkdtempSync(join(tmpdir(), 'relay-hook-status-cache-parent-gap-'))
try {
const transcriptPath = join(dir, 'rollout-parent.jsonl')
writeFileSync(
transcriptPath,
`${JSON.stringify({ type: 'event_msg', payload: { type: 'task_started' } })}\n`
)
const state = createHookListenerState()
const initial = reconcileRelayCodexEvent(state, {
...event('PostToolUse'),
codexSubagentsAuthoritative: true,
providerSession: { key: 'session_id', id: 'session-1', transcriptPath }
})
expect(initial.codexSubagentsAuthoritative).toBe(true)
writeFileSync(
transcriptPath,
`${JSON.stringify({ type: 'event_msg', payload: { type: 'task_started' } })}\n${JSON.stringify({ type: 'ignored', payload: 'x'.repeat(1024 * 1024) })}\n${JSON.stringify({ type: 'event_msg', payload: { type: 'task_complete' } })}\n`
)
const afterGap = reconcileRelayCodexEvent(
state,
{
...event('PostToolUse'),
providerSession: { key: 'session_id', id: 'session-1', transcriptPath }
},
{ reconcileParentState: true }
)
expect(afterGap.codexSubagentsAuthoritative).toBeUndefined()
} finally {
rmSync(dir, { recursive: true, force: true })
}
src/shared/agent-hook-listener/listener-event.ts
+6
View File
@@ -40,6 +40,12 @@ export type AgentHookEventPayload = {
providerSessionOnly?: boolean
/** True when this event is a relay cache replay rather than a live hook. */
isReplay?: boolean
/** Relay transcript reconciliation proved the Codex roster is complete, including an empty roster. */
codexSubagentsAuthoritative?: boolean
/** Transport omitted a Codex roster, so absence must not clear cached children. */
codexSubagentsShed?: boolean
/** Restart reconciliation proved the parent transcript's current lifecycle state. */
codexAuthoritativeParentState?: 'working' | 'waiting' | 'done'
reconcileDiagnostic?: AgentReconcileDiagnostic | null
/** Transport-only Claude background-work evidence used to reject false input-based interrupts. */
claudeRunningNonAgentTask?: boolean
src/shared/agent-hook-listener/providers/codex-state.ts
+23 -3
View File
@@ -99,7 +99,12 @@ export function reconcileRemoteCodexState(
eventName: string | undefined,
agentId: string | undefined,
payload: ParsedAgentStatusPayload,
previous: ParsedAgentStatusPayload | undefined
previous: ParsedAgentStatusPayload | undefined,
options: {
subagentsAuthoritative?: boolean
subagentsShed?: boolean
authoritativeParentState?: CodexLeadTurnState['state']
} = {}
): ParsedAgentStatusPayload {
if (previous?.agentType === 'codex') {
seedCodexStateFromSnapshot(state, paneKey, previous)
@@ -112,16 +117,31 @@ export function reconcileRemoteCodexState(
return payload
}
const roster = getOrCreateCodexSubagentRoster(state, paneKey)
if (options.subagentsAuthoritative) {
roster.clear()
}
if (payload.subagents) {
seedCodexSubagentRoster(roster, payload.subagents)
}
if (options.authoritativeParentState) {
const previousLead = state.codexLeadStateByPaneKey.get(paneKey)
state.codexLeadStateByPaneKey.set(paneKey, {
state: options.authoritativeParentState,
model: payload.model ?? previousLead?.model
})
}
if (agentId) {
if (eventName === 'SubagentStop') {
finishCodexSubagent(roster, agentId)
}
} else {
const leadState = codexLeadStateForHookEvent(eventName)
if (eventName === 'SessionStart' || (eventName === 'Stop' && !payload.subagents)) {
const leadState = options.authoritativeParentState
? undefined
: codexLeadStateForHookEvent(eventName)
if (
(eventName === 'SessionStart' && !options.subagentsAuthoritative && !options.subagentsShed) ||
(eventName === 'Stop' && !payload.subagents && !options.subagentsShed)
) {
roster.clear()
}
if (leadState) {
src/shared/agent-hook-relay.ts
+14
View File
@@ -104,6 +104,10 @@ export type AgentHookRelayEnvelope = {
providerSessionOnly?: boolean
/** True when the relay is replaying its cache after Orca reconnects. */
isReplay?: boolean
/** Optional proof that the relay read the parent transcript and the Codex roster is complete. */
codexSubagentsAuthoritative?: boolean
/** Optional parent lifecycle state proved by restart transcript reconciliation. */
codexAuthoritativeParentState?: 'working' | 'waiting' | 'done'
/** Optional restart reconciliation diagnostic; older peers ignore it. */
reconcileDiagnostic?: AgentReconcileDiagnostic | null
/** Claude background-work evidence for input-interrupt inference on the receiving host. */
@@ -128,6 +132,16 @@ export const AGENT_HOOK_NOTIFICATION_METHOD = 'agent.hook' as const
export const AGENT_HOOK_SHED_FIELDS_KEY = 'shedFields' as const
const AGENT_HOOK_SHED_SUBAGENTS_DIGEST_PREFIX = 'subagents:sha256:'
export function hasShedSubagentsField(shedFields: unknown): boolean {
return (
Array.isArray(shedFields) &&
shedFields.some(
(field) =>
typeof field === 'string' && field.startsWith(AGENT_HOOK_SHED_SUBAGENTS_DIGEST_PREFIX)
)
)
}
function subagentRosterDigest(subagents: readonly AgentSubagentSnapshot[]): string {
const stableRoster = subagents.map(({ id, state, startedAt, agentType, model, description }) => [
id,
src/shared/codex-subagent-transcript-seeding.ts
+7 -2
View File
@@ -9,7 +9,8 @@ export function seedCodexSubagentTranscriptFromSnapshot(
state: CodexSubagentTranscriptState,
snapshots: readonly (Pick<AgentSubagentSnapshot, 'id' | 'description' | 'startedAt' | 'model'> &
Partial<Pick<AgentSubagentSnapshot, 'state'>>)[],
transcriptPath?: string
transcriptPath?: string,
options: { authoritative?: boolean } = {}
): void {
const normalizedPath = transcriptPath?.trim()
if (
@@ -18,11 +19,14 @@ export function seedCodexSubagentTranscriptFromSnapshot(
extname(normalizedPath) === '.jsonl' &&
state.parent.filePath !== normalizedPath
) {
state.parent = { filePath: normalizedPath, offset: 0, carry: '' }
state.parent = { filePath: normalizedPath, offset: 0, carry: '', coverageAuthoritative: false }
state.subagents.clear()
state.parentTerminalObserved = undefined
state.parentReadable = undefined
}
if (options.authoritative) {
state.parent.coverageAuthoritative = true
}
for (const snapshot of snapshots) {
if (!SAFE_THREAD_ID.test(snapshot.id) || state.subagents.has(snapshot.id)) {
continue
@@ -30,6 +34,7 @@ export function seedCodexSubagentTranscriptFromSnapshot(
state.subagents.set(snapshot.id, {
offset: 0,
carry: '',
coverageAuthoritative: false,
description: snapshot.description,
model: snapshot.model,
restoredState:
src/shared/codex-subagent-transcript.ts
+14 -2
View File
@@ -19,6 +19,7 @@ type JsonlCursor = {
filePath?: string
offset: number
carry: string
coverageAuthoritative: boolean
}
type TrackedTranscriptSubagent = JsonlCursor & {
@@ -61,8 +62,12 @@ function readJsonlCursor(cursor: JsonlCursor): JsonRecord[] | undefined {
if (stats.size < cursor.offset) {
cursor.offset = 0
cursor.carry = ''
cursor.coverageAuthoritative = false
}
if (stats.size === cursor.offset) {
if (cursor.offset === 0) {
cursor.coverageAuthoritative = true
}
return []
}
const bytesToRead = Math.min(stats.size - cursor.offset, TRANSCRIPT_READ_MAX_BYTES)
@@ -81,6 +86,12 @@ function readJsonlCursor(cursor: JsonlCursor): JsonRecord[] | undefined {
}
}
const skippedPrefix = start !== cursor.offset
if (skippedPrefix) {
cursor.coverageAuthoritative = false
}
if (start === 0 && cursor.offset === 0) {
cursor.coverageAuthoritative = true
}
const content = `${skippedPrefix ? '' : cursor.carry}${buffer.toString('utf8', 0, bytesRead)}`
const lines = content.split('\n')
cursor.offset = start + bytesRead
@@ -168,7 +179,7 @@ function childIsComplete(records: JsonRecord[]): boolean {
export function createCodexSubagentTranscriptState(): CodexSubagentTranscriptState {
return {
parent: { offset: 0, carry: '' },
parent: { offset: 0, carry: '', coverageAuthoritative: false },
subagents: new Map()
}
}
@@ -195,7 +206,7 @@ export function reconcileCodexSubagentTranscript(
for (const id of state.subagents.keys()) {
finishCodexSubagent(roster, id)
}
state.parent = { filePath: normalizedPath, offset: 0, carry: '' }
state.parent = { filePath: normalizedPath, offset: 0, carry: '', coverageAuthoritative: false }
state.subagents.clear()
state.parentTerminalObserved = undefined
state.parentReadable = undefined
@@ -221,6 +232,7 @@ export function reconcileCodexSubagentTranscript(
const tracked = state.subagents.get(activity.id) ?? {
offset: 0,
carry: '',
coverageAuthoritative: false,
startedAt: activity.startedAt
}
tracked.description = activity.description ?? tracked.description