Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 26 additions & 17 deletions docs/reference/agent-status-store.md
Original file line number Diff line number Diff line change
Expand Up @@ -246,23 +246,32 @@ reader does not mistake them for drift:
- The Claude hook lane knows a background shell (or monitor, or workflow) by
Claude's task id, in one per-pane record (`claude-non-agent-work.ts`) on the
host that runs the session. A task enters it from the main agent's launching
`PostToolUse` (`tool_response.backgroundTaskId`, with that call's
`tool_use_id`) or from the `background_tasks` inventory Claude attaches to
`Stop`, which replaces the record. It leaves on the next inventory that omits
it, on a `TaskStop` result naming it, or on the `queue-operation` `enqueue`
line Claude writes to its transcript the moment the task ends (finished,
killed with its turn's Ctrl+C, or from `/tasks` while Claude idles), read by
the transcript watch below and trusted only when both its task id and its
tool-use id match the launch Orca recorded, since a prompt typed while Claude
is busy writes the same kind of line. A task can end before Orca handles its
launching hook, so its line can precede where the watch reads from; accepting
the launch reads back a bounded tail once for that exact pair. A `startup` or
`resume` `SessionStart` clears the record with the pane's children and crons,
since a new process can never end its predecessor's tasks. `/clear` keeps it:
the shell survives and reports its end in the new session's transcript, and
the pane stays on the tail of the last turn that ended, keeping that turn's
verdict and turn stamp so the shell's end is not announced as a new turn (not
yet after an Esc, which Orca does not infer, so that turn never ended). The
`PostToolUse`, with that call's `tool_use_id`, or from the
`background_tasks` inventory Claude attaches to `Stop`, which replaces the
record. The launch reader takes a shell's `tool_response.backgroundTaskId`;
otherwise a `status: "async_launched"` result's `taskId` and `taskType`,
captured only for workflows (`local_workflow`). It skips kinds the shared
kind table calls an agent, since agents belong to the subagent roster, and
admits unknown kinds. It skips any other result, such as a remote
`remote_launched` launch, which the next inventory still records. A task
leaves on the next inventory that omits it, on a `TaskStop` result naming
it, or on the `queue-operation` `enqueue` line Claude writes to its
transcript the moment the task ends, read by the transcript watch below and
trusted only when both its task id and its tool-use id match the launch Orca
recorded, since a prompt typed while Claude is busy writes the same kind of
line. Claude writes that line for a shell however it ended (finished, killed
with its turn's Ctrl+C, or from `/tasks` while Claude idles) and for a
workflow that completes, but not for a workflow stopped or paused from
`/tasks`, which leaves only on the next inventory. A task can end before
Orca handles its launching hook, so its line can precede where the watch
reads from; accepting the launch reads back a bounded tail once for that
exact pair. A `startup` or `resume` `SessionStart` clears the record with the
pane's children and crons, since a new process can never end its
predecessor's tasks. `/clear` keeps it: the task survives and reports its end
in the new session's transcript, and the pane stays on the tail of the last
turn that ended, keeping that turn's verdict and turn stamp so the task's end
is not announced as a new turn (not yet after an Esc, which Orca does not
infer, so that turn never ended). The
pane's process ending, its PTY exiting and the relay stopping clear it with
the rest of the pane. The desktop keeps no copy for a relayed pane; it reads
the row's `claudeRunningNonAgentTask`, which the relay restates on every row.
Expand Down
132 changes: 132 additions & 0 deletions src/main/agent-hooks/claude-background-task-capture.test-fixture.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,132 @@
// Replays the captured Claude background-task stories (src/shared/__fixtures__/
// claude-background-{shell,workflow}-*-hooks.jsonl) through the server's own HTTP ingress, with each
// hook's `transcript_path` pointed at a temp file. The only record of a task ending with no hook is
// the `queue-operation` line Claude appends to its transcript; a story appends the captured line
// where it was written and lets the listener's transcript watch find it on a real tick.
// Callers mock telemetry and fake only `Date`, and call `cleanUpCaptureReplays` after each test.
import { appendFileSync, mkdtempSync, rmSync, statSync, writeFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { expect, vi } from 'vitest'
import { AgentHookServer } from './server'
import { buildBody, PANE, postHookEvent } from './server.test-fixtures'
import {
hookAt,
type CapturedRecord,
type CapturedTranscriptScan
} from './claude-cancel-capture.test-fixture'

const temporaryPaths: string[] = []
const running: AgentHookServer[] = []

export function cleanUpCaptureReplays(): void {
for (const server of running.splice(0)) {
server.stop()
}
for (const path of temporaryPaths.splice(0)) {
rmSync(path, { recursive: true, force: true })
}
}

export function temporaryDir(): string {
const dir = mkdtempSync(join(tmpdir(), 'orca-background-task-'))
temporaryPaths.push(dir)
return dir
}

export function transcriptFile(): string {
const path = join(temporaryDir(), 'session.jsonl')
writeFileSync(path, '')
return path
}

export async function startServer(): Promise<AgentHookServer> {
const server = new AgentHookServer()
running.push(server)
await server.start({ env: 'production' })
return server
}

export function row(server: AgentHookServer) {
const entry = server.getStatusSnapshotForPane(PANE)[0]
if (!entry) {
throw new Error('the pane has no row')
}
return entry
}

/** The background-task fact the stored row carries (`claudeRunningNonAgentTask`). */
export function storedTaskFact(server: AgentHookServer): boolean | undefined {
return server._getStateForTests().lastStatusByPaneKey.get(PANE)?.claudeRunningNonAgentTask
}

export function watch(server: AgentHookServer) {
return server._getStateForTests().claudeTranscriptCursorByPaneKey.get(PANE)
}

/** The renderer's part: capture the row as the baseline and, once the settle window passes with
* no hook, ask the server to infer from the Ctrl+C. */
export function pressCtrlC(server: AgentHookServer): boolean {
const baseline = row(server)
return server.inferInterrupt({
paneKey: PANE,
baselineUpdatedAt: baseline.receivedAt,
baselineStateStartedAt: baseline.stateStartedAt,
baselinePrompt: baseline.prompt,
baselineAgentType: 'claude',
intent: 'ctrl-c'
})
}

export function queueLine(records: CapturedRecord[], label: string): CapturedTranscriptScan {
const scan = records.find((record) => record.kind === 'transcript' && record.label === label)
if (scan?.kind !== 'transcript' || scan.lines.length !== 1) {
throw new Error(`Captured transcript line ${label} not found`)
}
return scan
}

/** The capture's t=0 on the wall clock, from the stamp Claude wrote on its enqueue line. */
export function captureEpoch(records: CapturedRecord[]): number {
const scan = queueLine(records, 'queue-operation-enqueue')
// JSON.parse returns any; the timestamp is checked by Date.parse.
const parsed: Record<string, unknown> = JSON.parse(scan.lines[0])
return Date.parse(String(parsed.timestamp)) - scan.t * 1000
}

/** Replays hooks on the capture's clock, pointing each session's transcript at a temp file. A
* capture with no enqueue line passes its own `t0`. */
export function replayer(
server: AgentHookServer,
records: CapturedRecord[],
files: Map<string, string>,
t0 = captureEpoch(records)
) {
return async (indices: number[]) => {
for (const index of indices) {
const hook = hookAt(records, index)
vi.setSystemTime(t0 + hook.t * 1000)
const reported = String(hook.payload.transcript_path)
const transcript = files.get(reported) ?? files.get('*')
await expect(
postHookEvent(server, buildBody({ ...hook.payload, transcript_path: transcript }))
).resolves.toMatchObject({ status: 204 })
}
}
}

/** Appends a captured transcript line at its captured instant. */
export function writeCaptured(
records: CapturedRecord[],
scan: CapturedTranscriptScan,
path: string
): void {
vi.setSystemTime(captureEpoch(records) + scan.t * 1000)
appendFileSync(path, `${scan.lines[0]}\n`)
}

/** Waits for a tick to have read everything the transcript holds. */
export async function watchCaughtUp(server: AgentHookServer, transcript: string): Promise<void> {
const size = statSync(transcript).size
await vi.waitFor(() => expect(watch(server)?.offset).toBe(size), { timeout: 3_000 })
}
7 changes: 6 additions & 1 deletion src/main/agent-hooks/ended-process-reconciliation.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,12 @@ describe('reconcileEndedProcessForPaneKeys', () => {
const server = await startServer()
try {
claudeRow(server, 'done')
recordClaudeNonAgentTaskLaunch(server._getStateForTests(), PANE, 'btest0001', 'toolu_test')
recordClaudeNonAgentTaskLaunch(
server._getStateForTests(),
PANE,
{ taskId: 'btest0001', kind: 'unknown' },
'toolu_test'
)
server._getStateForTests().claudeActiveSessionCronPaneKeys.add(PANE)

expect(server.reconcileEndedProcessForPaneKeys([PANE])).toBe(1)
Expand Down
139 changes: 23 additions & 116 deletions src/main/agent-hooks/server-claude-background-shell-captures.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,19 +5,27 @@
// record of a task ending with no hook (a Ctrl+C'd turn's shell, a /tasks kill) is the
// `queue-operation` line Claude appends to its transcript; the stories append the captured line
// where it was written and let the listener's transcript watch find it on a real tick.
import { appendFileSync, mkdtempSync, rmSync, statSync, writeFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { appendFileSync, writeFileSync } from 'node:fs'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { AgentHookServer, _internals } from './server'
import { _internals } from './server'
import { buildBody, PANE, postHookEvent } from './server.test-fixtures'
import { cancelLabelled, hookAt, loadCapture } from './claude-cancel-capture.test-fixture'
import {
cancelLabelled,
hookAt,
loadCapture,
type CapturedRecord,
type CapturedTranscriptScan
} from './claude-cancel-capture.test-fixture'
captureEpoch,
cleanUpCaptureReplays,
pressCtrlC,
queueLine,
replayer,
row,
startServer,
storedTaskFact,
temporaryDir,
transcriptFile,
watch,
watchCaughtUp,
writeCaptured
} from './claude-background-task-capture.test-fixture'

const { getCohortAtEmitMock, trackMock } = vi.hoisted(() => ({
getCohortAtEmitMock: vi.fn(),
Expand All @@ -27,9 +35,6 @@ const { getCohortAtEmitMock, trackMock } = vi.hoisted(() => ({
vi.mock('../telemetry/client', () => ({ track: trackMock }))
vi.mock('../telemetry/cohort-classifier', () => ({ getCohortAtEmit: getCohortAtEmitMock }))

const temporaryPaths: string[] = []
const running: AgentHookServer[] = []

beforeEach(() => {
_internals.resetCachesForTests()
trackMock.mockReset()
Expand All @@ -40,109 +45,11 @@ beforeEach(() => {
})

afterEach(() => {
for (const server of running.splice(0)) {
server.stop()
}
for (const path of temporaryPaths.splice(0)) {
rmSync(path, { recursive: true, force: true })
}
cleanUpCaptureReplays()
vi.useRealTimers()
vi.restoreAllMocks()
})

function temporaryDir(): string {
const dir = mkdtempSync(join(tmpdir(), 'orca-background-shell-'))
temporaryPaths.push(dir)
return dir
}

function transcriptFile(): string {
const path = join(temporaryDir(), 'session.jsonl')
writeFileSync(path, '')
return path
}

async function startServer(): Promise<AgentHookServer> {
const server = new AgentHookServer()
running.push(server)
await server.start({ env: 'production' })
return server
}

function row(server: AgentHookServer) {
const entry = server.getStatusSnapshotForPane(PANE)[0]
if (!entry) {
throw new Error('the pane has no row')
}
return entry
}

function storedShellFact(server: AgentHookServer): boolean | undefined {
return server._getStateForTests().lastStatusByPaneKey.get(PANE)?.claudeRunningNonAgentTask
}

function watch(server: AgentHookServer) {
return server._getStateForTests().claudeTranscriptCursorByPaneKey.get(PANE)
}

/** The renderer's part: capture the row as the baseline and, once the settle window passes with
* no hook, ask the server to infer from the Ctrl+C. */
function pressCtrlC(server: AgentHookServer): boolean {
const baseline = row(server)
return server.inferInterrupt({
paneKey: PANE,
baselineUpdatedAt: baseline.receivedAt,
baselineStateStartedAt: baseline.stateStartedAt,
baselinePrompt: baseline.prompt,
baselineAgentType: 'claude',
intent: 'ctrl-c'
})
}

function queueLine(records: CapturedRecord[], label: string): CapturedTranscriptScan {
const scan = records.find((record) => record.kind === 'transcript' && record.label === label)
if (scan?.kind !== 'transcript' || scan.lines.length !== 1) {
throw new Error(`Captured transcript line ${label} not found`)
}
return scan
}

/** The capture's t=0 on the wall clock, from the stamp Claude wrote on its enqueue line. */
function captureEpoch(records: CapturedRecord[]): number {
const scan = queueLine(records, 'queue-operation-enqueue')
// JSON.parse returns any; the timestamp is checked by Date.parse.
const parsed: Record<string, unknown> = JSON.parse(scan.lines[0])
return Date.parse(String(parsed.timestamp)) - scan.t * 1000
}

/** Replays hooks on the capture's clock, pointing each session's transcript at a temp file. */
function replayer(server: AgentHookServer, records: CapturedRecord[], files: Map<string, string>) {
const t0 = captureEpoch(records)
return async (indices: number[]) => {
for (const index of indices) {
const hook = hookAt(records, index)
vi.setSystemTime(t0 + hook.t * 1000)
const reported = String(hook.payload.transcript_path)
const transcript = files.get(reported) ?? files.get('*')
await expect(
postHookEvent(server, buildBody({ ...hook.payload, transcript_path: transcript }))
).resolves.toMatchObject({ status: 204 })
}
}
}

/** Appends a captured transcript line at its captured instant. */
function writeCaptured(records: CapturedRecord[], scan: CapturedTranscriptScan, path: string) {
vi.setSystemTime(captureEpoch(records) + scan.t * 1000)
appendFileSync(path, `${scan.lines[0]}\n`)
}

/** Waits for a tick to have read everything the transcript holds. */
async function watchCaughtUp(server: AgentHookServer, transcript: string): Promise<void> {
const size = statSync(transcript).size
await vi.waitFor(() => expect(watch(server)?.offset).toBe(size), { timeout: 3_000 })
}

describe('Ctrl+C in the turn that launched a background shell (captured, 2.1.284)', () => {
const records = loadCapture('claude-background-shell-ctrl-c-hooks')
const cancel = cancelLabelled(records, 'S1-ctrl-c-mid-essay')
Expand Down Expand Up @@ -170,7 +77,7 @@ describe('Ctrl+C in the turn that launched a background shell (captured, 2.1.284
mainAgent: { state: 'done', outcome: 'cancellation' }
})
expect(row(server).interrupted).toBeUndefined()
expect(storedShellFact(server)).toBe(true)
expect(storedTaskFact(server)).toBe(true)

// The next typed turn's Stop lists the same shell running.
await replay([7, 9, 8])
Expand All @@ -187,7 +94,7 @@ describe('Ctrl+C in the turn that launched a background shell (captured, 2.1.284
writeCaptured(records, killedAtExit, transcript)
await vi.waitFor(() => expect(row(server).state).toBe('done'), { timeout: 3_000 })
expect(row(server).workingMode).toBeUndefined()
expect(storedShellFact(server)).toBe(false)
expect(storedTaskFact(server)).toBe(false)
})

it('retires the shell only on a line naming both its task id and its launching call', async () => {
Expand Down Expand Up @@ -338,10 +245,10 @@ describe('a /tasks kill during a running turn (captured, 2.1.285)', () => {
const replay = replayer(server, records, new Map([['*', transcript]]))
await replay([0, 1, 2, 3, 4, 5, 6, 7, 8])
expect(row(server)).toMatchObject({ state: 'working', mainAgent: { state: 'working' } })
expect(storedShellFact(server)).toBe(true)
expect(storedTaskFact(server)).toBe(true)

writeCaptured(records, queueLine(records, 'queue-operation-enqueue'), transcript)
await vi.waitFor(() => expect(storedShellFact(server)).toBe(false), { timeout: 3_000 })
await vi.waitFor(() => expect(storedTaskFact(server)).toBe(false), { timeout: 3_000 })
expect(row(server)).toMatchObject({ state: 'working', mainAgent: { state: 'working' } })

// 40 s later the turn's next tool boundary absorbs the notification, and its Stop lists nothing.
Expand Down Expand Up @@ -419,7 +326,7 @@ describe('/clear while a background shell runs (captured, 2.1.285)', () => {
await replay([9])
expect(row(server)).toMatchObject({ state: 'working', mainAgent: { state: 'working' } })
appendFileSync(cleared, `${completed.lines[0]}\n`)
await vi.waitFor(() => expect(storedShellFact(server)).toBe(false), { timeout: 3_000 })
await vi.waitFor(() => expect(storedTaskFact(server)).toBe(false), { timeout: 3_000 })
expect(row(server)).toMatchObject({ state: 'working', mainAgent: { state: 'working' } })

await replay([10])
Expand Down
Loading
Loading