mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 16:03:49 +00:00
fix: close codex restart reconciliation gaps
This commit is contained in:
@@ -268,6 +268,7 @@ describe('AgentHookServer Codex subagent transcript polling', () => {
|
||||
const parentPath = join(dir, 'rollout-parent.jsonl')
|
||||
writeFileSync(parentPath, spawnLine(CHILD_ID, '/root/missing'))
|
||||
const first = new AgentHookServer()
|
||||
const server = new AgentHookServer()
|
||||
await first.start({ env: 'production', userDataPath: dir })
|
||||
try {
|
||||
const post = async (server: AgentHookServer): Promise<void> => {
|
||||
@@ -293,7 +294,6 @@ describe('AgentHookServer Codex subagent transcript polling', () => {
|
||||
}
|
||||
await post(first)
|
||||
first.stop()
|
||||
const server = new AgentHookServer()
|
||||
await server.start({ env: 'production', userDataPath: dir })
|
||||
await vi.waitFor(
|
||||
() =>
|
||||
@@ -311,6 +311,7 @@ describe('AgentHookServer Codex subagent transcript polling', () => {
|
||||
observedAt: expect.any(Number)
|
||||
})
|
||||
} finally {
|
||||
server.stop()
|
||||
first.stop()
|
||||
}
|
||||
})
|
||||
|
||||
@@ -71,6 +71,18 @@ describe('AgentHookServer pane authority', () => {
|
||||
])
|
||||
})
|
||||
|
||||
it('re-arms Codex restart reconciliation on the destination pane after transfer', () => {
|
||||
const server = new AgentHookServer()
|
||||
const schedule = vi.fn()
|
||||
;(
|
||||
server as unknown as { scheduleCodexRestartReconciliation: (paneKey: string) => void }
|
||||
).scheduleCodexRestartReconciliation = schedule
|
||||
|
||||
server.transferPaneAuthority(SOURCE, TARGET, 'pty-1')
|
||||
|
||||
expect(schedule).toHaveBeenCalledWith(TARGET)
|
||||
})
|
||||
|
||||
it('requires live PTY ownership for a first transfer and trusts the chained alias afterward', () => {
|
||||
const server = new AgentHookServer()
|
||||
|
||||
|
||||
@@ -2087,6 +2087,7 @@ export class AgentHookServer {
|
||||
this.clearCodexSubagentPoll(previousOwnerPaneKey)
|
||||
this.clearCodexRestartReconcile(previousOwnerPaneKey)
|
||||
this.bumpCodexPaneGeneration(previousOwnerPaneKey)
|
||||
this.scheduleCodexRestartReconciliation(toPaneKey)
|
||||
// Why: the live process keeps posting the physical source key after detach; persist a chain-safe mapping to the current owner.
|
||||
this.legacyPaneKeyAliases.set(physicalPaneKey, {
|
||||
stablePaneKey: toPaneKey,
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
import type { AgentHookRelayEnvelope } from '../shared/agent-hook-relay'
|
||||
|
||||
export type RelayHookForward = (envelope: AgentHookRelayEnvelope) => void
|
||||
export type RelayHookServerOptions = {
|
||||
endpointDir?: string
|
||||
env?: string
|
||||
token?: string
|
||||
preferredPort?: number
|
||||
forward: RelayHookForward
|
||||
}
|
||||
export type RelayHookServerStartOptions = { publishEndpoint?: boolean }
|
||||
@@ -6,11 +6,17 @@ import { RelayAgentHookServer } from './agent-hook-server'
|
||||
import type { AgentHookResultRetryScheduler } from './agent-hook-result-retry-scheduler'
|
||||
import { endpointDirForRelaySocket } from './agent-hook-endpoint-coordinates'
|
||||
import type { AgentHookRelayEnvelope } from '../shared/agent-hook-relay'
|
||||
import type { AgentHookEventPayload } from '../shared/agent-hook-listener'
|
||||
import { makePaneKey } from '../shared/stable-pane-id'
|
||||
import * as agentHookListener from '../shared/agent-hook-listener/grok-result-discovery'
|
||||
|
||||
const LEAF_ID = '11111111-1111-4111-8111-111111111111'
|
||||
const PANE_KEY = makePaneKey('tab-1', LEAF_ID)
|
||||
const MALFORMED_PANE_KEYS = [
|
||||
makePaneKey('tab-1', '22222222-2222-4222-8222-222222222222'),
|
||||
makePaneKey('tab-1', '33333333-3333-4333-8333-333333333333'),
|
||||
makePaneKey('tab-1', '44444444-4444-4444-8444-444444444444')
|
||||
]
|
||||
|
||||
type RelayServerInternals = {
|
||||
state: { lastStatusByPaneKey: Map<string, unknown> }
|
||||
@@ -372,17 +378,17 @@ describe('RelayAgentHookServer', () => {
|
||||
JSON.stringify({
|
||||
version: 1,
|
||||
entries: [
|
||||
{ event: { paneKey: PANE_KEY }, meta: { source: 'claude' } },
|
||||
{ event: { paneKey: MALFORMED_PANE_KEYS[0] }, meta: { source: 'claude' } },
|
||||
{
|
||||
event: {
|
||||
paneKey: PANE_KEY,
|
||||
paneKey: MALFORMED_PANE_KEYS[1],
|
||||
payload: { state: 'working', prompt: 'missing agent type' }
|
||||
},
|
||||
meta: { source: 'claude' }
|
||||
},
|
||||
{
|
||||
event: {
|
||||
paneKey: PANE_KEY,
|
||||
paneKey: MALFORMED_PANE_KEYS[2],
|
||||
payload: null
|
||||
},
|
||||
meta: { source: 'claude' }
|
||||
@@ -435,6 +441,68 @@ describe('RelayAgentHookServer', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('sanitizes malformed provider session metadata during hydration', async () => {
|
||||
writeFileSync(
|
||||
join(dir, 'hook-status-cache.json'),
|
||||
JSON.stringify({
|
||||
version: 1,
|
||||
entries: [
|
||||
{
|
||||
event: {
|
||||
paneKey: PANE_KEY,
|
||||
payload: { state: 'working', prompt: 'bad transcript path', agentType: 'codex' },
|
||||
providerSession: { key: 'session_id', id: 'session-1', transcriptPath: 5 }
|
||||
},
|
||||
meta: { source: 'codex' }
|
||||
}
|
||||
]
|
||||
}),
|
||||
'utf8'
|
||||
)
|
||||
|
||||
const forward = vi.fn<(envelope: AgentHookRelayEnvelope) => void>()
|
||||
const server = new RelayAgentHookServer({ endpointDir: dir, forward })
|
||||
await expect(server.start()).resolves.toBeUndefined()
|
||||
try {
|
||||
expect(server.replayCachedPayloadsForPanes()).toBe(1)
|
||||
expect(forward).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
providerSession: { key: 'session_id', id: 'session-1' }
|
||||
})
|
||||
)
|
||||
} finally {
|
||||
server.stop()
|
||||
}
|
||||
})
|
||||
|
||||
it('ignores request continuations that arrive after the relay stops', async () => {
|
||||
const forward = vi.fn<(envelope: AgentHookRelayEnvelope) => void>()
|
||||
const server = new RelayAgentHookServer({ endpointDir: dir, forward })
|
||||
await server.start()
|
||||
server.stop()
|
||||
|
||||
;(
|
||||
server as unknown as {
|
||||
applyEvent: (
|
||||
event: AgentHookEventPayload,
|
||||
source: 'codex',
|
||||
env?: string,
|
||||
version?: string
|
||||
) => void
|
||||
}
|
||||
).applyEvent(
|
||||
{
|
||||
paneKey: PANE_KEY,
|
||||
connectionId: null,
|
||||
payload: { state: 'working', prompt: 'late', agentType: 'codex' }
|
||||
},
|
||||
'codex'
|
||||
)
|
||||
|
||||
expect(server.replayCachedPayloadsForPanes()).toBe(0)
|
||||
expect(forward).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('forwards and replays Pi session identity as metadata-only', async () => {
|
||||
const forward = vi.fn<(envelope: AgentHookRelayEnvelope) => void>()
|
||||
const server = new RelayAgentHookServer({ endpointDir: dir, forward })
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
const STATUS_CACHE_WRITE_DEBOUNCE_MS = 25
|
||||
|
||||
/** Coalesces bursty hook updates while flushing synchronously at lifecycle boundaries. */
|
||||
export class RelayHookStatusCacheWriter {
|
||||
private timer: ReturnType<typeof setTimeout> | undefined
|
||||
|
||||
constructor(private readonly write: () => void) {}
|
||||
|
||||
schedule(): void {
|
||||
if (this.timer) {
|
||||
return
|
||||
}
|
||||
this.timer = setTimeout(() => {
|
||||
this.timer = undefined
|
||||
this.write()
|
||||
}, STATUS_CACHE_WRITE_DEBOUNCE_MS)
|
||||
if (typeof this.timer.unref === 'function') {
|
||||
this.timer.unref()
|
||||
}
|
||||
}
|
||||
|
||||
flush(): void {
|
||||
if (this.timer) {
|
||||
clearTimeout(this.timer)
|
||||
this.timer = undefined
|
||||
}
|
||||
this.write()
|
||||
}
|
||||
}
|
||||
@@ -16,6 +16,8 @@ import { codexRosterToSnapshots } from '../shared/codex-subagent-roster'
|
||||
import { reconcileCodexSubagentTranscript } from '../shared/codex-subagent-transcript'
|
||||
import { buildRelayHookEnvelope } from './agent-hook-envelope-build'
|
||||
import { parsePaneKey } from '../shared/stable-pane-id'
|
||||
import { normalizeAgentProviderSession } from '../shared/agent-session-resume'
|
||||
import { normalizeAgentReconcileDiagnostic } from '../shared/agent-reconcile-diagnostic'
|
||||
|
||||
export type RelayHookStatusMeta = { source: AgentHookSource; env?: string; version?: string }
|
||||
type PersistedCache = {
|
||||
@@ -62,8 +64,25 @@ function sanitizeHydratedEntry(
|
||||
) {
|
||||
return null
|
||||
}
|
||||
const providerSession =
|
||||
rawEvent.providerSession === undefined
|
||||
? undefined
|
||||
: normalizeAgentProviderSession(rawEvent.providerSession)
|
||||
if (rawEvent.providerSession !== undefined && !providerSession) {
|
||||
return null
|
||||
}
|
||||
const reconcileDiagnostic = normalizeAgentReconcileDiagnostic(rawEvent.reconcileDiagnostic)
|
||||
if (rawEvent.reconcileDiagnostic !== undefined && reconcileDiagnostic === undefined) {
|
||||
return null
|
||||
}
|
||||
return {
|
||||
event: { ...rawEvent, paneKey, payload } as AgentHookEventPayload,
|
||||
event: {
|
||||
...rawEvent,
|
||||
paneKey,
|
||||
payload,
|
||||
...(providerSession ? { providerSession } : {}),
|
||||
...(reconcileDiagnostic !== undefined ? { reconcileDiagnostic } : {})
|
||||
} as AgentHookEventPayload,
|
||||
meta: {
|
||||
source: rawMeta.source,
|
||||
env: typeof rawMeta.env === 'string' ? rawMeta.env : undefined,
|
||||
|
||||
@@ -8,6 +8,7 @@ export function listenRelayHttpServer(
|
||||
return new Promise((resolve, reject) => {
|
||||
const onError = (error: Error): void => {
|
||||
server.off('listening', onListening)
|
||||
server.close()
|
||||
reject(error)
|
||||
}
|
||||
const onListening = (): void => {
|
||||
|
||||
Reference in New Issue
Block a user