mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
Keep idle Claude sessions selected during nested Codex work (#29386)
* Preserve idle foreground Claude session ownership through nested hooks * Discard obsolete hook probes and cancel queued restarts
This commit is contained in:
@@ -20968,39 +20968,44 @@
|
||||
"surfaces": [
|
||||
"agent hook normalization",
|
||||
"sidebar child rows",
|
||||
"mobile status snapshots"
|
||||
],
|
||||
"platforms": [
|
||||
"macos",
|
||||
"linux",
|
||||
"windows"
|
||||
],
|
||||
"providers": [
|
||||
"local",
|
||||
"daemon",
|
||||
"ssh",
|
||||
"wsl",
|
||||
"remote-runtime"
|
||||
],
|
||||
"coveredPlatforms": [
|
||||
"macos"
|
||||
"mobile status snapshots",
|
||||
"saved agent resume sessions",
|
||||
"native chat transcript selection"
|
||||
],
|
||||
"platforms": ["macos", "linux", "windows"],
|
||||
"providers": ["local", "daemon", "ssh", "wsl", "remote-runtime"],
|
||||
"coveredPlatforms": ["macos", "linux"],
|
||||
"coveredProviders": [],
|
||||
"coverageNotes": "Contract tests exercise local HTTP hooks, loopback SSH/WSL relay forwarding and legacy relay snapshots. No live SSH, WSL, Linux or Windows execution is claimed. Daemon PTYs use the same hook ingress; folder workspace routing is covered. Hidden macOS Electron validation used the real WorktreeCardAgents component and store in a temporary harness, receiving real hook IPC: 60 nested Codex updates produced zero row mutations, retained 32 children and kept height at 1170px. Screenshots inspected; synthetic hooks, no new model-backed turn.",
|
||||
"coverageNotes": "Contract tests exercise local HTTP hooks, loopback SSH/WSL relay forwarding and legacy relay snapshots. No live SSH, WSL or Windows Electron execution is claimed. Daemon PTYs use the same hook ingress; folder workspace routing is covered. Hidden macOS Electron validation used the real WorktreeCardAgents component and store in a temporary harness, receiving real hook IPC: 60 nested Codex updates produced zero row mutations, retained 32 children and kept height at 1170px. Screenshots inspected; synthetic hooks, no new model-backed turn. Follow-up hidden Electron verification uses an identified foreground PTY stand-in, retains idle Claude native chat through six nested Codex hooks, reads its transcript through a mobile-scoped paired client, and returns to the terminal without a Codex error. Three recorded macOS runs and three nonrecorded Linux-container runs pass. Linux recording mode timed out during Electron startup before test execution; visual proof is from macOS.",
|
||||
"motivatingLinks": [
|
||||
"User report (2026-10-15): Claude status in nwparker/SSH alternates between a 32-child roster and an empty roster when nested Codex reviews report on the inherited terminal pane.",
|
||||
"https://github.com/stablyai/orca/pull/8211",
|
||||
"https://github.com/stablyai/orca/pull/3990"
|
||||
],
|
||||
"invariant": "A different provider inheriting an active pane key cannot replace its child roster, main turn, prompt, resume identity, model or freshness, nor mutate the owning provider caches. Same-provider children still update; finished or stale owners and distinct execution hosts/workspaces permit provider replacement. WSL local and relay transports retain the same owner.",
|
||||
"oracle": "Send a full Claude child roster, interleave Codex review lifecycle/tool/approval/completion hooks, and require identical snapshots plus zero publications. A later Claude child event must retain the original prompt and roster, and a real child completion must shrink it.",
|
||||
"invariant": "A different provider inheriting an active pane key cannot replace its child roster, main turn, prompt, resume identity, model or freshness, nor mutate the owning provider caches. Same-provider children still update. Hosts advertising process ownership retain an identified idle parent until positive exit or suspension, and unavailable reads preserve ownership. Older hosts retain the previous finished/stale replacement policy; distinct execution hosts/workspaces permit replacement. WSL local and relay transports retain the same owner.",
|
||||
"oracle": "Send a full Claude child roster, interleave Codex review lifecycle/tool/approval/completion hooks, and require identical snapshots plus zero publications. A later Claude child event must retain the original prompt and roster, and a real child completion must shrink it. After Claude Stop, keep its mobile transcript and saved resume identity while its identified process remains live or unverifiable; accept the first provider replacement after confirmed exit or suspension. Replay must admit records in order, retain the file on failed admission, and concurrent starts, stop during replay, or a second stop during a queued restart must not leave a listener. Pending owner probes cannot publish or accept an old request after stop, restart, or status disable/re-enable; interrupted replay stays durable.",
|
||||
"commands": [
|
||||
"SKIP_BUILD=1 ORCA_BACKGROUND_LAUNCH=1 xvfb-run -a pnpm exec playwright test --config tests/playwright.config.ts tests/e2e/nested-agent-idle-session-owner.spec.ts --project=electron-headless --workers=1 --repeat-each=3",
|
||||
"ORCA_BACKGROUND_LAUNCH=1 pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/server-idle-process-owner.test.ts src/wsl-guest/relay-idle-process-owner.test.ts tests/e2e/nested-agent-idle-owner.unit.test.ts src/shared/agent-process-presence-probe.test.ts src/shared/agent-hook-spool-drain.test.ts src/main/agent-hooks/server-agent-presence.test.ts src/wsl-guest/relay-agent-presence.test.ts src/main/agent-hooks/server-nested-agent-roster.test.ts src/main/agent-hooks/server-nested-agent-boundaries.test.ts src/main/agent-hooks/server-nested-agent-consumers.test.ts src/main/agent-hooks/server-nested-agent-model.test.ts src/main/agent-hooks/server-ingest-remote.test.ts src/main/agent-hooks/server-hook-http-ingest.test.ts src/main/agent-hooks/server-status-listener-fanout.test.ts src/main/agent-hooks/terminal-handle-row-identity.test.ts src/renderer/src/store/slices/agent-status-nested-resume.test.ts src/wsl-guest/agent-hook-provider-normalization.test.ts src/main/agent-hooks/server-start-failure-lifecycle.test.ts src/main/agent-hooks/server-startup-lifecycle.test.ts src/main/agent-hooks/spool.test.ts src/shared/agent-hook-spool-read.test.ts src/wsl-guest/agent-hook-server.test.ts src/main/agent-hooks/server-http-owner-probe-lifecycle.test.ts src/wsl-guest/relay-http-owner-probe-lifecycle.test.ts --maxWorkers=2",
|
||||
"ORCA_BACKGROUND_LAUNCH=1 pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/server-idle-process-owner.test.ts src/wsl-guest/relay-idle-process-owner.test.ts tests/e2e/nested-agent-idle-owner.unit.test.ts src/shared/agent-process-presence-probe.test.ts src/shared/agent-hook-spool-drain.test.ts src/main/agent-hooks/server-agent-presence.test.ts src/wsl-guest/relay-agent-presence.test.ts src/main/agent-hooks/server-nested-agent-roster.test.ts src/main/agent-hooks/server-nested-agent-boundaries.test.ts src/main/agent-hooks/server-nested-agent-consumers.test.ts src/main/agent-hooks/server-nested-agent-model.test.ts src/main/agent-hooks/server-ingest-remote.test.ts src/main/agent-hooks/server-hook-http-ingest.test.ts src/main/agent-hooks/server-status-listener-fanout.test.ts src/main/agent-hooks/terminal-handle-row-identity.test.ts src/renderer/src/store/slices/agent-status-nested-resume.test.ts src/wsl-guest/agent-hook-provider-normalization.test.ts src/main/agent-hooks/server-start-failure-lifecycle.test.ts src/main/agent-hooks/server-startup-lifecycle.test.ts src/main/agent-hooks/spool.test.ts src/shared/agent-hook-spool-read.test.ts src/wsl-guest/agent-hook-server.test.ts --maxWorkers=2",
|
||||
"ORCA_BACKGROUND_LAUNCH=1 ORCA_E2E_RECORD_VIDEO=1 pnpm exec playwright test --config tests/playwright.config.ts tests/e2e/nested-agent-idle-session-owner.spec.ts --project=electron-headless --workers=1 --repeat-each=3",
|
||||
"ORCA_BACKGROUND_LAUNCH=1 pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/server-idle-process-owner.test.ts src/wsl-guest/relay-idle-process-owner.test.ts tests/e2e/nested-agent-idle-owner.unit.test.ts src/shared/agent-process-presence-probe.test.ts src/shared/agent-hook-spool-drain.test.ts src/main/agent-hooks/server-startup-lifecycle.test.ts src/main/agent-hooks/server-start-failure-lifecycle.test.ts --maxWorkers=2",
|
||||
"ORCA_BACKGROUND_LAUNCH=1 pnpm exec playwright test --config tests/playwright.config.ts tests/e2e/nested-agent-idle-session-owner.spec.ts --project=electron-headless --workers=1",
|
||||
"ORCA_BACKGROUND_LAUNCH=1 pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/server-nested-agent-roster.test.ts src/main/agent-hooks/server-nested-agent-boundaries.test.ts src/main/agent-hooks/server-nested-agent-consumers.test.ts src/main/agent-hooks/server-nested-agent-model.test.ts src/main/agent-hooks/server-ingest-remote.test.ts src/main/agent-hooks/server-hook-http-ingest.test.ts src/main/agent-hooks/server-turn-started-at.test.ts src/main/agent-hooks/server-status-listener-fanout.test.ts src/main/agent-hooks/terminal-handle-row-identity.test.ts src/renderer/src/store/slices/agent-status-tool-assistant-fields.test.ts src/renderer/src/store/slices/agent-status-provider-session.test.ts src/renderer/src/store/slices/agent-status-nested-resume.test.ts src/wsl-guest/agent-hook-provider-normalization.test.ts --maxWorkers=2",
|
||||
"ORCA_BACKGROUND_LAUNCH=1 pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/server-nested-agent-roster.test.ts src/main/agent-hooks/server-nested-agent-model.test.ts src/main/agent-hooks/server-ingest-remote.test.ts src/renderer/src/store/slices/agent-status-tool-assistant-fields.test.ts --maxWorkers=2",
|
||||
"ORCA_BACKGROUND_LAUNCH=1 pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/server-nested-agent-roster.test.ts src/main/agent-hooks/server-nested-agent-model.test.ts src/main/agent-hooks/server-ingest-remote.test.ts src/main/agent-hooks/server-agent-presence.test.ts src/renderer/src/store/slices/agent-status-tool-assistant-fields.test.ts src/main/agent-hooks/server-hook-http-ingest.test.ts src/wsl-guest/agent-hook-server.test.ts --maxWorkers=2",
|
||||
"ORCA_BACKGROUND_LAUNCH=1 pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/server-nested-agent-roster.test.ts src/main/agent-hooks/server-ingest-remote.test.ts src/main/agent-hooks/server-nested-agent-model.test.ts src/main/agent-hooks/server-hook-http-ingest.test.ts src/main/agent-hooks/server-turn-started-at.test.ts src/main/agent-hooks/server-status-listener-fanout.test.ts src/main/agent-hooks/terminal-handle-row-identity.test.ts src/renderer/src/store/slices/agent-status-tool-assistant-fields.test.ts --maxWorkers=2"
|
||||
],
|
||||
"testFiles": [
|
||||
"src/main/agent-hooks/server-http-owner-probe-lifecycle.test.ts",
|
||||
"src/wsl-guest/relay-http-owner-probe-lifecycle.test.ts",
|
||||
"src/shared/agent-hook-spool-drain.test.ts",
|
||||
"src/main/agent-hooks/server-startup-lifecycle.test.ts",
|
||||
"src/main/agent-hooks/server-start-failure-lifecycle.test.ts",
|
||||
"src/main/agent-hooks/server-idle-process-owner.test.ts",
|
||||
"src/wsl-guest/relay-idle-process-owner.test.ts",
|
||||
"tests/e2e/nested-agent-idle-owner.unit.test.ts",
|
||||
"tests/e2e/nested-agent-idle-session-owner.spec.ts",
|
||||
"src/main/agent-hooks/server-status-listener-fanout.test.ts",
|
||||
"src/main/agent-hooks/terminal-handle-row-identity.test.ts",
|
||||
"src/wsl-guest/agent-hook-provider-normalization.test.ts",
|
||||
@@ -21022,6 +21027,51 @@
|
||||
}
|
||||
],
|
||||
"evidenceRuns": [
|
||||
{
|
||||
"date": "2026-10-15",
|
||||
"runner": "local",
|
||||
"platform": "linux",
|
||||
"command": "SKIP_BUILD=1 ORCA_BACKGROUND_LAUNCH=1 xvfb-run -a pnpm exec playwright test --config tests/playwright.config.ts tests/e2e/nested-agent-idle-session-owner.spec.ts --project=electron-headless --workers=1 --repeat-each=3",
|
||||
"result": "passed",
|
||||
"durationSeconds": 51.3,
|
||||
"summary": "Three hidden Electron runs pass in an isolated Docker Ubuntu24 arm64 container on the final generation-guard build using the actual foreground PTY fixture. Native chat, paired mobile transcript, and Show terminal retain Claude identity. Original CI failure reproduced with the worker-PID fixture; corrected foreground fixture passes. Recording-enabled attempts timed out at fixture startup before test execution; attached video is from macOS."
|
||||
},
|
||||
{
|
||||
"date": "2026-10-15",
|
||||
"runner": "local",
|
||||
"platform": "macos",
|
||||
"command": "ORCA_BACKGROUND_LAUNCH=1 pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/server-idle-process-owner.test.ts src/wsl-guest/relay-idle-process-owner.test.ts tests/e2e/nested-agent-idle-owner.unit.test.ts src/shared/agent-process-presence-probe.test.ts src/shared/agent-hook-spool-drain.test.ts src/main/agent-hooks/server-agent-presence.test.ts src/wsl-guest/relay-agent-presence.test.ts src/main/agent-hooks/server-nested-agent-roster.test.ts src/main/agent-hooks/server-nested-agent-boundaries.test.ts src/main/agent-hooks/server-nested-agent-consumers.test.ts src/main/agent-hooks/server-nested-agent-model.test.ts src/main/agent-hooks/server-ingest-remote.test.ts src/main/agent-hooks/server-hook-http-ingest.test.ts src/main/agent-hooks/server-status-listener-fanout.test.ts src/main/agent-hooks/terminal-handle-row-identity.test.ts src/renderer/src/store/slices/agent-status-nested-resume.test.ts src/wsl-guest/agent-hook-provider-normalization.test.ts src/main/agent-hooks/server-start-failure-lifecycle.test.ts src/main/agent-hooks/server-startup-lifecycle.test.ts src/main/agent-hooks/spool.test.ts src/shared/agent-hook-spool-read.test.ts src/wsl-guest/agent-hook-server.test.ts src/main/agent-hooks/server-http-owner-probe-lifecycle.test.ts src/wsl-guest/relay-http-owner-probe-lifecycle.test.ts --maxWorkers=2",
|
||||
"result": "passed",
|
||||
"durationSeconds": 75.69,
|
||||
"summary": "197 tests across 24 files pass after adversarial review. New regressions cover pending HTTP and process-probe publication after stop/restart/status disable/re-enable, fixed-token relay restarts, interrupted spool durability, and a second stop while a restart is queued."
|
||||
},
|
||||
{
|
||||
"date": "2026-10-15",
|
||||
"runner": "local",
|
||||
"platform": "macos",
|
||||
"command": "ORCA_BACKGROUND_LAUNCH=1 ORCA_E2E_RECORD_VIDEO=1 pnpm exec playwright test --config tests/playwright.config.ts tests/e2e/nested-agent-idle-session-owner.spec.ts --project=electron-headless --workers=1 --repeat-each=3",
|
||||
"result": "passed",
|
||||
"durationSeconds": 52.2,
|
||||
"summary": "Three recorded hidden Electron runs pass using an identified stand-in process in the actual foreground PTY. Claude native chat survives six nested Codex hook types, a paired mobile client reads the same transcript, and Show terminal returns without an error. Screenshots inspected and fresh proof attached. Controlled hooks/transcript; no live model turn."
|
||||
},
|
||||
{
|
||||
"date": "2026-10-15",
|
||||
"runner": "local",
|
||||
"platform": "macos",
|
||||
"command": "ORCA_BACKGROUND_LAUNCH=1 pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/server-idle-process-owner.test.ts src/wsl-guest/relay-idle-process-owner.test.ts tests/e2e/nested-agent-idle-owner.unit.test.ts src/shared/agent-process-presence-probe.test.ts src/shared/agent-hook-spool-drain.test.ts src/main/agent-hooks/server-agent-presence.test.ts src/wsl-guest/relay-agent-presence.test.ts src/main/agent-hooks/server-nested-agent-roster.test.ts src/main/agent-hooks/server-nested-agent-boundaries.test.ts src/main/agent-hooks/server-nested-agent-consumers.test.ts src/main/agent-hooks/server-nested-agent-model.test.ts src/main/agent-hooks/server-ingest-remote.test.ts src/main/agent-hooks/server-hook-http-ingest.test.ts src/main/agent-hooks/server-status-listener-fanout.test.ts src/main/agent-hooks/terminal-handle-row-identity.test.ts src/renderer/src/store/slices/agent-status-nested-resume.test.ts src/wsl-guest/agent-hook-provider-normalization.test.ts src/main/agent-hooks/server-start-failure-lifecycle.test.ts src/main/agent-hooks/server-startup-lifecycle.test.ts src/main/agent-hooks/spool.test.ts src/shared/agent-hook-spool-read.test.ts src/wsl-guest/agent-hook-server.test.ts --maxWorkers=2",
|
||||
"result": "passed",
|
||||
"durationSeconds": 8.34,
|
||||
"summary": "187 tests across 22 files pass on the final follow-up, covering idle process ownership, stale suspension/resume/unavailable/exit, legacy hosts, relay cache isolation, saved resume records, concurrent startup and durable ordered spool replay."
|
||||
},
|
||||
{
|
||||
"date": "2026-10-15",
|
||||
"runner": "local",
|
||||
"platform": "macos",
|
||||
"command": "ORCA_BACKGROUND_LAUNCH=1 ORCA_E2E_RECORD_VIDEO=1 pnpm exec playwright test --config tests/playwright.config.ts tests/e2e/nested-agent-idle-session-owner.spec.ts --project=electron-headless --workers=1 --repeat-each=3",
|
||||
"result": "passed",
|
||||
"durationSeconds": 32.1,
|
||||
"summary": "Three hidden Electron runs pass. Actual native chat retains the foreground Claude reply after six nested Codex hooks; a mobile-scoped paired client reads that Claude transcript; Show terminal returns without the Codex verification error. Screenshots visually inspected and video attached to PR. Controlled hooks/transcripts with a real OS process identity; no model-backed turn or physical phone."
|
||||
},
|
||||
{
|
||||
"date": "2026-10-15",
|
||||
"runner": "local",
|
||||
@@ -21060,19 +21110,19 @@
|
||||
},
|
||||
"redGreenEvidence": {
|
||||
"status": "partial",
|
||||
"evidence": "Live incident records alternate between a Claude 32-child roster and a Codex empty roster. Executing the original HEAD identity function accepts a nested working status; the patched function rejects it. Local and relay HTTP roster tests pass and hidden Electron real-component proof preserves all 32 rendered children through 60 nested hooks. No archived CI red/green run yet."
|
||||
"evidence": "Live incident records alternate between a Claude 32-child roster and a Codex empty roster. Executing the original HEAD identity function accepts a nested working status; the patched function rejects it. Local and relay HTTP roster tests pass and hidden Electron real-component proof preserves all 32 rendered children through 60 nested hooks. The idle-owner follow-up reproduced eight failing assertions across three files on the prior production paths and passes with the process-owner policy. Review also reproduced stale-suspension, duplicate-listener, second-stop queued-restart, and stale HTTP/probe publication failures before their fixes. No archived CI red/green run yet."
|
||||
},
|
||||
"performanceBudget": {
|
||||
"required": true,
|
||||
"evidence": "Main reuses the constant-time provider identity decision and publishes no rejected updates. Legacy relays retain provider-specific prompt/tool/previous-status caches, bounded by MAX_CACHED_PANES for each finite provider and cleared on pane teardown or stop; no polling, timers or subprocesses added."
|
||||
"evidence": "Main reuses the constant-time provider identity decision and publishes no rejected updates. Legacy relays retain provider-specific prompt/tool/previous-status caches, bounded by MAX_CACHED_PANES for each finite provider and cleared on pane teardown or stop; no polling or timers added. Cross-provider hooks on updated hosts reuse the bounded process probe before normalization; macOS performs one ps call per non-overlapping conflicting hook (one-second timeout), and concurrent checks share the existing pending promise."
|
||||
},
|
||||
"promotionCriteria": [
|
||||
"Save red/green evidence and CI soak before promotion."
|
||||
],
|
||||
"promotionCriteria": ["Save red/green evidence and CI soak before promotion."],
|
||||
"knownGaps": [
|
||||
"A nested CLI started before any owner is observed may still establish the initial pane identity.",
|
||||
"Same-provider nested sessions remain governed by the existing process/session ownership rules.",
|
||||
"Live SSH/WSL, Linux, Windows and physical mobile validation are not included."
|
||||
"Live SSH/WSL, Windows Electron and physical mobile validation are not included.",
|
||||
"Idle-process protection needs a captured process identity and an updated execution host. Windows and older hosts retain the prior finished/stale fallback.",
|
||||
"No new model-backed Claude/Codex turn was run; hooks and transcript fixtures drive the verification."
|
||||
],
|
||||
"demotionRule": "Keep experimental until CI soak; investigate any roster loss or blocked genuine provider replacement without weakening the snapshot and publication assertions."
|
||||
}
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
import { AgentHookServer } from './server'
|
||||
import { buildBody, postHookEvent } from './server.test-fixtures'
|
||||
import type { probeAgentProcessPresence } from '../../shared/agent-process-presence-probe'
|
||||
|
||||
vi.mock('../telemetry/client', () => ({ track: vi.fn() }))
|
||||
vi.mock('../telemetry/cohort-classifier', () => ({ getCohortAtEmit: () => ({}) }))
|
||||
const probe = vi.hoisted(() => vi.fn<typeof probeAgentProcessPresence>(async () => 'live'))
|
||||
vi.mock('../../shared/agent-process-presence-probe', () => ({ probeAgentProcessPresence: probe }))
|
||||
const servers: AgentHookServer[] = []
|
||||
|
||||
afterEach(() => {
|
||||
for (const server of servers.splice(0)) {
|
||||
server.stop()
|
||||
}
|
||||
probe.mockReset().mockResolvedValue('live')
|
||||
})
|
||||
|
||||
async function idleClaude(server: AgentHookServer, session = 'claude-parent'): Promise<void> {
|
||||
for (const hook_event_name of ['UserPromptSubmit', 'Stop']) {
|
||||
await postHookEvent(
|
||||
server,
|
||||
buildBody(
|
||||
{ hook_event_name, session_id: session, prompt: session },
|
||||
{
|
||||
agentProcess: JSON.stringify({ pid: 4001, platform: 'linux', startTime: 'birth' })
|
||||
}
|
||||
)
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
it.each(['stop', 'restart', 'disable', 're-enable'] as const)(
|
||||
'discards an old hook when %s interrupts its owner probe',
|
||||
async (action) => {
|
||||
const server = new AgentHookServer()
|
||||
servers.push(server)
|
||||
await server.start({ env: 'production' })
|
||||
await idleClaude(server)
|
||||
let release: (() => void) | undefined
|
||||
probe.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
release = () => resolve('exited')
|
||||
})
|
||||
)
|
||||
const pending = postHookEvent(
|
||||
server,
|
||||
buildBody({
|
||||
hook_event_name: 'UserPromptSubmit',
|
||||
session_id: 'obsolete-codex',
|
||||
prompt: 'Obsolete turn'
|
||||
}),
|
||||
'/hook/codex'
|
||||
)
|
||||
await vi.waitFor(() => expect(probe).toHaveBeenCalledOnce())
|
||||
if (action === 'stop' || action === 'restart') {
|
||||
server.stop()
|
||||
if (action === 'restart') {
|
||||
await server.start({ env: 'production' })
|
||||
await idleClaude(server, 'replacement-claude')
|
||||
}
|
||||
} else {
|
||||
server.setStatusHooksEnabled(false)
|
||||
if (action === 're-enable') {
|
||||
server.setStatusHooksEnabled(true)
|
||||
}
|
||||
}
|
||||
const before = server.getStatusSnapshot()
|
||||
const changed = vi.fn()
|
||||
const emitted = vi.fn()
|
||||
server.subscribeStatusChanges(changed)
|
||||
server.setListener(emitted)
|
||||
emitted.mockClear()
|
||||
release?.()
|
||||
expect((await pending).status).toBe(204)
|
||||
expect(server.getStatusSnapshot()).toEqual(before)
|
||||
expect(changed).not.toHaveBeenCalled()
|
||||
expect(emitted).not.toHaveBeenCalled()
|
||||
if (action === 'restart' || action === 're-enable') {
|
||||
probe.mockResolvedValue('exited')
|
||||
await postHookEvent(
|
||||
server,
|
||||
buildBody({
|
||||
hook_event_name: 'UserPromptSubmit',
|
||||
session_id: 'current-codex',
|
||||
prompt: 'Current turn'
|
||||
}),
|
||||
'/hook/codex'
|
||||
)
|
||||
expect(server.getStatusSnapshot()[0]).toMatchObject({
|
||||
agentType: 'codex',
|
||||
providerSession: { id: 'current-codex' }
|
||||
})
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
it('starts a new owner check after re-enabling without waiting on the discarded generation', async () => {
|
||||
const server = new AgentHookServer()
|
||||
servers.push(server)
|
||||
await server.start({ env: 'production' })
|
||||
await idleClaude(server)
|
||||
let release: (() => void) | undefined
|
||||
probe.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
release = () => resolve('live')
|
||||
})
|
||||
)
|
||||
const postCodex = (session: string) =>
|
||||
postHookEvent(
|
||||
server,
|
||||
buildBody({ hook_event_name: 'UserPromptSubmit', session_id: session, prompt: session }),
|
||||
'/hook/codex'
|
||||
)
|
||||
const obsolete = postCodex('obsolete-codex')
|
||||
await vi.waitFor(() => expect(probe).toHaveBeenCalledOnce())
|
||||
server.setStatusHooksEnabled(false)
|
||||
server.setStatusHooksEnabled(true)
|
||||
probe.mockResolvedValue('exited')
|
||||
const current = postCodex('current-codex')
|
||||
try {
|
||||
await vi.waitFor(() => expect(probe).toHaveBeenCalledTimes(2))
|
||||
expect((await current).status).toBe(204)
|
||||
expect(server.getStatusSnapshot()[0].providerSession?.id).toBe('current-codex')
|
||||
} finally {
|
||||
release?.()
|
||||
await Promise.all([obsolete, current])
|
||||
}
|
||||
expect(server.getStatusSnapshot()[0].providerSession?.id).toBe('current-codex')
|
||||
})
|
||||
@@ -0,0 +1,426 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { AgentHookServer } from './server'
|
||||
import { buildBody, PANE, postHookEvent } from './server.test-fixtures'
|
||||
import { buildRuntimeMobileAgentStatus } from '../runtime/runtime-mobile-agent-status-builder'
|
||||
import { getAgentResumeArgv } from '../../shared/agent-session-resume'
|
||||
import type { probeAgentProcessPresence } from '../../shared/agent-process-presence-probe'
|
||||
|
||||
vi.mock('../telemetry/client', () => ({ track: vi.fn() }))
|
||||
vi.mock('../telemetry/cohort-classifier', () => ({ getCohortAtEmit: () => ({}) }))
|
||||
const probe = vi.hoisted(() => vi.fn<typeof probeAgentProcessPresence>(async () => 'live'))
|
||||
vi.mock('../../shared/agent-process-presence-probe', () => ({ probeAgentProcessPresence: probe }))
|
||||
const servers: AgentHookServer[] = []
|
||||
const owner = {
|
||||
agent: 'claude',
|
||||
sessionOwner: true as const,
|
||||
process: { pid: 4001, platform: 'linux', startTime: 'birth-4001' }
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
for (const server of servers.splice(0)) {
|
||||
server.stop()
|
||||
}
|
||||
probe.mockReset().mockResolvedValue('live')
|
||||
vi.restoreAllMocks()
|
||||
})
|
||||
|
||||
async function start(userDataPath?: string): Promise<AgentHookServer> {
|
||||
const server = new AgentHookServer()
|
||||
servers.push(server)
|
||||
await server.start({ env: 'production', userDataPath })
|
||||
return server
|
||||
}
|
||||
|
||||
async function idleClaude(server: AgentHookServer): Promise<void> {
|
||||
for (const hook_event_name of ['UserPromptSubmit', 'Stop']) {
|
||||
await postHookEvent(
|
||||
server,
|
||||
buildBody(
|
||||
{
|
||||
hook_event_name,
|
||||
session_id: 'claude-parent',
|
||||
prompt: 'Foreground task'
|
||||
},
|
||||
{ agentProcess: JSON.stringify(owner.process) }
|
||||
)
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
function mobile(server: AgentHookServer) {
|
||||
return buildRuntimeMobileAgentStatus(
|
||||
null,
|
||||
{
|
||||
type: 'terminal',
|
||||
id: 'tab-1::leaf',
|
||||
parentTabId: 'tab-1',
|
||||
leafId: 'leaf',
|
||||
title: 'Claude',
|
||||
launchAgent: 'claude',
|
||||
isActive: true
|
||||
},
|
||||
null,
|
||||
null,
|
||||
() => server.getStatusSnapshot(),
|
||||
{
|
||||
getPaneKey: () => PANE,
|
||||
getLeaf: () => null,
|
||||
getTrackedTitle: () => null
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
function assertClaudeSession(server: AgentHookServer): void {
|
||||
expect(mobile(server)).toMatchObject({
|
||||
agentStatus: {
|
||||
agentType: 'claude',
|
||||
providerSession: { id: 'claude-parent' }
|
||||
}
|
||||
})
|
||||
const row = server.getStatusSnapshot().find((row) => row.paneKey === PANE)
|
||||
expect(row?.agentType).toBe('claude')
|
||||
if (!row?.providerSession) {
|
||||
throw new Error('Missing foreground session')
|
||||
}
|
||||
expect(getAgentResumeArgv('claude', row.providerSession)).toEqual([
|
||||
'claude',
|
||||
'--resume',
|
||||
'claude-parent'
|
||||
])
|
||||
}
|
||||
|
||||
describe('idle foreground process owns its provider session', () => {
|
||||
it.each(['live', 'unverifiable'] as const)(
|
||||
'rechecks prior suspension before a nested provider when the owner is now %s',
|
||||
async (verdict) => {
|
||||
const server = await start()
|
||||
await idleClaude(server)
|
||||
const emitted = vi.fn()
|
||||
server.setListener(emitted)
|
||||
emitted.mockClear()
|
||||
probe.mockImplementation(async (_identity, _read, _platform, observe) => {
|
||||
observe?.(true)
|
||||
return 'unverifiable'
|
||||
})
|
||||
await server.checkAgentPresence(PANE)
|
||||
expect(emitted.mock.lastCall?.[0].agentPresence?.suspended).toBe(true)
|
||||
probe.mockImplementation(async (_identity, _read, _platform, observe) => {
|
||||
if (verdict === 'live') {
|
||||
observe?.(false)
|
||||
}
|
||||
return verdict
|
||||
})
|
||||
await postHookEvent(
|
||||
server,
|
||||
buildBody({
|
||||
hook_event_name: 'UserPromptSubmit',
|
||||
session_id: 'nested-codex',
|
||||
prompt: 'Background review'
|
||||
}),
|
||||
'/hook/codex'
|
||||
)
|
||||
expect(probe).toHaveBeenCalledTimes(2)
|
||||
assertClaudeSession(server)
|
||||
expect(emitted.mock.lastCall?.[0].agentPresence?.suspended).toBeUndefined()
|
||||
}
|
||||
)
|
||||
|
||||
it.each([false, true])(
|
||||
'keeps interrupted spool replay durable (re-enable: %s)',
|
||||
async (reEnable) => {
|
||||
const directory = mkdtempSync(join(tmpdir(), 'idle-owner-disabled-spool-'))
|
||||
let server: AgentHookServer | undefined
|
||||
try {
|
||||
const first = await start(directory)
|
||||
await idleClaude(first)
|
||||
first.flushStatusPersistSync()
|
||||
first.stop()
|
||||
const spool = join(directory, 'agent-hooks', 'spool')
|
||||
mkdirSync(spool, { recursive: true })
|
||||
const path = join(spool, 'replacement.jsonl')
|
||||
const contents = `${JSON.stringify({
|
||||
paneKey: PANE,
|
||||
tabId: 'tab-1',
|
||||
worktreeId: 'wt-1',
|
||||
source: 'codex',
|
||||
receivedAt: Date.now(),
|
||||
payload: { hook_event_name: 'UserPromptSubmit', session_id: 'replacement' }
|
||||
})}\n`
|
||||
writeFileSync(path, contents)
|
||||
server = new AgentHookServer()
|
||||
servers.push(server)
|
||||
await server.start({
|
||||
env: 'production',
|
||||
userDataPath: directory,
|
||||
statusHooksEnabled: false
|
||||
})
|
||||
let release: (() => void) | undefined
|
||||
probe.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
release = () => resolve('live')
|
||||
})
|
||||
)
|
||||
const log = vi.spyOn(console, 'error').mockImplementation(() => {})
|
||||
server.setStatusHooksEnabled(true)
|
||||
expect(probe).toHaveBeenCalledTimes(1)
|
||||
server.setStatusHooksEnabled(false)
|
||||
if (reEnable) {
|
||||
server.setStatusHooksEnabled(true)
|
||||
}
|
||||
release?.()
|
||||
await vi.waitFor(() => expect(log).toHaveBeenCalled())
|
||||
expect(readFileSync(path, 'utf8')).toBe(contents)
|
||||
expect(server.getStatusSnapshot()[0].agentType).toBe('claude')
|
||||
} finally {
|
||||
server?.stop()
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['live', 'exited'] as const)(
|
||||
'orders spooled replacements after checking a hydrated %s process owner',
|
||||
async (verdict) => {
|
||||
const directory = mkdtempSync(join(tmpdir(), 'idle-owner-spool-'))
|
||||
let restarted: AgentHookServer | undefined
|
||||
try {
|
||||
const first = await start(directory)
|
||||
await idleClaude(first)
|
||||
first.flushStatusPersistSync()
|
||||
first.stop()
|
||||
probe.mockResolvedValue(verdict)
|
||||
const spool = join(directory, 'agent-hooks', 'spool')
|
||||
mkdirSync(spool, { recursive: true })
|
||||
writeFileSync(
|
||||
join(spool, 'replacement.jsonl'),
|
||||
['UserPromptSubmit', 'PreToolUse']
|
||||
.map(
|
||||
(hookEventName) =>
|
||||
`${JSON.stringify({
|
||||
paneKey: PANE,
|
||||
tabId: 'tab-1',
|
||||
worktreeId: 'wt-1',
|
||||
source: 'codex',
|
||||
receivedAt: Date.now(),
|
||||
hookEventName,
|
||||
payload: {
|
||||
hook_event_name: hookEventName,
|
||||
session_id: 'spooled-codex',
|
||||
prompt: 'Replacement task',
|
||||
tool_name: 'shell'
|
||||
}
|
||||
})}\n`
|
||||
)
|
||||
.join('')
|
||||
)
|
||||
restarted = await start(directory)
|
||||
expect(restarted.getStatusSnapshot()[0]).toMatchObject({
|
||||
agentType: verdict === 'live' ? 'claude' : 'codex',
|
||||
providerSession: { id: verdict === 'live' ? 'claude-parent' : 'spooled-codex' }
|
||||
})
|
||||
expect(restarted.getStatusSnapshot()[0].prompt).toBe(
|
||||
verdict === 'live' ? 'Foreground task' : 'Replacement task'
|
||||
)
|
||||
} finally {
|
||||
restarted?.stop()
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
it('accepts the first foreground replacement after positive suspension without claiming exit', async () => {
|
||||
const server = await start()
|
||||
await idleClaude(server)
|
||||
probe.mockImplementation(async (_identity, _read, _platform, observe) => {
|
||||
observe?.(true)
|
||||
return 'unverifiable'
|
||||
})
|
||||
const emitted = vi.fn()
|
||||
server.setListener(emitted)
|
||||
emitted.mockClear()
|
||||
await postHookEvent(
|
||||
server,
|
||||
buildBody({
|
||||
hook_event_name: 'UserPromptSubmit',
|
||||
session_id: 'foreground-codex',
|
||||
prompt: 'New foreground task'
|
||||
}),
|
||||
'/hook/codex'
|
||||
)
|
||||
expect(emitted.mock.calls[0]?.[0].agentPresence).toMatchObject({
|
||||
agent: 'claude',
|
||||
suspended: true
|
||||
})
|
||||
expect(emitted.mock.calls[0]?.[0].agentPresence.ended).toBeUndefined()
|
||||
expect(server.getStatusSnapshot()[0]).toMatchObject({
|
||||
agentType: 'codex',
|
||||
providerSession: { id: 'foreground-codex' }
|
||||
})
|
||||
})
|
||||
|
||||
it.each(['done', 'working'] as const)(
|
||||
'preserves the legacy remote %s replacement policy without host support',
|
||||
async (state) => {
|
||||
const server = await start()
|
||||
const { sessionOwner: _sessionOwner, ...legacyOwner } = owner
|
||||
server.ingestRemote(
|
||||
{
|
||||
paneKey: PANE,
|
||||
agentPresence: legacyOwner,
|
||||
payload: { agentType: 'claude', state, prompt: 'Legacy owner' }
|
||||
},
|
||||
'ssh-old'
|
||||
)
|
||||
server.ingestRemote(
|
||||
{
|
||||
paneKey: PANE,
|
||||
payload: { agentType: 'codex', state: 'working', prompt: 'Replacement' }
|
||||
},
|
||||
'ssh-old'
|
||||
)
|
||||
expect(server.getStatusSnapshot()[0].agentType).toBe(state === 'done' ? 'codex' : 'claude')
|
||||
expect(probe).not.toHaveBeenCalled()
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['live', 'unverifiable'] as const)(
|
||||
'retains mobile and resume selection after turn completion when owner is %s',
|
||||
async (verdict) => {
|
||||
const server = await start()
|
||||
probe.mockResolvedValue(verdict)
|
||||
await idleClaude(server)
|
||||
const before = server.getStatusSnapshot()
|
||||
const emit = vi.fn()
|
||||
server.setListener(emit)
|
||||
emit.mockClear()
|
||||
for (const hook_event_name of [
|
||||
'SessionStart',
|
||||
'UserPromptSubmit',
|
||||
'PreToolUse',
|
||||
'PermissionRequest',
|
||||
'Stop'
|
||||
]) {
|
||||
await postHookEvent(
|
||||
server,
|
||||
buildBody({
|
||||
hook_event_name,
|
||||
session_id: 'nested-codex',
|
||||
prompt: 'Background review',
|
||||
tool_name: 'shell'
|
||||
}),
|
||||
'/hook/codex'
|
||||
)
|
||||
assertClaudeSession(server)
|
||||
expect(server.getStatusSnapshot()).toEqual(before)
|
||||
}
|
||||
expect(probe).toHaveBeenCalledTimes(5)
|
||||
expect(emit).not.toHaveBeenCalled()
|
||||
await postHookEvent(
|
||||
server,
|
||||
buildBody(
|
||||
{ hook_event_name: 'PreToolUse', session_id: 'claude-parent', tool_name: 'Read' },
|
||||
{ agentProcess: JSON.stringify(owner.process) }
|
||||
)
|
||||
)
|
||||
expect(server.getStatusSnapshot()[0].prompt).toBe('Foreground task')
|
||||
}
|
||||
)
|
||||
|
||||
it('accepts the first replacement hook after a positive process exit', async () => {
|
||||
const server = await start()
|
||||
await idleClaude(server)
|
||||
probe.mockResolvedValue('exited')
|
||||
await postHookEvent(
|
||||
server,
|
||||
buildBody({
|
||||
hook_event_name: 'UserPromptSubmit',
|
||||
session_id: 'replacement-codex',
|
||||
prompt: 'New foreground task'
|
||||
}),
|
||||
'/hook/codex'
|
||||
)
|
||||
expect(server.getStatusSnapshot()[0]).toMatchObject({
|
||||
agentType: 'codex',
|
||||
providerSession: { id: 'replacement-codex' },
|
||||
prompt: 'New foreground task'
|
||||
})
|
||||
})
|
||||
|
||||
it('protects an old idle owner without treating elapsed time as process exit', async () => {
|
||||
const server = await start()
|
||||
await idleClaude(server)
|
||||
const now = Date.now()
|
||||
vi.spyOn(Date, 'now').mockReturnValue(now + 24 * 60 * 60 * 1000)
|
||||
await postHookEvent(
|
||||
server,
|
||||
buildBody({ hook_event_name: 'UserPromptSubmit', session_id: 'nested-codex' }),
|
||||
'/hook/codex'
|
||||
)
|
||||
assertClaudeSession(server)
|
||||
})
|
||||
|
||||
it('keeps remote ownership until its execution host reports exit, without probing locally', async () => {
|
||||
const server = await start()
|
||||
const parent = {
|
||||
paneKey: PANE,
|
||||
tabId: 'tab-1',
|
||||
worktreeId: 'wt-1',
|
||||
source: 'claude',
|
||||
hookEventName: 'Stop',
|
||||
agentPresence: owner,
|
||||
providerSession: { key: 'session_id', id: 'claude-parent' },
|
||||
payload: { agentType: 'claude', state: 'done', prompt: 'Foreground task' }
|
||||
}
|
||||
server.ingestRemote(parent, 'ssh-a')
|
||||
const nested = {
|
||||
...parent,
|
||||
source: 'codex',
|
||||
hookEventName: 'UserPromptSubmit',
|
||||
providerSession: { key: 'session_id', id: 'nested-codex' },
|
||||
payload: { agentType: 'codex', state: 'working', prompt: 'Background review' }
|
||||
}
|
||||
server.ingestRemote(nested, 'ssh-a')
|
||||
assertClaudeSession(server)
|
||||
expect(probe).not.toHaveBeenCalled()
|
||||
server.ingestRemote(
|
||||
{
|
||||
...parent,
|
||||
hookEventName: 'AgentProcessExit',
|
||||
providerSessionOnly: true,
|
||||
agentPresence: { ...owner, ended: true }
|
||||
},
|
||||
'ssh-a'
|
||||
)
|
||||
server.ingestRemote({ ...nested, agentPresence: undefined }, 'ssh-a')
|
||||
expect(server.getStatusSnapshot()[0]).toMatchObject({
|
||||
agentType: 'codex',
|
||||
providerSession: { id: 'nested-codex' }
|
||||
})
|
||||
})
|
||||
|
||||
it('does not carry a process owner into a different execution host', async () => {
|
||||
const server = await start()
|
||||
await idleClaude(server)
|
||||
server.ingestRemote(
|
||||
{
|
||||
paneKey: PANE,
|
||||
tabId: 'tab-1',
|
||||
worktreeId: 'wt-1',
|
||||
source: 'codex',
|
||||
providerSession: { key: 'session_id', id: 'remote-codex' },
|
||||
payload: { agentType: 'codex', state: 'working', prompt: 'Remote task' }
|
||||
},
|
||||
'ssh-other'
|
||||
)
|
||||
expect(server.getStatusSnapshot()[0]).toMatchObject({
|
||||
agentType: 'codex',
|
||||
providerSession: { id: 'remote-codex' }
|
||||
})
|
||||
expect(await server.checkAgentPresence(PANE)).toBeNull()
|
||||
expect(probe).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
@@ -20,6 +20,7 @@ vi.mock('../telemetry/client', () => ({ track: trackMock }))
|
||||
vi.mock('../telemetry/cohort-classifier', () => ({ getCohortAtEmit: getCohortAtEmitMock }))
|
||||
|
||||
import { AgentHookServer, _internals } from './server'
|
||||
import { RelayAgentHookServer } from '../../wsl-guest/agent-hook-server'
|
||||
import { makePaneKey } from '../../shared/stable-pane-id'
|
||||
|
||||
const PANE = makePaneKey('tab-lifecycle', '11111111-1111-4111-8111-111111111111')
|
||||
@@ -37,6 +38,54 @@ afterEach(() => {
|
||||
})
|
||||
|
||||
describe('AgentHookServer startup failure lifecycle', () => {
|
||||
it.each(['main', 'relay'] as const)(
|
||||
'shares one %s listener across concurrent startup requests',
|
||||
async (host) => {
|
||||
const directory = mkdtempSync(join(tmpdir(), 'orca-concurrent-hook-start-'))
|
||||
const server =
|
||||
host === 'main'
|
||||
? new AgentHookServer()
|
||||
: new RelayAgentHookServer({ endpointDir: directory, forward: () => undefined })
|
||||
try {
|
||||
await Promise.all([server.start(), server.start()])
|
||||
expect(createServerMock).toHaveBeenCalledTimes(1)
|
||||
} finally {
|
||||
server.stop()
|
||||
for (const result of createServerMock.mock.results) {
|
||||
if (result.type === 'return') {
|
||||
result.value.close()
|
||||
}
|
||||
}
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['main', 'relay'] as const)(
|
||||
'does not revive a stopped %s startup and permits its replacement',
|
||||
async (host) => {
|
||||
const directory = mkdtempSync(join(tmpdir(), 'orca-stopped-hook-start-'))
|
||||
const server =
|
||||
host === 'main'
|
||||
? new AgentHookServer()
|
||||
: new RelayAgentHookServer({ endpointDir: directory, forward: () => undefined })
|
||||
try {
|
||||
const interrupted = server.start()
|
||||
server.stop()
|
||||
await Promise.all([interrupted, server.start()])
|
||||
expect(createServerMock).toHaveBeenCalledTimes(1)
|
||||
} finally {
|
||||
server.stop()
|
||||
for (const result of createServerMock.mock.results) {
|
||||
if (result.type === 'return') {
|
||||
result.value.close()
|
||||
}
|
||||
}
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
it('rolls back only transport on bind failure and preserves owner state through retry', async () => {
|
||||
const userDataPath = mkdtempSync(join(tmpdir(), 'orca-hook-start-failure-'))
|
||||
const persisted = new AgentHookServer()
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
import { existsSync, mkdtempSync, rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { AgentHookServer } from './server'
|
||||
import { RelayAgentHookServer } from '../../wsl-guest/agent-hook-server'
|
||||
import type { drainAgentHookSpoolAsync } from '../../shared/agent-hook-spool'
|
||||
import type * as HookSpoolModule from '../../shared/agent-hook-spool'
|
||||
import { getEndpointFileName } from '../../shared/agent-hook-listener/endpoint-publication'
|
||||
|
||||
const drain = vi.hoisted(() => vi.fn<typeof drainAgentHookSpoolAsync>())
|
||||
vi.mock('../../shared/agent-hook-spool', async (importOriginal) => ({
|
||||
...(await importOriginal<typeof HookSpoolModule>()),
|
||||
drainAgentHookSpoolAsync: drain
|
||||
}))
|
||||
|
||||
const directories: string[] = []
|
||||
const servers: { stop: () => void }[] = []
|
||||
afterEach(() => {
|
||||
for (const server of servers.splice(0)) {
|
||||
server.stop()
|
||||
}
|
||||
for (const directory of directories.splice(0)) {
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
drain.mockReset()
|
||||
})
|
||||
|
||||
function fixture(mode: 'main' | 'relay') {
|
||||
const directory = mkdtempSync(join(tmpdir(), 'hook-startup-'))
|
||||
directories.push(directory)
|
||||
if (mode === 'main') {
|
||||
const server = new AgentHookServer()
|
||||
servers.push(server)
|
||||
return {
|
||||
start: () => server.start({ env: 'production', userDataPath: directory }),
|
||||
stop: () => server.stop(),
|
||||
port: () => Number(server.buildPtyEnv().ORCA_AGENT_HOOK_PORT ?? 0),
|
||||
endpoint: join(directory, 'agent-hooks', getEndpointFileName())
|
||||
}
|
||||
}
|
||||
const server = new RelayAgentHookServer({ endpointDir: directory, forward: vi.fn() })
|
||||
servers.push(server)
|
||||
return {
|
||||
start: () => server.start(),
|
||||
stop: () => server.stop(),
|
||||
port: () => server.getCoordinates().port,
|
||||
endpoint: server.getCoordinates().endpointFilePath
|
||||
}
|
||||
}
|
||||
|
||||
function pauseReplay() {
|
||||
let release: (() => void) | undefined
|
||||
drain.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
release = () => resolve(0)
|
||||
})
|
||||
)
|
||||
return () => release?.()
|
||||
}
|
||||
|
||||
it.each(['main', 'relay'] as const)(
|
||||
'shares one %s startup while replay is pending',
|
||||
async (mode) => {
|
||||
const server = fixture(mode)
|
||||
const release = pauseReplay()
|
||||
const first = server.start()
|
||||
const second = server.start()
|
||||
expect(second).toBe(first)
|
||||
expect(drain).toHaveBeenCalledTimes(1)
|
||||
expect(server.port()).toBe(0)
|
||||
release()
|
||||
await Promise.all([first, second])
|
||||
expect(server.port()).toBeGreaterThan(0)
|
||||
expect(existsSync(server.endpoint)).toBe(true)
|
||||
expect((await fetch(`http://127.0.0.1:${server.port()}`)).status).toBe(404)
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['main', 'relay'] as const)(
|
||||
'does not bind a %s listener after stop interrupts replay',
|
||||
async (mode) => {
|
||||
const server = fixture(mode)
|
||||
const release = pauseReplay()
|
||||
const starting = server.start()
|
||||
server.stop()
|
||||
release()
|
||||
await starting
|
||||
expect(server.port()).toBe(0)
|
||||
expect(existsSync(server.endpoint)).toBe(false)
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['main', 'relay'] as const)(
|
||||
'restarts %s after an interrupted startup settles',
|
||||
async (mode) => {
|
||||
const server = fixture(mode)
|
||||
const release = pauseReplay()
|
||||
const first = server.start()
|
||||
server.stop()
|
||||
drain.mockResolvedValue(0)
|
||||
const restarted = server.start()
|
||||
expect(drain).toHaveBeenCalledTimes(1)
|
||||
release()
|
||||
await Promise.all([first, restarted])
|
||||
expect(drain).toHaveBeenCalledTimes(2)
|
||||
expect(server.port()).toBeGreaterThan(0)
|
||||
expect(existsSync(server.endpoint)).toBe(true)
|
||||
}
|
||||
)
|
||||
|
||||
it('allows a fresh main startup after replay admission rejects', async () => {
|
||||
const server = fixture('main')
|
||||
drain.mockRejectedValueOnce(new Error('admission unavailable'))
|
||||
await expect(server.start()).rejects.toThrow('admission unavailable')
|
||||
expect(server.port()).toBe(0)
|
||||
drain.mockResolvedValue(0)
|
||||
await server.start()
|
||||
expect(server.port()).toBeGreaterThan(0)
|
||||
expect(existsSync(server.endpoint)).toBe(true)
|
||||
})
|
||||
|
||||
it.each(['main', 'relay'] as const)(
|
||||
'cancels a queued %s restart when stopped again before replay settles',
|
||||
async (mode) => {
|
||||
const server = fixture(mode)
|
||||
const release = pauseReplay()
|
||||
const first = server.start()
|
||||
server.stop()
|
||||
const restarted = server.start()
|
||||
server.stop()
|
||||
release()
|
||||
await Promise.all([first, restarted])
|
||||
expect(drain).toHaveBeenCalledTimes(1)
|
||||
expect(server.port()).toBe(0)
|
||||
expect(existsSync(server.endpoint)).toBe(false)
|
||||
}
|
||||
)
|
||||
@@ -9,7 +9,11 @@ import { AgentHookServerLifecycle } from './server-lifecycle'
|
||||
export abstract class AgentHookServerAgentPresence extends AgentHookServerLifecycle {
|
||||
private readonly presenceChecks = new WeakMap<
|
||||
AgentHookEventPayload,
|
||||
Promise<AgentProcessVerdict | null>
|
||||
{
|
||||
token: string
|
||||
statusHookGeneration: number
|
||||
promise: Promise<AgentProcessVerdict | null>
|
||||
}
|
||||
>()
|
||||
|
||||
/** A live hook proves its own process alive; only another process's hook casts doubt on the owner. */
|
||||
@@ -39,14 +43,22 @@ export abstract class AgentHookServerAgentPresence extends AgentHookServerLifecy
|
||||
if (row.connectionId !== null) {
|
||||
return Promise.resolve('unverifiable')
|
||||
}
|
||||
const token = this.token
|
||||
const statusHookGeneration = this.statusHookGeneration
|
||||
const pending = this.presenceChecks.get(row)
|
||||
if (pending) {
|
||||
return pending
|
||||
if (pending?.token === token && pending.statusHookGeneration === statusHookGeneration) {
|
||||
return pending.promise
|
||||
}
|
||||
const presence = row.agentPresence
|
||||
const check = probeAgentProcessPresence(presence.process)
|
||||
let suspended: boolean | undefined
|
||||
const check = probeAgentProcessPresence(presence.process, undefined, undefined, (observed) => {
|
||||
suspended = observed
|
||||
})
|
||||
.then((verdict) => {
|
||||
if (
|
||||
!this.statusHooksEnabled ||
|
||||
this.token !== token ||
|
||||
this.statusHookGeneration !== statusHookGeneration ||
|
||||
this.state.lastStatusByPaneKey.get(resolved) !== row ||
|
||||
row.agentPresence !== presence
|
||||
) {
|
||||
@@ -57,15 +69,23 @@ export abstract class AgentHookServerAgentPresence extends AgentHookServerLifecy
|
||||
preserveResumeIdentity: true,
|
||||
endedPresence: { ...presence, ended: true }
|
||||
})
|
||||
} else if ((suspended === true) !== (presence.suspended === true)) {
|
||||
const { suspended: _suspended, ...unsuspended } = presence
|
||||
this.applyNormalizedStatus({
|
||||
...row,
|
||||
hookEventName: undefined,
|
||||
hasExplicitPrompt: undefined,
|
||||
agentPresence: suspended ? { ...unsuspended, suspended: true } : unsuspended
|
||||
})
|
||||
}
|
||||
return verdict
|
||||
})
|
||||
.finally(() => {
|
||||
if (this.presenceChecks.get(row) === check) {
|
||||
if (this.presenceChecks.get(row)?.promise === check) {
|
||||
this.presenceChecks.delete(row)
|
||||
}
|
||||
})
|
||||
this.presenceChecks.set(row, check)
|
||||
this.presenceChecks.set(row, { token, statusHookGeneration, promise: check })
|
||||
return check
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { buildSpoolHookBody, type SpoolRecord } from '../../../shared/agent-hook-spool'
|
||||
import { checkHookOwnerPresence } from '../../../shared/agent-hook-owner-presence'
|
||||
import { normalizeHookPayload } from '../../../shared/agent-hook-listener'
|
||||
import { isAgentHookSource, type AgentHookSource } from '../../../shared/agent-hook-relay'
|
||||
import {
|
||||
@@ -81,11 +82,33 @@ export abstract class AgentHookServerIngestNormalization extends AgentHookServer
|
||||
}
|
||||
|
||||
// Spool records are durable replay evidence, not a live observation.
|
||||
protected ingestSpoolRecord(record: SpoolRecord): void {
|
||||
protected ingestSpoolRecord(record: SpoolRecord): Promise<void> | void {
|
||||
if (!isAgentHookSource(record.source)) {
|
||||
return
|
||||
}
|
||||
const token = this.token
|
||||
const statusHookGeneration = this.statusHookGeneration
|
||||
const apply = () => {
|
||||
if (
|
||||
!this.statusHooksEnabled ||
|
||||
this.token !== token ||
|
||||
this.statusHookGeneration !== statusHookGeneration
|
||||
) {
|
||||
throw new Error('Status hook spool replay interrupted')
|
||||
}
|
||||
this.applySpoolHook(record, body)
|
||||
}
|
||||
const body = this.normalizeHookBodyPaneKeyAlias(buildSpoolHookBody(record))
|
||||
const pending = checkHookOwnerPresence(this.state, record.source, body, this.env, (paneKey) =>
|
||||
this.checkAgentPresence(paneKey)
|
||||
)
|
||||
return pending ? pending.then(apply) : apply()
|
||||
}
|
||||
|
||||
private applySpoolHook(record: SpoolRecord, body: unknown): void {
|
||||
if (!isAgentHookSource(record.source)) {
|
||||
return
|
||||
}
|
||||
const normalized = this.normalizeLocalHookPayload(record.source, body, true)
|
||||
if (!normalized.event) {
|
||||
return
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { AgentHookStartup, listenForAgentHooks } from '../../../shared/agent-hook-startup'
|
||||
import { checkHookOwnerPresence } from '../../../shared/agent-hook-owner-presence'
|
||||
import { parsePaneKey } from '../../../shared/stable-pane-id'
|
||||
import { createServer, type IncomingMessage, type ServerResponse } from 'node:http'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
@@ -16,14 +18,25 @@ import { trackEmptyPaneKeyHook } from './server-transport-rules'
|
||||
import { AgentHookServerStatusHookLifecycle } from './server-status-hook-lifecycle'
|
||||
import { OPENCODE_STARTUP_PROMPT_CLAIM_PATH } from '../../../shared/opencode-startup-prompt'
|
||||
|
||||
type HookServerStartOptions = {
|
||||
env?: string
|
||||
userDataPath?: string
|
||||
endpointNamespace?: string
|
||||
statusHooksEnabled?: boolean
|
||||
}
|
||||
|
||||
export abstract class AgentHookServerLifecycle extends AgentHookServerStatusHookLifecycle {
|
||||
private readonly startup = new AgentHookStartup()
|
||||
|
||||
/** Start the loopback listener after hydration and spool replay have settled. */
|
||||
async start(options?: {
|
||||
env?: string
|
||||
userDataPath?: string
|
||||
endpointNamespace?: string
|
||||
statusHooksEnabled?: boolean
|
||||
}): Promise<void> {
|
||||
start(options?: HookServerStartOptions): Promise<void> {
|
||||
return this.startup.start((isCurrent) => this.startListener(options, isCurrent))
|
||||
}
|
||||
|
||||
private async startListener(
|
||||
options: HookServerStartOptions | undefined,
|
||||
isCurrent: () => boolean
|
||||
): Promise<void> {
|
||||
if (this.server) {
|
||||
if (options?.statusHooksEnabled !== undefined) {
|
||||
this.setStatusHooksEnabled(options.statusHooksEnabled)
|
||||
@@ -43,7 +56,10 @@ export abstract class AgentHookServerLifecycle extends AgentHookServerStatusHook
|
||||
this.endpointFileWritten = false
|
||||
this.closeStatusSnapshotWriter()
|
||||
if (this.statusHooksEnabled) {
|
||||
this.initializeStatusHookOwner()
|
||||
await this.initializeStatusHookOwner()
|
||||
}
|
||||
if (!isCurrent()) {
|
||||
return
|
||||
}
|
||||
const handleRequest = async (req: IncomingMessage, res: ServerResponse): Promise<void> => {
|
||||
if (req.method !== 'POST') {
|
||||
@@ -57,6 +73,9 @@ export abstract class AgentHookServerLifecycle extends AgentHookServerStatusHook
|
||||
res.end()
|
||||
return
|
||||
}
|
||||
const statusHookGeneration = this.statusHookGeneration
|
||||
const isStatusCurrent = () =>
|
||||
isCurrent() && this.statusHooksEnabled && this.statusHookGeneration === statusHookGeneration
|
||||
// Why: bound request time so a stalled client can't hold a socket open (slowloris).
|
||||
// Why: track our own destroy so the slowloris cap can't be misread as outside interference.
|
||||
let destroyedBySlowlorisCap = false
|
||||
@@ -67,6 +86,11 @@ export abstract class AgentHookServerLifecycle extends AgentHookServerStatusHook
|
||||
const pathname = new URL(req.url ?? '/', 'http://127.0.0.1').pathname
|
||||
try {
|
||||
const body = await readRequestBody(req)
|
||||
if (!isCurrent()) {
|
||||
res.writeHead(204)
|
||||
res.end()
|
||||
return
|
||||
}
|
||||
if (pathname === OPENCODE_STARTUP_PROMPT_CLAIM_PATH) {
|
||||
res.writeHead(200, { 'content-type': 'application/json' })
|
||||
const claim = this.onStartupPromptClaim?.(body)
|
||||
@@ -78,7 +102,13 @@ export abstract class AgentHookServerLifecycle extends AgentHookServerStatusHook
|
||||
)
|
||||
return
|
||||
}
|
||||
if (!this.statusHooksEnabled) {
|
||||
if (!isStatusCurrent()) {
|
||||
res.writeHead(404)
|
||||
res.end()
|
||||
return
|
||||
}
|
||||
await this.initializeStatusHookOwner()
|
||||
if (!isStatusCurrent()) {
|
||||
res.writeHead(404)
|
||||
res.end()
|
||||
return
|
||||
@@ -107,6 +137,14 @@ export abstract class AgentHookServerLifecycle extends AgentHookServerStatusHook
|
||||
res.end()
|
||||
return
|
||||
}
|
||||
await checkHookOwnerPresence(this.state, source, aliasedBody, this.env, (paneKey) =>
|
||||
this.checkAgentPresence(paneKey)
|
||||
)
|
||||
if (!isStatusCurrent()) {
|
||||
res.writeHead(204)
|
||||
res.end()
|
||||
return
|
||||
}
|
||||
const normalized = this.normalizeLocalHookPayload(source, aliasedBody)
|
||||
const statusDisposition = normalized.event
|
||||
? this.getAgentStatusDisposition(normalized.event.paneKey, {
|
||||
@@ -176,32 +214,26 @@ export abstract class AgentHookServerLifecycle extends AgentHookServerStatusHook
|
||||
this.server = createServer((req, res) => {
|
||||
void handleRequest(req, res)
|
||||
})
|
||||
const server = this.server
|
||||
try {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const onStartupError = (err: Error): void => {
|
||||
this.server?.off('listening', onListening)
|
||||
reject(err)
|
||||
}
|
||||
const onListening = (): void => {
|
||||
this.server?.off('error', onStartupError)
|
||||
this.server?.on('error', (err) => {
|
||||
console.error('[agent-hooks] server error', err)
|
||||
})
|
||||
const address = this.server!.address()
|
||||
await listenForAgentHooks(
|
||||
server,
|
||||
0,
|
||||
isCurrent,
|
||||
() => {
|
||||
const address = server.address()
|
||||
if (address && typeof address === 'object') {
|
||||
this.port = address.port
|
||||
}
|
||||
this.maybeWriteEndpointFile()
|
||||
resolve()
|
||||
}
|
||||
this.server!.once('error', onStartupError)
|
||||
this.server!.listen(0, '127.0.0.1', onListening)
|
||||
})
|
||||
},
|
||||
(error) => console.error('[agent-hooks] server error', error)
|
||||
)
|
||||
} catch (error) {
|
||||
this.rollbackTransportStart()
|
||||
throw error
|
||||
}
|
||||
if (this.statusHooksEnabled) {
|
||||
if (isCurrent() && this.statusHooksEnabled) {
|
||||
this.startOpenCodeBinderLoop()
|
||||
}
|
||||
}
|
||||
@@ -217,6 +249,7 @@ export abstract class AgentHookServerLifecycle extends AgentHookServerStatusHook
|
||||
/** `keepStatus`: quit only. The chat teardown after it still reads each chat's subagents and
|
||||
* background tasks from the status store to decide what to offer to resume; exit frees it. */
|
||||
stop(options?: { keepStatus?: boolean }): void {
|
||||
this.startup.stop()
|
||||
// Terminal status may still have a pending write while hook ingress is disabled.
|
||||
if (this.statusHooksEnabled || this.statusPersistTimer) {
|
||||
this.flushStatusPersistSync()
|
||||
|
||||
@@ -93,6 +93,7 @@ export abstract class AgentHookServerState {
|
||||
protected onStartupPromptClaim: ((body: unknown) => boolean | 'pending') | null = null
|
||||
protected clearStartupPromptClaims: (() => void) | null = null
|
||||
protected statusHooksEnabled = true
|
||||
protected statusHookGeneration = 0
|
||||
protected onPaneStatusCleared: PaneStatusClearListener | null = null
|
||||
protected paneStatusClearListeners = new Set<PaneStatusClearListener>()
|
||||
protected statusDropListeners = new Set<StatusDropListener>()
|
||||
@@ -181,7 +182,7 @@ export abstract class AgentHookServerState {
|
||||
): EnrichedAgentHookEventPayload | undefined
|
||||
protected abstract clearTmuxInnerSubjects(paneKey: string): void
|
||||
protected abstract clearTmuxTabSubjects(tabId: string): void
|
||||
protected abstract ingestSpoolRecord(record: SpoolRecord): void
|
||||
protected abstract ingestSpoolRecord(record: SpoolRecord): Promise<void> | void
|
||||
protected abstract emitPaneStatusCleared(clear: AgentStatusClearIpcPayload): void
|
||||
protected abstract buildStatusChangeNotification(): {
|
||||
statuses: AgentHookStatusChangeEntry[]
|
||||
|
||||
@@ -1,11 +1,14 @@
|
||||
import { drainAgentHookSpool, type SpoolRecord } from '../../../shared/agent-hook-spool'
|
||||
import { drainAgentHookSpoolAsync, type SpoolRecord } from '../../../shared/agent-hook-spool'
|
||||
import { AgentHookServerRuntimeEnv } from './server-runtime-env'
|
||||
|
||||
export abstract class AgentHookServerStatusHookLifecycle extends AgentHookServerRuntimeEnv {
|
||||
private statusHookInitialization: Promise<void> | null = null
|
||||
|
||||
setStatusHooksEnabled(enabled: boolean): void {
|
||||
if (enabled === this.statusHooksEnabled) {
|
||||
return
|
||||
}
|
||||
this.statusHookGeneration += 1
|
||||
if (!enabled) {
|
||||
this.flushStatusPersistSync()
|
||||
this.stopOpenCodeBinderLoop()
|
||||
@@ -17,12 +20,23 @@ export abstract class AgentHookServerStatusHookLifecycle extends AgentHookServer
|
||||
}
|
||||
this.statusHooksEnabled = enabled
|
||||
if (enabled && this.server) {
|
||||
this.initializeStatusHookOwner()
|
||||
void this.initializeStatusHookOwner().catch((error: unknown) => {
|
||||
console.error('[agent-hooks] status initialization failed', error)
|
||||
})
|
||||
this.startOpenCodeBinderLoop()
|
||||
}
|
||||
}
|
||||
|
||||
protected initializeStatusHookOwner(): void {
|
||||
protected initializeStatusHookOwner(): Promise<void> {
|
||||
this.statusHookInitialization ??= this.loadStatusHookOwner().finally(() => {
|
||||
this.statusHookInitialization = null
|
||||
})
|
||||
return this.statusHookInitialization
|
||||
}
|
||||
|
||||
private async loadStatusHookOwner(): Promise<void> {
|
||||
const token = this.token
|
||||
const statusHookGeneration = this.statusHookGeneration
|
||||
if (!this.ownerStateInitialized) {
|
||||
// Why: hydrate before binding the listener so an early hook POST runs against a populated map.
|
||||
if (this.lastStatusFilePath) {
|
||||
@@ -32,15 +46,21 @@ export abstract class AgentHookServerStatusHookLifecycle extends AgentHookServer
|
||||
// Drain before binding the listener so replay cannot race a live hook during startup.
|
||||
if (this.endpointDir) {
|
||||
const replayedPaneKeys = new Set<string>()
|
||||
drainAgentHookSpool({
|
||||
await drainAgentHookSpoolAsync({
|
||||
endpointDir: this.endpointDir,
|
||||
getPersistedLaunchTokenHash: (paneKey) =>
|
||||
this.hydratedLaunchTokenHashByPaneKey.get(this.resolvePaneKeyAlias(paneKey)),
|
||||
ingest: (record: SpoolRecord) => {
|
||||
this.ingestSpoolRecord(record)
|
||||
ingest: async (record: SpoolRecord) => {
|
||||
if (this.token !== token || this.statusHookGeneration !== statusHookGeneration) {
|
||||
throw new Error('Status hook spool replay interrupted')
|
||||
}
|
||||
await this.ingestSpoolRecord(record)
|
||||
replayedPaneKeys.add(this.resolvePaneKeyAlias(record.paneKey))
|
||||
}
|
||||
})
|
||||
if (this.token !== token || this.statusHookGeneration !== statusHookGeneration) {
|
||||
return
|
||||
}
|
||||
// Why: the owner may have died while Orca was down; check each replayed pane once.
|
||||
for (const paneKey of replayedPaneKeys) {
|
||||
void this.checkAgentPresence(paneKey)
|
||||
|
||||
@@ -102,7 +102,8 @@ export abstract class AgentHookServerStatusUpdate extends AgentHookServerStatusA
|
||||
agentType: previous.payload.agentType,
|
||||
state: previous.payload.state,
|
||||
updatedAt: previous.receivedAt,
|
||||
restoredUnconfirmed: previous.restoredUnconfirmed
|
||||
restoredUnconfirmed: previous.restoredUnconfirmed,
|
||||
agentPresence: previous.providerSessionOnly ? undefined : previous.agentPresence
|
||||
}
|
||||
: undefined,
|
||||
incoming: terminalOwnedPayload.payload.agentType,
|
||||
|
||||
@@ -173,7 +173,8 @@ export function normalizeHookPayload(
|
||||
agentType: previousStatus.payload.agentType,
|
||||
state: previousStatus.payload.state,
|
||||
updatedAt: previousReceivedAt,
|
||||
restoredUnconfirmed: previousStatus.restoredUnconfirmed
|
||||
restoredUnconfirmed: previousStatus.restoredUnconfirmed,
|
||||
agentPresence: previousStatus.providerSessionOnly ? undefined : previousStatus.agentPresence
|
||||
},
|
||||
incoming: source,
|
||||
now: Date.now()
|
||||
@@ -224,7 +225,9 @@ export function normalizeHookPayload(
|
||||
// Why: presence needs the agent's own process; without it the hook cannot speak for liveness.
|
||||
const agentProcess =
|
||||
source === 'claude' ? readAgentProcessIdentity(record.agentProcess) : undefined
|
||||
const agentPresence = agentProcess ? { agent: source, process: agentProcess } : undefined
|
||||
const agentPresence = agentProcess
|
||||
? { agent: source, process: agentProcess, sessionOwner: true as const }
|
||||
: undefined
|
||||
const sessionEndReason = readString(hookPayloadRecord, 'reason')
|
||||
if (
|
||||
eventName === 'SessionEnd' &&
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
import { parseHookEnvelope } from './agent-hook-listener/hook-envelope'
|
||||
import type { HookListenerState } from './agent-hook-listener/listener-state'
|
||||
import type { AgentHookSource } from './agent-hook-relay'
|
||||
import { hasOtherAgentProcessOwner } from './agent-process-presence'
|
||||
import { sameAgentStatusTerminalOwner } from './agent-status-terminal-owner'
|
||||
|
||||
/** Confirm an old owner's exit on its execution host before accepting a replacement. */
|
||||
export function checkHookOwnerPresence(
|
||||
state: HookListenerState,
|
||||
source: AgentHookSource,
|
||||
body: unknown,
|
||||
env: string,
|
||||
check: (paneKey: string) => Promise<unknown>
|
||||
): Promise<unknown> | undefined {
|
||||
const envelope = parseHookEnvelope(state, source, body, env)
|
||||
if (!envelope) {
|
||||
return
|
||||
}
|
||||
const previous = state.lastStatusByPaneKey.get(envelope.paneKey)
|
||||
// A suspended owner may have resumed without emitting another hook.
|
||||
const presence = previous?.agentPresence
|
||||
if (
|
||||
previous &&
|
||||
!previous.providerSessionOnly &&
|
||||
sameAgentStatusTerminalOwner(
|
||||
{ connectionId: previous.connectionId ?? null, worktreeId: previous.worktreeId },
|
||||
{ connectionId: null, worktreeId: envelope.worktreeId }
|
||||
) &&
|
||||
hasOtherAgentProcessOwner(presence && { ...presence, suspended: undefined }, source)
|
||||
) {
|
||||
return check(envelope.paneKey)
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
@@ -1,18 +1,22 @@
|
||||
import type { AgentHookEventPayload } from './agent-hook-listener/listener-event'
|
||||
import { isSameAgentProcess } from './agent-process-presence'
|
||||
import { sameAgentStatusTerminalOwner } from './agent-status-terminal-owner'
|
||||
|
||||
/** A pane has one owning agent; only the owner's own proven process can end it. */
|
||||
export function transitionHookPresence(
|
||||
incoming: AgentHookEventPayload,
|
||||
previous: AgentHookEventPayload | undefined
|
||||
): AgentHookEventPayload | undefined {
|
||||
if (previous && !sameAgentStatusTerminalOwner(previous, incoming)) {
|
||||
previous = undefined
|
||||
}
|
||||
const recorded = previous?.agentPresence
|
||||
const owner = recorded && !recorded.ended && !previous?.providerSessionOnly ? recorded : undefined
|
||||
const sender = incoming.agentPresence?.process
|
||||
// Why: only an admitted exit is marked ended (Claude's process-ending SessionEnd, or a host-proved
|
||||
// exit); other agents' SessionEnd hooks are ordinary status updates.
|
||||
const exit = incoming.agentPresence?.ended === true
|
||||
if (owner) {
|
||||
if (owner && (exit || !owner.suspended)) {
|
||||
if (exit) {
|
||||
const fromOwner = owner.process && sender && isSameAgentProcess(owner.process, sender)
|
||||
return fromOwner
|
||||
@@ -23,6 +27,9 @@ export function transitionHookPresence(
|
||||
}
|
||||
: undefined
|
||||
}
|
||||
if (sender && owner.process && isSameAgentProcess(sender, owner.process)) {
|
||||
return incoming
|
||||
}
|
||||
// Why: nested agents inherit ORCA_PANE_KEY; their hooks update status, never ownership.
|
||||
return incoming.agentPresence === owner ? incoming : { ...incoming, agentPresence: owner }
|
||||
}
|
||||
@@ -41,5 +48,8 @@ export function transitionHookPresence(
|
||||
if (!agent || agent === 'unknown') {
|
||||
return incoming.agentPresence ? { ...incoming, agentPresence: undefined } : incoming
|
||||
}
|
||||
return { ...incoming, agentPresence: { agent, ...(sender ? { process: sender } : {}) } }
|
||||
return {
|
||||
...incoming,
|
||||
agentPresence: { ...incoming.agentPresence, agent, ...(sender ? { process: sender } : {}) }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
appendFileSync,
|
||||
mkdirSync,
|
||||
mkdtempSync,
|
||||
readFileSync,
|
||||
rmSync,
|
||||
writeFileSync
|
||||
} from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { drainAgentHookSpool, drainAgentHookSpoolAsync } from './agent-hook-spool'
|
||||
|
||||
const directories: string[] = []
|
||||
afterEach(() => {
|
||||
for (const directory of directories.splice(0)) {
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
function fixture() {
|
||||
const endpointDir = mkdtempSync(join(tmpdir(), 'orca-spool-order-'))
|
||||
directories.push(endpointDir)
|
||||
mkdirSync(join(endpointDir, 'spool'))
|
||||
const path = join(endpointDir, 'spool', 'pane.jsonl')
|
||||
const line = (event: string) =>
|
||||
`${JSON.stringify({ paneKey: 'tab:1', source: 'codex', receivedAt: Date.now(), payload: event })}\n`
|
||||
const contents = line('start') + line('tool')
|
||||
writeFileSync(path, contents)
|
||||
return { endpointDir, path, contents, line, getPersistedLaunchTokenHash: () => undefined }
|
||||
}
|
||||
|
||||
it.each(['sync', 'async'] as const)(
|
||||
'retains the complete file after a %s admission fails',
|
||||
async (mode) => {
|
||||
const data = fixture()
|
||||
const ingest = vi.fn((record) => {
|
||||
if (record.payload === 'tool') {
|
||||
throw new Error('admission failed')
|
||||
}
|
||||
})
|
||||
if (mode === 'sync') {
|
||||
expect(() => drainAgentHookSpool({ ...data, ingest })).toThrow('admission failed')
|
||||
} else {
|
||||
await expect(
|
||||
drainAgentHookSpoolAsync({ ...data, ingest: async (record) => ingest(record) })
|
||||
).rejects.toThrow('admission failed')
|
||||
}
|
||||
expect(readFileSync(data.path, 'utf8')).toBe(data.contents)
|
||||
const retried = vi.fn()
|
||||
expect(await drainAgentHookSpoolAsync({ ...data, ingest: retried })).toBe(2)
|
||||
expect(retried.mock.calls.map(([record]) => record.payload)).toEqual(['start', 'tool'])
|
||||
expect(readFileSync(data.path, 'utf8')).toBe('')
|
||||
}
|
||||
)
|
||||
|
||||
it('waits for the owner probe before later records and preserves appended records', async () => {
|
||||
const data = fixture()
|
||||
let release: (() => void) | undefined
|
||||
const probe = new Promise<void>((resolve) => {
|
||||
release = resolve
|
||||
})
|
||||
const admitted: unknown[] = []
|
||||
const ingest = vi.fn(async (record) => {
|
||||
if (record.payload === 'start') {
|
||||
await probe
|
||||
}
|
||||
admitted.push(record.payload)
|
||||
})
|
||||
const draining = drainAgentHookSpoolAsync({ ...data, ingest })
|
||||
expect(ingest).toHaveBeenCalledTimes(1)
|
||||
expect(admitted).toEqual([])
|
||||
expect(readFileSync(data.path, 'utf8')).toBe(data.contents)
|
||||
const appended = data.line('stop')
|
||||
appendFileSync(data.path, appended)
|
||||
release?.()
|
||||
expect(await draining).toBe(2)
|
||||
expect(admitted).toEqual(['start', 'tool'])
|
||||
expect(readFileSync(data.path, 'utf8')).toBe(appended)
|
||||
})
|
||||
@@ -108,16 +108,15 @@ export type SpoolDrainOptions = {
|
||||
}
|
||||
|
||||
/** Drain JSONL files in place; never replace or unlink an inode held by a hook writer. */
|
||||
export function drainAgentHookSpool(options: SpoolDrainOptions): number {
|
||||
const spoolDir = join(options.endpointDir, 'spool')
|
||||
function spoolCandidates(endpointDir: string): { path: string; mtimeMs: number }[] {
|
||||
const spoolDir = join(endpointDir, 'spool')
|
||||
let names: string[]
|
||||
try {
|
||||
names = readdirSync(spoolDir)
|
||||
} catch {
|
||||
return 0
|
||||
return []
|
||||
}
|
||||
const now = options.now ?? Date.now()
|
||||
const candidates = names
|
||||
return names
|
||||
.map((name) => {
|
||||
const path = join(spoolDir, name)
|
||||
try {
|
||||
@@ -132,39 +131,73 @@ export function drainAgentHookSpool(options: SpoolDrainOptions): number {
|
||||
.filter((entry): entry is { path: string; mtimeMs: number } => entry !== null)
|
||||
.sort((a, b) => a.mtimeMs - b.mtimeMs)
|
||||
.slice(0, AGENT_HOOK_SPOOL_MAX_FILES)
|
||||
}
|
||||
|
||||
function consumeSpoolPrefix(path: string, consumed: number): void {
|
||||
try {
|
||||
const fd = openSync(path, 'r+')
|
||||
try {
|
||||
// Keep the inode so a concurrent append handle cannot silently orphan writes.
|
||||
const size = fstatSync(fd).size
|
||||
if (size > consumed) {
|
||||
// Why: a hook may have appended between the read and here; shift the unread tail
|
||||
// to the front instead of truncating to zero, which would erase it.
|
||||
const tail = Buffer.alloc(size - consumed)
|
||||
readSync(fd, tail, 0, tail.length, consumed)
|
||||
writeSync(fd, tail, 0, tail.length, 0)
|
||||
ftruncateSync(fd, tail.length)
|
||||
} else {
|
||||
ftruncateSync(fd, 0)
|
||||
}
|
||||
} finally {
|
||||
closeSync(fd)
|
||||
}
|
||||
} catch {
|
||||
// A concurrently removed or inaccessible file is harmless; the next launch retries it.
|
||||
}
|
||||
}
|
||||
|
||||
function matchesLaunchToken(
|
||||
record: SpoolRecord,
|
||||
options: Pick<SpoolDrainOptions, 'getPersistedLaunchTokenHash'>
|
||||
): boolean {
|
||||
const expected = options.getPersistedLaunchTokenHash(record.paneKey)
|
||||
return !expected || launchTokenHash(record.launchToken) === expected
|
||||
}
|
||||
|
||||
export function drainAgentHookSpool(options: SpoolDrainOptions): number {
|
||||
const now = options.now ?? Date.now()
|
||||
let drained = 0
|
||||
for (const candidate of candidates) {
|
||||
for (const candidate of spoolCandidates(options.endpointDir)) {
|
||||
const { records, consumed } = readSpoolFile(candidate.path, now)
|
||||
for (const record of records) {
|
||||
const expected = options.getPersistedLaunchTokenHash(record.paneKey)
|
||||
const actual = launchTokenHash(record.launchToken)
|
||||
if (expected && actual !== expected) {
|
||||
continue
|
||||
if (matchesLaunchToken(record, options)) {
|
||||
options.ingest({ ...record, isReplay: true })
|
||||
drained += 1
|
||||
}
|
||||
options.ingest({ ...record, isReplay: true })
|
||||
drained += 1
|
||||
}
|
||||
try {
|
||||
const fd = openSync(candidate.path, 'r+')
|
||||
try {
|
||||
// Keep the inode so a concurrent append handle cannot silently orphan writes.
|
||||
const size = fstatSync(fd).size
|
||||
if (size > consumed) {
|
||||
// Why: a hook may have appended between the read and here; shift the unread tail
|
||||
// to the front instead of truncating to zero, which would erase it.
|
||||
const tail = Buffer.alloc(size - consumed)
|
||||
readSync(fd, tail, 0, tail.length, consumed)
|
||||
writeSync(fd, tail, 0, tail.length, 0)
|
||||
ftruncateSync(fd, tail.length)
|
||||
} else {
|
||||
ftruncateSync(fd, 0)
|
||||
}
|
||||
} finally {
|
||||
closeSync(fd)
|
||||
}
|
||||
} catch {
|
||||
// A concurrently removed or inaccessible file is harmless; the next launch retries it.
|
||||
}
|
||||
consumeSpoolPrefix(candidate.path, consumed)
|
||||
}
|
||||
return drained
|
||||
}
|
||||
|
||||
/** Await each admission before consuming a file so owner probes preserve replay order and retry. */
|
||||
export async function drainAgentHookSpoolAsync(
|
||||
options: Omit<SpoolDrainOptions, 'ingest'> & {
|
||||
ingest: (record: SpoolRecord) => Promise<void> | void
|
||||
}
|
||||
): Promise<number> {
|
||||
const now = options.now ?? Date.now()
|
||||
let drained = 0
|
||||
for (const candidate of spoolCandidates(options.endpointDir)) {
|
||||
const { records, consumed } = readSpoolFile(candidate.path, now)
|
||||
for (const record of records) {
|
||||
if (matchesLaunchToken(record, options)) {
|
||||
await options.ingest({ ...record, isReplay: true })
|
||||
drained += 1
|
||||
}
|
||||
}
|
||||
consumeSpoolPrefix(candidate.path, consumed)
|
||||
}
|
||||
return drained
|
||||
}
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
import type { Server } from 'node:http'
|
||||
|
||||
/** Coalesce starts and let stop invalidate replay before it binds a listener. */
|
||||
export class AgentHookStartup {
|
||||
private generation = 0
|
||||
private pending: { generation: number; promise: Promise<void> } | null = null
|
||||
|
||||
start(bind: (isCurrent: () => boolean) => Promise<void>): Promise<void> {
|
||||
const generation = this.generation
|
||||
if (this.pending) {
|
||||
return this.pending.generation === generation
|
||||
? this.pending.promise
|
||||
: this.pending.promise
|
||||
.catch(() => {})
|
||||
.then(() => {
|
||||
if (generation === this.generation) {
|
||||
return this.start(bind)
|
||||
}
|
||||
return undefined
|
||||
})
|
||||
}
|
||||
const isCurrent = () => generation === this.generation
|
||||
const promise = bind(isCurrent)
|
||||
.catch((error: unknown) => {
|
||||
if (isCurrent()) {
|
||||
throw error
|
||||
}
|
||||
})
|
||||
.finally(() => {
|
||||
if (this.pending?.promise === promise) {
|
||||
this.pending = null
|
||||
}
|
||||
})
|
||||
this.pending = { generation, promise }
|
||||
return promise
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
this.generation += 1
|
||||
}
|
||||
}
|
||||
|
||||
export function listenForAgentHooks(
|
||||
server: Server,
|
||||
port: number,
|
||||
isCurrent: () => boolean,
|
||||
onBound: () => void,
|
||||
onRuntimeError: (error: Error) => void
|
||||
): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const cleanup = () => {
|
||||
server.off('close', onClosed)
|
||||
server.off('listening', onListening)
|
||||
server.off('error', onStartupError)
|
||||
}
|
||||
const onClosed = () => {
|
||||
cleanup()
|
||||
resolve()
|
||||
}
|
||||
const onStartupError = (error: Error) => {
|
||||
cleanup()
|
||||
reject(error)
|
||||
}
|
||||
const onListening = () => {
|
||||
cleanup()
|
||||
if (!isCurrent()) {
|
||||
server.close()
|
||||
resolve()
|
||||
return
|
||||
}
|
||||
server.on('error', onRuntimeError)
|
||||
onBound()
|
||||
resolve()
|
||||
}
|
||||
server.once('close', onClosed)
|
||||
server.once('listening', onListening)
|
||||
server.once('error', onStartupError)
|
||||
server.listen(port, '127.0.0.1')
|
||||
})
|
||||
}
|
||||
@@ -8,6 +8,30 @@ import type { AgentProcessIdentity } from './agent-process-presence'
|
||||
const identity: AgentProcessIdentity = { pid: 4242, platform: 'linux', startTime: 'boot:100' }
|
||||
|
||||
describe('single-process presence evidence', () => {
|
||||
it('reports suspension only after matching the process identity', async () => {
|
||||
const observe = vi.fn()
|
||||
const read = vi.fn(async (): Promise<AgentProcessObservation> => ({
|
||||
verdict: 'live',
|
||||
startTime: identity.startTime,
|
||||
zombie: false,
|
||||
stopped: true
|
||||
}))
|
||||
expect(await probeAgentProcessPresence(identity, read, 'linux', observe)).toBe('unverifiable')
|
||||
expect(observe).toHaveBeenCalledExactlyOnceWith(true)
|
||||
observe.mockClear()
|
||||
read.mockResolvedValue({ verdict: 'unverifiable' })
|
||||
expect(await probeAgentProcessPresence(identity, read, 'linux', observe)).toBe('unverifiable')
|
||||
expect(observe).not.toHaveBeenCalled()
|
||||
read.mockResolvedValue({
|
||||
verdict: 'live',
|
||||
startTime: 'another-birth',
|
||||
zombie: false,
|
||||
stopped: true
|
||||
})
|
||||
expect(await probeAgentProcessPresence(identity, read, 'linux', observe)).toBe('exited')
|
||||
expect(observe).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it.each<[AgentProcessObservation, string]>([
|
||||
[{ verdict: 'live', startTime: 'boot:100', zombie: false }, 'live'],
|
||||
[{ verdict: 'live', startTime: 'boot:101', zombie: false }, 'exited'],
|
||||
|
||||
@@ -79,7 +79,8 @@ export async function readAgentProcess(pid: number): Promise<AgentProcessObserva
|
||||
export async function probeAgentProcessPresence(
|
||||
identity: AgentProcessIdentity | undefined,
|
||||
read: (pid: number) => Promise<AgentProcessObservation> = readAgentProcess,
|
||||
platform: NodeJS.Platform = process.platform
|
||||
platform: NodeJS.Platform = process.platform,
|
||||
onSuspensionObserved?: (suspended: boolean) => void
|
||||
): Promise<AgentProcessVerdict> {
|
||||
if (!identity || identity.platform !== platform) {
|
||||
return 'unverifiable'
|
||||
@@ -91,6 +92,7 @@ export async function probeAgentProcessPresence(
|
||||
if (observed.zombie || observed.startTime !== identity.startTime) {
|
||||
return 'exited'
|
||||
}
|
||||
onSuspensionObserved?.(observed.stopped === true)
|
||||
// Why: a suspended (Ctrl-Z) agent still exists but is not running in its terminal.
|
||||
return observed.stopped ? 'unverifiable' : 'live'
|
||||
}
|
||||
|
||||
@@ -12,6 +12,10 @@ export type AgentProcessPresence = {
|
||||
agent: AgentType
|
||||
process?: AgentProcessIdentity
|
||||
ended?: true
|
||||
/** This host verifies process ownership before accepting a different provider. */
|
||||
sessionOwner?: true
|
||||
/** Positive host observation of suspension, distinct from an unavailable process read. */
|
||||
suspended?: true
|
||||
}
|
||||
|
||||
export type AgentProcessVerdict = 'live' | 'unverifiable' | 'exited'
|
||||
@@ -61,10 +65,28 @@ export function readAgentProcessPresence(value: unknown): AgentProcessPresence |
|
||||
return {
|
||||
agent: value.agent,
|
||||
...(process ? { process } : {}),
|
||||
...('ended' in value && value.ended === true ? { ended: true as const } : {})
|
||||
...('ended' in value && value.ended === true ? { ended: true as const } : {}),
|
||||
...('sessionOwner' in value && value.sessionOwner === true
|
||||
? { sessionOwner: true as const }
|
||||
: {}),
|
||||
...('suspended' in value && value.suspended === true ? { suspended: true as const } : {})
|
||||
}
|
||||
}
|
||||
|
||||
export function isSameAgentProcess(a: AgentProcessIdentity, b: AgentProcessIdentity): boolean {
|
||||
return a.pid === b.pid && a.platform === b.platform && a.startTime === b.startTime
|
||||
}
|
||||
|
||||
export function hasOtherAgentProcessOwner(
|
||||
presence: AgentProcessPresence | undefined,
|
||||
incoming: string
|
||||
): boolean {
|
||||
return Boolean(
|
||||
presence?.process &&
|
||||
presence.sessionOwner &&
|
||||
!presence.ended &&
|
||||
!presence.suspended &&
|
||||
presence.agent !== incoming &&
|
||||
presence.agent !== 'unknown'
|
||||
)
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import {
|
||||
type AgentStatusState,
|
||||
type AgentType
|
||||
} from './agent-status-types'
|
||||
import { hasOtherAgentProcessOwner, type AgentProcessPresence } from './agent-process-presence'
|
||||
import { resolveCanonicalPaneAgentIdentity } from './pane-agent-identity-adapter'
|
||||
import type { TuiAgent } from './tui-agent'
|
||||
|
||||
@@ -12,6 +13,7 @@ type ExistingAgentIdentity = {
|
||||
state: AgentStatusState
|
||||
updatedAt: number
|
||||
restoredUnconfirmed?: boolean
|
||||
agentPresence?: AgentProcessPresence
|
||||
}
|
||||
|
||||
type AgentIdentityResolution = {
|
||||
@@ -68,11 +70,14 @@ export function resolveAgentStatusIdentity(args: {
|
||||
hookIsLive: true,
|
||||
completedHookAgent: args.existing.state === 'done' ? (existingAgentType as TuiAgent) : undefined
|
||||
})
|
||||
if (isActiveExistingIdentity(args.existing, args.now, staleAfterMs)) {
|
||||
if (
|
||||
!args.existing.agentPresence?.suspended &&
|
||||
(hasOtherAgentProcessOwner(args.existing.agentPresence, incomingAgentType) ||
|
||||
isActiveExistingIdentity(args.existing, args.now, staleAfterMs))
|
||||
) {
|
||||
return {
|
||||
// Why: child agent CLIs inherit ORCA_PANE_KEY from their parent terminal.
|
||||
// While the parent turn is active, do not let a nested hook steal the
|
||||
// pane's visible identity.
|
||||
// An idle turn does not end its foreground process or resume identity.
|
||||
agentType: existingAgentType,
|
||||
inheritedFromActivePane: true
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { transitionHookPresence } from '../shared/agent-hook-presence-transition'
|
||||
import { isSameAgentProcess } from '../shared/agent-process-presence'
|
||||
import { hasOtherAgentProcessOwner, isSameAgentProcess } from '../shared/agent-process-presence'
|
||||
import { sameAgentStatusTerminalOwner } from '../shared/agent-status-terminal-owner'
|
||||
import { cacheRelayLegacyAgentStatus } from '../shared/agent-status-legacy-relay-cache'
|
||||
import type { HookListenerState } from '../shared/agent-hook-listener/listener-state'
|
||||
import type { AgentHookEventPayload } from '../shared/agent-hook-listener/listener-event'
|
||||
@@ -35,6 +36,14 @@ export function applyRelayHookEvent(
|
||||
return undefined
|
||||
}
|
||||
const previous = host.state.lastStatusByPaneKey.get(incoming.paneKey)
|
||||
if (
|
||||
previous &&
|
||||
!previous.providerSessionOnly &&
|
||||
sameAgentStatusTerminalOwner(previous, incoming) &&
|
||||
hasOtherAgentProcessOwner(previous.agentPresence, incoming.payload.agentType ?? 'unknown')
|
||||
) {
|
||||
return undefined
|
||||
}
|
||||
const cancellation = reconcileRelayClaudeCancel(host.state, previous, incoming, source)
|
||||
if (cancellation.hold) {
|
||||
return previous
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { hasOtherAgentProcessOwner } from '../shared/agent-process-presence'
|
||||
import { sameAgentStatusTerminalOwner } from '../shared/agent-status-terminal-owner'
|
||||
import { normalizeHookPayload } from '../shared/agent-hook-listener'
|
||||
import {
|
||||
clearPaneCacheState,
|
||||
@@ -38,6 +40,15 @@ export class RelayProviderHookNormalization {
|
||||
cacheRelayLegacyAgentStatus(state, event, MAX_CACHED_PANES, (paneKey) =>
|
||||
clearPaneCacheState(providerState, paneKey)
|
||||
)
|
||||
const owner = canonical.lastStatusByPaneKey.get(event.paneKey)
|
||||
if (
|
||||
owner &&
|
||||
!owner.providerSessionOnly &&
|
||||
sameAgentStatusTerminalOwner(owner, event) &&
|
||||
hasOtherAgentProcessOwner(owner.agentPresence, source)
|
||||
) {
|
||||
return event
|
||||
}
|
||||
const prompt = state.lastPromptByPaneKey.get(event.paneKey)
|
||||
const tool = state.lastToolByPaneKey.get(event.paneKey)
|
||||
if (prompt !== undefined) {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { checkHookOwnerPresence } from '../shared/agent-hook-owner-presence'
|
||||
import type { IncomingMessage, ServerResponse } from 'node:http'
|
||||
import { HOOK_REQUEST_SLOWLORIS_MS } from '../shared/agent-hook-listener/listener-limits'
|
||||
import { mergeAgentHookRequestHeaders } from '../shared/agent-hook-listener/hook-envelope'
|
||||
@@ -17,9 +18,11 @@ export async function handleRelayHookRequest(
|
||||
req: IncomingMessage,
|
||||
res: ServerResponse,
|
||||
options: {
|
||||
isCurrent: () => boolean
|
||||
token: string
|
||||
env: string
|
||||
state: HookListenerState
|
||||
checkAgentPresence: (paneKey: string) => Promise<void>
|
||||
providerNormalization: RelayProviderHookNormalization
|
||||
isPaneSurfaceRetired: (paneKey: string) => boolean
|
||||
getAgentLaunchToken: (paneKey: string) => string | undefined
|
||||
@@ -59,12 +62,29 @@ export async function handleRelayHookRequest(
|
||||
return
|
||||
}
|
||||
const body = await readRequestBody(req)
|
||||
if (!options.isCurrent()) {
|
||||
res.writeHead(204)
|
||||
res.end()
|
||||
return
|
||||
}
|
||||
const hookBody = mergeAgentHookRequestHeaders(body, req.headers)
|
||||
if (await options.ingestTmuxHook?.(source, hookBody)) {
|
||||
res.writeHead(204)
|
||||
res.end()
|
||||
return
|
||||
}
|
||||
await checkHookOwnerPresence(
|
||||
options.state,
|
||||
source,
|
||||
hookBody,
|
||||
options.env,
|
||||
options.checkAgentPresence
|
||||
)
|
||||
if (!options.isCurrent()) {
|
||||
res.writeHead(204)
|
||||
res.end()
|
||||
return
|
||||
}
|
||||
let admittedOpenCodeTuiOwner = false
|
||||
const event = options.providerNormalization.normalize(
|
||||
options.state,
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
import { createServer, type IncomingMessage, type Server, type ServerResponse } from 'node:http'
|
||||
import { listenForAgentHooks } from '../shared/agent-hook-startup'
|
||||
|
||||
export async function bindRelayHookServer(options: {
|
||||
preferredPort: number
|
||||
isCurrent: () => boolean
|
||||
handleRequest: (req: IncomingMessage, res: ServerResponse) => Promise<void>
|
||||
setServer: (server: Server | null) => void
|
||||
setPort: (port: number) => void
|
||||
onFallback: () => void
|
||||
}): Promise<void> {
|
||||
const listenOn = async (port: number) => {
|
||||
const server = createServer(options.handleRequest)
|
||||
options.setServer(server)
|
||||
try {
|
||||
await listenForAgentHooks(
|
||||
server,
|
||||
port,
|
||||
options.isCurrent,
|
||||
() => {
|
||||
const address = server.address()
|
||||
if (address && typeof address === 'object') {
|
||||
options.setPort(address.port)
|
||||
}
|
||||
},
|
||||
(error) => process.stderr.write(`[relay-hook-server] server error: ${error.message}\n`)
|
||||
)
|
||||
} catch (error) {
|
||||
options.setServer(null)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
try {
|
||||
await listenOn(options.preferredPort)
|
||||
} catch (error) {
|
||||
if (!options.isCurrent()) {
|
||||
return
|
||||
}
|
||||
// A busy preferred port still permits endpoint-file discovery on an ephemeral port.
|
||||
if (
|
||||
options.preferredPort > 0 &&
|
||||
error instanceof Error &&
|
||||
'code' in error &&
|
||||
error.code === 'EADDRINUSE'
|
||||
) {
|
||||
options.onFallback()
|
||||
await listenOn(0)
|
||||
} else {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,5 @@
|
||||
import { AgentHookStartup } from '../shared/agent-hook-startup'
|
||||
import { bindRelayHookServer } from './agent-hook-server-bind'
|
||||
import {
|
||||
inferRelayClaudeInterrupt,
|
||||
type RelayInterruptHost
|
||||
@@ -12,7 +14,7 @@ import type {
|
||||
} from './agent-hook-server-contract'
|
||||
import { handleRelayHookRequest } from './agent-hook-request'
|
||||
import { RelayAgentPresence } from './relay-agent-presence'
|
||||
import { createServer, type IncomingMessage, type ServerResponse } from 'node:http'
|
||||
import type { createServer } from 'node:http'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import { join } from 'node:path'
|
||||
|
||||
@@ -48,6 +50,7 @@ import { createRelayClaudeTerminalInterrupts } from './claude-terminal-interrupt
|
||||
|
||||
export class RelayAgentHookServer extends RelayAgentHookCanonicalStatus {
|
||||
private server: ReturnType<typeof createServer> | null = null
|
||||
private readonly startup = new AgentHookStartup()
|
||||
private port = 0
|
||||
private token = ''
|
||||
private env: string
|
||||
@@ -104,7 +107,14 @@ export class RelayAgentHookServer extends RelayAgentHookCanonicalStatus {
|
||||
})
|
||||
}
|
||||
|
||||
async start(options: RelayHookServerStartOptions = {}): Promise<void> {
|
||||
start(options: RelayHookServerStartOptions = {}): Promise<void> {
|
||||
return this.startup.start((isCurrent) => this.startListener(options, isCurrent))
|
||||
}
|
||||
|
||||
private async startListener(
|
||||
options: RelayHookServerStartOptions,
|
||||
isCurrent: () => boolean
|
||||
): Promise<void> {
|
||||
if (this.server) {
|
||||
return
|
||||
}
|
||||
@@ -112,19 +122,41 @@ export class RelayAgentHookServer extends RelayAgentHookCanonicalStatus {
|
||||
this.token = this.fixedToken ?? randomUUID()
|
||||
this.endpointFileWritten = false
|
||||
this.portFallbackApplied = false
|
||||
drainRelayHookSpool(this.endpointDir, (record) => this.ingestSpoolRecord(record))
|
||||
try {
|
||||
await this.listenOn(this.preferredPort)
|
||||
} catch (err) {
|
||||
// Why: fall back to an ephemeral port on EADDRINUSE; clients use the endpoint file.
|
||||
if (this.preferredPort > 0 && (err as NodeJS.ErrnoException)?.code === 'EADDRINUSE') {
|
||||
this.portFallbackApplied = true
|
||||
await this.listenOn(0)
|
||||
} else {
|
||||
throw err
|
||||
}
|
||||
await drainRelayHookSpool(this.endpointDir, (record) =>
|
||||
this.ingestSpoolRecord(record, isCurrent)
|
||||
)
|
||||
if (!isCurrent()) {
|
||||
return
|
||||
}
|
||||
if (options.publishEndpoint !== false) {
|
||||
await bindRelayHookServer({
|
||||
preferredPort: this.preferredPort,
|
||||
isCurrent,
|
||||
handleRequest: (req, res) =>
|
||||
handleRelayHookRequest(req, res, {
|
||||
isCurrent,
|
||||
token: this.token,
|
||||
env: this.env,
|
||||
state: this.state,
|
||||
providerNormalization: this.providerNormalization,
|
||||
checkAgentPresence: (paneKey) => this.checkAgentPresence(paneKey),
|
||||
isPaneSurfaceRetired: this.isPaneSurfaceRetired,
|
||||
getAgentLaunchToken: this.getAgentLaunchToken,
|
||||
applyEvent: (event, source, env, version) => this.applyEvent(event, source, env, version),
|
||||
ingestTmuxHook: (source, body) => this.ingestCanonicalTmuxHook(source, body, this.env),
|
||||
retryScheduler: this.retryScheduler,
|
||||
transportInterference: this.transportInterference
|
||||
}),
|
||||
setServer: (server) => {
|
||||
this.server = server
|
||||
},
|
||||
setPort: (port) => {
|
||||
this.port = port
|
||||
},
|
||||
onFallback: () => {
|
||||
this.portFallbackApplied = true
|
||||
}
|
||||
})
|
||||
if (isCurrent() && options.publishEndpoint !== false) {
|
||||
this.publishEndpointFile()
|
||||
}
|
||||
}
|
||||
@@ -133,32 +165,6 @@ export class RelayAgentHookServer extends RelayAgentHookCanonicalStatus {
|
||||
return this.portFallbackApplied
|
||||
}
|
||||
|
||||
private listenOn(port: number): Promise<void> {
|
||||
this.server = createServer((req, res) => this.handleRequest(req, res))
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
const onStartupError = (err: Error): void => {
|
||||
this.server?.off('listening', onListening)
|
||||
// Why: clear failed server refs so later start() calls can retry.
|
||||
this.server = null
|
||||
reject(err)
|
||||
}
|
||||
const onListening = (): void => {
|
||||
this.server?.off('error', onStartupError)
|
||||
this.server?.on('error', (err) => {
|
||||
process.stderr.write(`[relay-hook-server] server error: ${err.message}\n`)
|
||||
})
|
||||
const address = this.server!.address()
|
||||
if (address && typeof address === 'object') {
|
||||
this.port = address.port
|
||||
}
|
||||
resolve()
|
||||
}
|
||||
this.server!.once('error', onStartupError)
|
||||
// Why: loopback only — reachable by the in-box agent CLI (127.0.0.1), not from outside the box.
|
||||
this.server!.listen(port, '127.0.0.1', onListening)
|
||||
})
|
||||
}
|
||||
|
||||
publishEndpointFile(): boolean {
|
||||
if (this.port <= 0 || !this.token) {
|
||||
this.endpointFileWritten = false
|
||||
@@ -176,6 +182,7 @@ export class RelayAgentHookServer extends RelayAgentHookCanonicalStatus {
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
this.startup.stop()
|
||||
this.server?.close()
|
||||
this.server = null
|
||||
this.port = 0
|
||||
@@ -277,21 +284,6 @@ export class RelayAgentHookServer extends RelayAgentHookCanonicalStatus {
|
||||
|
||||
// ─── Private ──────────────────────────────────────────────────────
|
||||
|
||||
private async handleRequest(req: IncomingMessage, res: ServerResponse): Promise<void> {
|
||||
await handleRelayHookRequest(req, res, {
|
||||
token: this.token,
|
||||
env: this.env,
|
||||
state: this.state,
|
||||
providerNormalization: this.providerNormalization,
|
||||
isPaneSurfaceRetired: this.isPaneSurfaceRetired,
|
||||
getAgentLaunchToken: this.getAgentLaunchToken,
|
||||
applyEvent: (event, source, env, version) => this.applyEvent(event, source, env, version),
|
||||
ingestTmuxHook: (source, body) => this.ingestCanonicalTmuxHook(source, body, this.env),
|
||||
retryScheduler: this.retryScheduler,
|
||||
transportInterference: this.transportInterference
|
||||
})
|
||||
}
|
||||
|
||||
private applyEvent(
|
||||
incoming: AgentHookEventPayload,
|
||||
source: AgentHookSource,
|
||||
@@ -319,9 +311,11 @@ export class RelayAgentHookServer extends RelayAgentHookCanonicalStatus {
|
||||
)
|
||||
}
|
||||
|
||||
private ingestSpoolRecord(record: SpoolRecord): void {
|
||||
ingestRelayHookSpoolRecord(record, this.state, this.env, {
|
||||
private ingestSpoolRecord(record: SpoolRecord, isCurrent: () => boolean): Promise<void> | void {
|
||||
return ingestRelayHookSpoolRecord(record, this.state, this.env, {
|
||||
providerNormalization: this.providerNormalization,
|
||||
checkAgentPresence: (paneKey) => this.checkAgentPresence(paneKey),
|
||||
isCurrent,
|
||||
apply: (...args) => this.applyEvent(...args, { isReplay: true }),
|
||||
isPaneSurfaceRetired: this.isPaneSurfaceRetired,
|
||||
getAgentLaunchToken: this.getAgentLaunchToken
|
||||
|
||||
@@ -1,11 +1,12 @@
|
||||
import { admitRelayOpenCodeTui } from './agent-hook-request'
|
||||
import { checkHookOwnerPresence } from '../shared/agent-hook-owner-presence'
|
||||
import { normalizeHookPayload } from '../shared/agent-hook-listener'
|
||||
import type { HookListenerState } from '../shared/agent-hook-listener/listener-state'
|
||||
import type { AgentHookEventPayload } from '../shared/agent-hook-listener/listener-event'
|
||||
import { isAgentHookSource, type AgentHookSource } from '../shared/agent-hook-relay'
|
||||
import {
|
||||
buildSpoolHookBody,
|
||||
drainAgentHookSpool,
|
||||
drainAgentHookSpoolAsync,
|
||||
type SpoolRecord
|
||||
} from '../shared/agent-hook-spool'
|
||||
import { hookBodyEnv, hookBodyVersion } from './agent-hook-envelope-build'
|
||||
@@ -16,7 +17,9 @@ export function ingestRelayHookSpoolRecord(
|
||||
state: HookListenerState,
|
||||
env: string,
|
||||
host: {
|
||||
isCurrent?: () => boolean
|
||||
providerNormalization?: RelayProviderHookNormalization
|
||||
checkAgentPresence?: (paneKey: string) => Promise<void>
|
||||
apply: (
|
||||
event: AgentHookEventPayload,
|
||||
source: AgentHookSource,
|
||||
@@ -26,29 +29,41 @@ export function ingestRelayHookSpoolRecord(
|
||||
isPaneSurfaceRetired: (paneKey: string) => boolean
|
||||
getAgentLaunchToken: (paneKey: string) => string | undefined
|
||||
}
|
||||
): void {
|
||||
): Promise<void> | void {
|
||||
if (!isAgentHookSource(record.source)) {
|
||||
return
|
||||
}
|
||||
const body = buildSpoolHookBody(record)
|
||||
const options: Parameters<typeof normalizeHookPayload>[4] = {
|
||||
admitOpenCodeTui: (identity) => admitRelayOpenCodeTui(host, identity),
|
||||
deferCompactOwnershipToClient: true
|
||||
}
|
||||
const event = host.providerNormalization
|
||||
? host.providerNormalization.normalize(state, record.source, body, env, options)
|
||||
: normalizeHookPayload(state, record.source, body, env, options)
|
||||
if (event) {
|
||||
host.apply(event, record.source, hookBodyEnv(body), hookBodyVersion(body))
|
||||
const apply = () => {
|
||||
if (host.isCurrent && !host.isCurrent()) {
|
||||
throw new Error('Relay hook spool replay interrupted')
|
||||
}
|
||||
if (!isAgentHookSource(record.source)) {
|
||||
return
|
||||
}
|
||||
const options: Parameters<typeof normalizeHookPayload>[4] = {
|
||||
admitOpenCodeTui: (identity) => admitRelayOpenCodeTui(host, identity),
|
||||
deferCompactOwnershipToClient: true
|
||||
}
|
||||
const event = host.providerNormalization
|
||||
? host.providerNormalization.normalize(state, record.source, body, env, options)
|
||||
: normalizeHookPayload(state, record.source, body, env, options)
|
||||
if (event) {
|
||||
host.apply(event, record.source, hookBodyEnv(body), hookBodyVersion(body))
|
||||
}
|
||||
}
|
||||
const pending = host.checkAgentPresence
|
||||
? checkHookOwnerPresence(state, record.source, body, env, host.checkAgentPresence)
|
||||
: undefined
|
||||
return pending ? pending.then(apply) : apply()
|
||||
}
|
||||
|
||||
export function drainRelayHookSpool(
|
||||
export async function drainRelayHookSpool(
|
||||
endpointDir: string,
|
||||
ingest: (record: SpoolRecord) => void
|
||||
): void {
|
||||
ingest: (record: SpoolRecord) => Promise<void> | void
|
||||
): Promise<void> {
|
||||
try {
|
||||
drainAgentHookSpool({
|
||||
await drainAgentHookSpoolAsync({
|
||||
endpointDir,
|
||||
getPersistedLaunchTokenHash: () => undefined,
|
||||
ingest
|
||||
|
||||
@@ -17,7 +17,10 @@ export class RelayAgentPresence {
|
||||
return existing
|
||||
}
|
||||
const presence = row.agentPresence
|
||||
const check = probeAgentProcessPresence(presence.process)
|
||||
let suspended: boolean | undefined
|
||||
const check = probeAgentProcessPresence(presence.process, undefined, undefined, (observed) => {
|
||||
suspended = observed
|
||||
})
|
||||
.then((verdict) => {
|
||||
if (verdict === 'exited' && current() === row) {
|
||||
publish({
|
||||
@@ -25,6 +28,14 @@ export class RelayAgentPresence {
|
||||
hookEventName: 'AgentProcessExit',
|
||||
agentPresence: { ...presence, ended: true }
|
||||
})
|
||||
} else if (current() === row && (suspended === true) !== (presence.suspended === true)) {
|
||||
const { suspended: _suspended, ...unsuspended } = presence
|
||||
publish({
|
||||
...row,
|
||||
hookEventName: undefined,
|
||||
hasExplicitPrompt: undefined,
|
||||
agentPresence: suspended ? { ...unsuspended, suspended: true } : unsuspended
|
||||
})
|
||||
}
|
||||
})
|
||||
.finally(() => this.pending.delete(row))
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { RelayAgentHookServer } from './agent-hook-server'
|
||||
import { makePaneKey } from '../shared/stable-pane-id'
|
||||
import type { AgentHookRelayEnvelope } from '../shared/agent-hook-relay'
|
||||
import type { probeAgentProcessPresence } from '../shared/agent-process-presence-probe'
|
||||
|
||||
const probe = vi.hoisted(() => vi.fn<typeof probeAgentProcessPresence>(async () => 'live'))
|
||||
vi.mock('../shared/agent-process-presence-probe', () => ({ probeAgentProcessPresence: probe }))
|
||||
afterEach(() => probe.mockReset().mockResolvedValue('live'))
|
||||
|
||||
it.each([false, true])(
|
||||
'discards a relay hook after stop during its probe (restart with fixed token: %s)',
|
||||
async (restart) => {
|
||||
const directory = mkdtempSync(join(tmpdir(), 'relay-http-owner-lifecycle-'))
|
||||
const forward = vi.fn<(event: AgentHookRelayEnvelope) => void>()
|
||||
const server = new RelayAgentHookServer({
|
||||
endpointDir: directory,
|
||||
token: 'fixed-host-issued-token',
|
||||
forward
|
||||
})
|
||||
const paneKey = makePaneKey('tab-1', '11111111-1111-4111-8111-111111111111')
|
||||
const post = (source: string, hook_event_name: string, session: string) => {
|
||||
const { port, token } = server.getCoordinates()
|
||||
return fetch(`http://127.0.0.1:${port}/hook/${source}`, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json', 'X-Orca-Agent-Hook-Token': token },
|
||||
body: JSON.stringify({
|
||||
paneKey,
|
||||
tabId: 'tab-1',
|
||||
worktreeId: 'wt-1',
|
||||
...(source === 'claude'
|
||||
? { agentProcess: { pid: 4001, platform: 'linux', startTime: 'birth' } }
|
||||
: {}),
|
||||
payload: { hook_event_name, session_id: session, prompt: session }
|
||||
})
|
||||
})
|
||||
}
|
||||
try {
|
||||
await server.start()
|
||||
for (const event of ['UserPromptSubmit', 'Stop']) {
|
||||
expect((await post('claude', event, 'claude-parent')).status).toBe(204)
|
||||
}
|
||||
let release: (() => void) | undefined
|
||||
probe.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
release = () => resolve('live')
|
||||
})
|
||||
)
|
||||
const pending = post('codex', 'UserPromptSubmit', 'obsolete-codex')
|
||||
await vi.waitFor(() => expect(probe).toHaveBeenCalledOnce())
|
||||
server.stop()
|
||||
if (restart) {
|
||||
await server.start()
|
||||
}
|
||||
forward.mockClear()
|
||||
release?.()
|
||||
expect((await pending).status).toBe(204)
|
||||
expect(forward).not.toHaveBeenCalled()
|
||||
expect(server.replayCachedPayloadsForPanes()).toBe(0)
|
||||
if (restart) {
|
||||
expect((await post('codex', 'UserPromptSubmit', 'current-codex')).status).toBe(204)
|
||||
expect(forward.mock.lastCall?.[0]).toMatchObject({
|
||||
providerSession: { id: 'current-codex' },
|
||||
payload: { agentType: 'codex' }
|
||||
})
|
||||
forward.mockClear()
|
||||
expect(server.replayCachedPayloadsForPanes()).toBe(1)
|
||||
expect(forward.mock.lastCall?.[0].providerSession?.id).toBe('current-codex')
|
||||
}
|
||||
} finally {
|
||||
server.stop()
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
)
|
||||
@@ -0,0 +1,224 @@
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { RelayAgentHookServer } from './agent-hook-server'
|
||||
import type { AgentHookRelayEnvelope } from '../shared/agent-hook-relay'
|
||||
import { makePaneKey } from '../shared/stable-pane-id'
|
||||
import type { probeAgentProcessPresence } from '../shared/agent-process-presence-probe'
|
||||
|
||||
const probe = vi.hoisted(() => vi.fn<typeof probeAgentProcessPresence>(async () => 'live'))
|
||||
vi.mock('../shared/agent-process-presence-probe', () => ({ probeAgentProcessPresence: probe }))
|
||||
afterEach(() => probe.mockReset().mockResolvedValue('live'))
|
||||
|
||||
it.each(['live', 'unverifiable', 'exited'] as const)(
|
||||
'reconciles prior relay suspension when the next process verdict is %s',
|
||||
async (verdict) => {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'relay-resumed-owner-'))
|
||||
const forward = vi.fn<(event: AgentHookRelayEnvelope) => void>()
|
||||
const server = new RelayAgentHookServer({ endpointDir: dir, forward })
|
||||
const paneKey = makePaneKey('tab-1', '11111111-1111-4111-8111-111111111111')
|
||||
try {
|
||||
await server.start()
|
||||
const { port, token } = server.getCoordinates()
|
||||
const post = async (source: string, event: string, session: string) => {
|
||||
const response = await fetch(`http://127.0.0.1:${port}/hook/${source}`, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json', 'X-Orca-Agent-Hook-Token': token },
|
||||
body: JSON.stringify({
|
||||
paneKey,
|
||||
tabId: 'tab-1',
|
||||
worktreeId: 'wt-1',
|
||||
...(source === 'claude'
|
||||
? { agentProcess: { pid: 4001, platform: 'linux', startTime: 'birth' } }
|
||||
: {}),
|
||||
payload: { hook_event_name: event, session_id: session, prompt: `${source} task` }
|
||||
})
|
||||
})
|
||||
expect(response.status).toBe(204)
|
||||
}
|
||||
await post('claude', 'UserPromptSubmit', 'claude-parent')
|
||||
await post('claude', 'Stop', 'claude-parent')
|
||||
probe.mockImplementation(async (_identity, _read, _platform, observe) => {
|
||||
observe?.(true)
|
||||
return 'unverifiable'
|
||||
})
|
||||
await server.checkAgentPresence(paneKey)
|
||||
expect(forward.mock.lastCall?.[0].agentPresence?.suspended).toBe(true)
|
||||
if (verdict === 'exited') {
|
||||
forward.mockClear()
|
||||
probe.mockResolvedValue('exited')
|
||||
await server.checkAgentPresence(paneKey)
|
||||
expect(forward.mock.lastCall?.[0]).toMatchObject({
|
||||
hookEventName: 'AgentProcessExit',
|
||||
providerSessionOnly: true,
|
||||
agentPresence: { ended: true }
|
||||
})
|
||||
server.replayCachedPayloadsForPanes()
|
||||
expect(forward.mock.lastCall?.[0].agentPresence?.ended).toBe(true)
|
||||
return
|
||||
}
|
||||
probe.mockImplementation(async (_identity, _read, _platform, observe) => {
|
||||
if (verdict === 'live') {
|
||||
observe?.(false)
|
||||
}
|
||||
return verdict
|
||||
})
|
||||
await post('codex', 'UserPromptSubmit', 'nested-codex')
|
||||
expect(probe).toHaveBeenCalledTimes(2)
|
||||
expect(forward.mock.lastCall?.[0]).toMatchObject({
|
||||
providerSession: { id: 'claude-parent' },
|
||||
payload: { agentType: 'claude' }
|
||||
})
|
||||
expect(forward.mock.lastCall?.[0].agentPresence?.suspended).toBeUndefined()
|
||||
} finally {
|
||||
server.stop()
|
||||
await rm(dir, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['live', 'unverifiable', 'suspended'] as const)(
|
||||
'keeps an idle %s foreground session in relay replay, then accepts a proved replacement',
|
||||
async (verdict) => {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'relay-idle-owner-'))
|
||||
const forward = vi.fn<(event: AgentHookRelayEnvelope) => void>()
|
||||
const server = new RelayAgentHookServer({ endpointDir: dir, forward })
|
||||
const paneKey = makePaneKey('tab-1', '11111111-1111-4111-8111-111111111111')
|
||||
try {
|
||||
await server.start()
|
||||
const { port, token } = server.getCoordinates()
|
||||
const post = async (source: string, event: string, session: string) => {
|
||||
const response = await fetch(`http://127.0.0.1:${port}/hook/${source}`, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json', 'X-Orca-Agent-Hook-Token': token },
|
||||
body: JSON.stringify({
|
||||
paneKey,
|
||||
tabId: 'tab-1',
|
||||
worktreeId: 'wt-1',
|
||||
...(source === 'claude'
|
||||
? { agentProcess: { pid: 4001, platform: 'linux', startTime: 'birth' } }
|
||||
: {}),
|
||||
payload: {
|
||||
hook_event_name: event,
|
||||
session_id: session,
|
||||
...(event === 'UserPromptSubmit' ? { prompt: `${source} task` } : {})
|
||||
}
|
||||
})
|
||||
})
|
||||
expect(response.status).toBe(204)
|
||||
}
|
||||
await post('claude', 'UserPromptSubmit', 'claude-parent')
|
||||
await post('claude', 'Stop', 'claude-parent')
|
||||
forward.mockClear()
|
||||
if (verdict === 'suspended') {
|
||||
probe.mockImplementation(async (_identity, _read, _platform, observe) => {
|
||||
observe?.(true)
|
||||
return 'unverifiable'
|
||||
})
|
||||
await post('codex', 'UserPromptSubmit', 'foreground-codex')
|
||||
expect(forward.mock.calls[0]?.[0].agentPresence).toMatchObject({
|
||||
agent: 'claude',
|
||||
suspended: true
|
||||
})
|
||||
expect(forward.mock.calls[0]?.[0].agentPresence?.ended).toBeUndefined()
|
||||
expect(forward.mock.lastCall?.[0]).toMatchObject({
|
||||
providerSession: { id: 'foreground-codex' },
|
||||
payload: { agentType: 'codex' }
|
||||
})
|
||||
expect(server.replayCachedPayloadsForPanes()).toBe(1)
|
||||
expect(forward.mock.lastCall?.[0]).toMatchObject({
|
||||
providerSession: { id: 'foreground-codex' }
|
||||
})
|
||||
return
|
||||
}
|
||||
probe.mockResolvedValue(verdict)
|
||||
for (const event of ['SessionStart', 'UserPromptSubmit', 'Stop']) {
|
||||
await post('codex', event, 'nested-codex')
|
||||
}
|
||||
expect(forward).not.toHaveBeenCalled()
|
||||
expect(probe).toHaveBeenCalledTimes(3)
|
||||
expect(server.replayCachedPayloadsForPanes()).toBe(1)
|
||||
expect(forward.mock.lastCall?.[0]).toMatchObject({
|
||||
isReplay: true,
|
||||
providerSession: { id: 'claude-parent' },
|
||||
payload: { agentType: 'claude', state: 'done', prompt: 'claude task' }
|
||||
})
|
||||
await post('claude', 'PreToolUse', 'claude-parent')
|
||||
expect(forward.mock.lastCall?.[0].payload.prompt).toBe('claude task')
|
||||
probe.mockResolvedValue('exited')
|
||||
await post('codex', 'UserPromptSubmit', 'replacement-codex')
|
||||
expect(forward.mock.calls.at(-2)?.[0]).toMatchObject({
|
||||
hookEventName: 'AgentProcessExit',
|
||||
providerSessionOnly: true,
|
||||
providerSession: { id: 'claude-parent' },
|
||||
agentPresence: { ended: true }
|
||||
})
|
||||
expect(forward.mock.lastCall?.[0]).toMatchObject({
|
||||
providerSession: { id: 'replacement-codex' },
|
||||
payload: { agentType: 'codex', prompt: 'codex task' }
|
||||
})
|
||||
server.replayCachedPayloadsForPanes()
|
||||
expect(forward.mock.lastCall?.[0]).toMatchObject({
|
||||
isReplay: true,
|
||||
providerSession: { id: 'replacement-codex' }
|
||||
})
|
||||
} finally {
|
||||
server.stop()
|
||||
await rm(dir, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['live', 'exited'] as const)(
|
||||
'checks a %s owner before admitting ordered cold relay spool replacements',
|
||||
async (verdict) => {
|
||||
const directory = await mkdtemp(join(tmpdir(), 'relay-owner-spool-'))
|
||||
const forward = vi.fn<(event: AgentHookRelayEnvelope) => void>()
|
||||
const server = new RelayAgentHookServer({ endpointDir: directory, forward })
|
||||
const paneKey = makePaneKey('tab-1', '11111111-1111-4111-8111-111111111111')
|
||||
try {
|
||||
await mkdir(join(directory, 'spool'))
|
||||
const records = [
|
||||
['claude', 'UserPromptSubmit'],
|
||||
['claude', 'Stop'],
|
||||
['codex', 'UserPromptSubmit'],
|
||||
['codex', 'PreToolUse']
|
||||
]
|
||||
.map(
|
||||
([source, hookEventName]) =>
|
||||
`${JSON.stringify({
|
||||
paneKey,
|
||||
tabId: 'tab-1',
|
||||
worktreeId: 'wt-1',
|
||||
source,
|
||||
hookEventName,
|
||||
receivedAt: Date.now(),
|
||||
...(source === 'claude'
|
||||
? { agentProcess: { pid: 4001, platform: 'linux', startTime: 'birth' } }
|
||||
: {}),
|
||||
payload: {
|
||||
hook_event_name: hookEventName,
|
||||
session_id: `${source}-session`,
|
||||
prompt: `${source} task`,
|
||||
tool_name: 'shell'
|
||||
}
|
||||
})}\n`
|
||||
)
|
||||
.join('')
|
||||
await writeFile(join(directory, 'spool', 'pane.jsonl'), records)
|
||||
probe.mockResolvedValue(verdict)
|
||||
await server.start()
|
||||
expect(forward.mock.lastCall?.[0]).toMatchObject({
|
||||
providerSession: { id: `${verdict === 'live' ? 'claude' : 'codex'}-session` },
|
||||
payload: { agentType: verdict === 'live' ? 'claude' : 'codex' }
|
||||
})
|
||||
if (verdict === 'exited') {
|
||||
expect(forward.mock.lastCall?.[0].hookEventName).toBe('PreToolUse')
|
||||
}
|
||||
} finally {
|
||||
server.stop()
|
||||
await rm(directory, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
)
|
||||
@@ -64,12 +64,16 @@ export function createRetentionFixtureDirectory(): string {
|
||||
|
||||
/** Long-lived stand-in for the agent a CLI dispatch spawns: it never exits, so
|
||||
* a pruned tab is never explained away by a dead PTY. */
|
||||
export function writeRetentionFixture(directory: string): string {
|
||||
export function writeRetentionFixture(
|
||||
directory: string,
|
||||
options: { processTitle?: string } = {}
|
||||
): string {
|
||||
const fixturePath = path.join(directory, 'host-created-agent-terminal.mjs')
|
||||
writeFileSync(
|
||||
fixturePath,
|
||||
[
|
||||
"import { appendFileSync } from 'node:fs'",
|
||||
...(options.processTitle ? [`process.title = ${JSON.stringify(options.processTitle)}`] : []),
|
||||
'const sink = process.argv[2]',
|
||||
'const record = (line) => appendFileSync(sink, `${line}\\n`)',
|
||||
'record(`READY:${process.pid}`)',
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
import { createTestStore } from '../../src/renderer/src/store/slices/store-test-helpers'
|
||||
import { sleepingRecordFromEntry } from '../../src/renderer/src/store/slices/agent-status-sleeping-records'
|
||||
import { getAgentResumeArgv } from '../../src/shared/agent-session-resume'
|
||||
import { readAgentProcess } from '../../src/shared/agent-process-presence-probe'
|
||||
import { AgentHookServer } from '../../src/main/agent-hooks/server'
|
||||
import { buildBody, PANE, postHookEvent } from '../../src/main/agent-hooks/server.test-fixtures'
|
||||
|
||||
vi.mock('../../src/main/telemetry/client', () => ({ track: vi.fn() }))
|
||||
vi.mock('../../src/main/telemetry/cohort-classifier', () => ({ getCohortAtEmit: () => ({}) }))
|
||||
afterEach(() => vi.useRealTimers())
|
||||
|
||||
it.skipIf(process.platform !== 'darwin' && process.platform !== 'linux')(
|
||||
'captures Claude resume from real host publications after idle nested Codex hooks with a live OS process',
|
||||
async () => {
|
||||
const observed = await readAgentProcess(process.pid)
|
||||
if (observed.verdict !== 'live') {
|
||||
throw new Error('Could not identify the test process')
|
||||
}
|
||||
const identity = { pid: process.pid, platform: process.platform, startTime: observed.startTime }
|
||||
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
|
||||
const store = createTestStore()
|
||||
const server = new AgentHookServer()
|
||||
server.setListener((event) => {
|
||||
store
|
||||
.getState()
|
||||
.setAgentStatus(
|
||||
event.paneKey,
|
||||
event.payload,
|
||||
undefined,
|
||||
{ updatedAt: event.receivedAt },
|
||||
{ worktreeId: 'wt-1', tabId: 'tab-1' },
|
||||
{ providerSession: event.providerSession }
|
||||
)
|
||||
})
|
||||
await server.start({ env: 'production' })
|
||||
try {
|
||||
for (const hook_event_name of ['UserPromptSubmit', 'Stop']) {
|
||||
await postHookEvent(
|
||||
server,
|
||||
buildBody(
|
||||
{ hook_event_name, session_id: 'claude-parent', prompt: 'Foreground task' },
|
||||
{ agentProcess: JSON.stringify(identity) }
|
||||
)
|
||||
)
|
||||
}
|
||||
expect(store.getState().agentStatusByPaneKey[PANE].state).toBe('done')
|
||||
for (const hook_event_name of ['SessionStart', 'UserPromptSubmit', 'Stop']) {
|
||||
await postHookEvent(
|
||||
server,
|
||||
buildBody({ hook_event_name, session_id: 'nested-codex', prompt: 'Background review' }),
|
||||
'/hook/codex'
|
||||
)
|
||||
const record = sleepingRecordFromEntry({
|
||||
state: store.getState(),
|
||||
entry: store.getState().agentStatusByPaneKey[PANE],
|
||||
worktreeId: 'wt-1',
|
||||
capturedAt: Date.now()
|
||||
})
|
||||
expect(record).toMatchObject({ agent: 'claude', providerSession: { id: 'claude-parent' } })
|
||||
if (!record) {
|
||||
throw new Error('Missing foreground resume record')
|
||||
}
|
||||
expect(getAgentResumeArgv(record.agent, record.providerSession)).toEqual([
|
||||
'claude',
|
||||
'--resume',
|
||||
'claude-parent'
|
||||
])
|
||||
}
|
||||
} finally {
|
||||
server.stop()
|
||||
}
|
||||
}
|
||||
)
|
||||
@@ -0,0 +1,168 @@
|
||||
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { test, expect } from './helpers/orca-app'
|
||||
import { ensureTerminalVisible, waitForActiveWorktree, waitForSessionReady } from './helpers/store'
|
||||
import {
|
||||
execInTerminal,
|
||||
waitForActivePanePtyId,
|
||||
waitForActivePaneHookDescriptor,
|
||||
waitForActiveTerminalManager
|
||||
} from './helpers/terminal'
|
||||
import {
|
||||
readSink,
|
||||
retentionFixtureCommand,
|
||||
writeRetentionFixture
|
||||
} from './helpers/host-created-terminal-retention-oracle'
|
||||
import { readHookEndpoint } from './helpers/agent-hook-endpoint'
|
||||
import {
|
||||
claudeTranscriptLines,
|
||||
enableNativeChatSetting,
|
||||
toggleTerminalTabToChatView
|
||||
} from './helpers/native-chat-transcript-fixture'
|
||||
import { createDesktopMobilePairingOffer } from './helpers/paired-electron-client'
|
||||
import { pairMobileClient } from './helpers/paired-mobile-client'
|
||||
import { readAgentProcess } from '../../src/shared/agent-process-presence-probe'
|
||||
import { resolveMobileNativeChat } from '../../mobile/src/session/mobile-native-chat-eligibility'
|
||||
import type { RuntimeMobileSessionTabsResult } from '../../src/shared/runtime-session-contracts'
|
||||
|
||||
const SESSION = '11111111-2222-4333-8444-555555555555'
|
||||
const REPLY = 'Foreground Claude reply stays selected while a nested Codex reviews.'
|
||||
|
||||
test('idle Claude keeps chat, mobile transcript, and return identity through nested Codex hooks', async ({
|
||||
orcaPage,
|
||||
electronApp,
|
||||
registerPostElectronShutdownCleanup
|
||||
}, testInfo) => {
|
||||
test.skip(process.platform === 'win32', 'Claude hook process capture is POSIX-only')
|
||||
const dir = mkdtempSync(join(tmpdir(), 'orca-idle-owner-e2e-'))
|
||||
registerPostElectronShutdownCleanup(async () => rmSync(dir, { recursive: true, force: true }))
|
||||
const transcriptPath = join(dir, `${SESSION}.jsonl`)
|
||||
writeFileSync(
|
||||
transcriptPath,
|
||||
claudeTranscriptLines({
|
||||
sessionId: SESSION,
|
||||
userText: 'Keep my foreground session.',
|
||||
assistantText: REPLY
|
||||
})
|
||||
)
|
||||
await waitForSessionReady(orcaPage)
|
||||
await waitForActiveWorktree(orcaPage)
|
||||
await ensureTerminalVisible(orcaPage)
|
||||
await waitForActiveTerminalManager(orcaPage, 30_000)
|
||||
const descriptor = await waitForActivePaneHookDescriptor(orcaPage)
|
||||
const ptyId = await waitForActivePanePtyId(orcaPage)
|
||||
const sinkPath = join(dir, 'foreground-agent.log')
|
||||
const fixturePath = writeRetentionFixture(dir, { processTitle: 'claude' })
|
||||
await execInTerminal(orcaPage, ptyId, retentionFixtureCommand(fixturePath, sinkPath))
|
||||
await expect.poll(() => readSink(sinkPath)).toMatch(/^READY:\d+/)
|
||||
const pid = Number(readSink(sinkPath).match(/^READY:(\d+)/)?.[1])
|
||||
const observed = await readAgentProcess(pid)
|
||||
expect(observed.verdict).toBe('live')
|
||||
if (observed.verdict !== 'live') {
|
||||
throw new Error('The foreground fixture must have a verifiable identity')
|
||||
}
|
||||
const processIdentity = { pid, platform: process.platform, startTime: observed.startTime }
|
||||
const [tabId] = descriptor.paneKey.split(':')
|
||||
const endpoint = await readHookEndpoint(electronApp)
|
||||
const post = async (source: string, hook_event_name: string) => {
|
||||
const response = await fetch(`http://127.0.0.1:${endpoint.port}/hook/${source}`, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json', 'X-Orca-Agent-Hook-Token': endpoint.token },
|
||||
body: JSON.stringify({
|
||||
...descriptor,
|
||||
tabId,
|
||||
env: endpoint.env,
|
||||
version: endpoint.version,
|
||||
...(source === 'claude' ? { agentProcess: processIdentity } : {}),
|
||||
payload: {
|
||||
hook_event_name,
|
||||
session_id: source === 'claude' ? SESSION : '66666666-7777-4888-8999-000000000000',
|
||||
...(source === 'claude' ? { transcript_path: transcriptPath } : {}),
|
||||
...(hook_event_name === 'UserPromptSubmit'
|
||||
? {
|
||||
prompt: source === 'claude' ? 'Keep my foreground session.' : 'Nested Codex review'
|
||||
}
|
||||
: {})
|
||||
}
|
||||
})
|
||||
})
|
||||
expect(response.status).toBe(204)
|
||||
}
|
||||
await post('claude', 'UserPromptSubmit')
|
||||
await post('claude', 'Stop')
|
||||
await expect
|
||||
.poll(() =>
|
||||
orcaPage.evaluate(
|
||||
(key) => window.__store?.getState().agentStatusByPaneKey[key]?.state,
|
||||
descriptor.paneKey
|
||||
)
|
||||
)
|
||||
.toBe('done')
|
||||
await enableNativeChatSetting(orcaPage)
|
||||
await toggleTerminalTabToChatView(orcaPage, { tabId, worktreeId: descriptor.worktreeId })
|
||||
const chat = orcaPage.locator('[data-native-chat-root="true"]')
|
||||
await expect(chat.getByText(REPLY, { exact: true })).toBeVisible()
|
||||
await orcaPage.screenshot({ path: testInfo.outputPath('01-claude-before-nested.png') })
|
||||
expect(
|
||||
await orcaPage.evaluate(
|
||||
(key) => window.api.agentStatus.hasVerifiableAgentProcess(key),
|
||||
descriptor.paneKey
|
||||
)
|
||||
).toBe(true)
|
||||
for (const event of [
|
||||
'SessionStart',
|
||||
'UserPromptSubmit',
|
||||
'PreToolUse',
|
||||
'PostToolUse',
|
||||
'PermissionRequest',
|
||||
'Stop'
|
||||
]) {
|
||||
await post('codex', event)
|
||||
await expect(chat.getByText(REPLY, { exact: true })).toBeVisible()
|
||||
await expect
|
||||
.poll(() =>
|
||||
orcaPage.evaluate(
|
||||
(key) => window.__store?.getState().agentStatusByPaneKey[key]?.providerSession?.id,
|
||||
descriptor.paneKey
|
||||
)
|
||||
)
|
||||
.toBe(SESSION)
|
||||
}
|
||||
await orcaPage.screenshot({ path: testInfo.outputPath('02-claude-after-nested.png') })
|
||||
|
||||
const phone = pairMobileClient(await createDesktopMobilePairingOffer(orcaPage))
|
||||
try {
|
||||
const result = await phone.call<RuntimeMobileSessionTabsResult>('session.tabs.list', {
|
||||
worktree: `id:${descriptor.worktreeId}`
|
||||
})
|
||||
const tab = result.tabs.find((tab) => tab.type === 'terminal' && tab.parentTabId === tabId)
|
||||
expect(tab).toBeDefined()
|
||||
const resolution = resolveMobileNativeChat(tab ?? null, true)
|
||||
expect(resolution).toMatchObject({ agent: 'claude', sessionId: SESSION })
|
||||
if (!resolution) {
|
||||
throw new Error('Phone lost the foreground transcript')
|
||||
}
|
||||
const transcript = await phone.call<unknown>('nativeChat.readSession', {
|
||||
...resolution,
|
||||
limit: 50
|
||||
})
|
||||
expect(JSON.stringify(transcript)).toContain(REPLY)
|
||||
expect(JSON.stringify(transcript)).not.toContain('Nested Codex review')
|
||||
} finally {
|
||||
phone.dispose()
|
||||
}
|
||||
|
||||
await orcaPage.getByRole('button', { name: 'Show terminal', exact: true }).click()
|
||||
await expect(chat).toHaveCount(0)
|
||||
await expect(
|
||||
orcaPage.getByText('Orca could not verify the originating Codex session file', { exact: false })
|
||||
).toHaveCount(0)
|
||||
expect(
|
||||
await orcaPage.evaluate(
|
||||
(key) => window.__store?.getState().agentStatusByPaneKey[key]?.providerSession?.id,
|
||||
descriptor.paneKey
|
||||
)
|
||||
).toBe(SESSION)
|
||||
await orcaPage.screenshot({ path: testInfo.outputPath('03-return-to-terminal.png') })
|
||||
})
|
||||
Reference in New Issue
Block a user