mirror of
https://github.com/stablyai/orca.git
synced 2026-10-31 00:57:45 +00:00
Stop native chat jumping when Codex starts a foreground command (#26503)
* Keep foreground Codex commands out of the background strip * Run the command-strip host regression in Node * Exercise retained commands after their source turns end
This commit is contained in:
@@ -203,6 +203,7 @@ export const SQLITE_RUNTIME_INCLUDE = [
|
||||
'src/main/runtime/structured-agent-session-startup-reconcile.test.ts',
|
||||
'src/main/runtime/structured-claude-attachment-grant.test.ts',
|
||||
'src/main/runtime/structured-claude-pending-rewind.test.ts',
|
||||
'src/main/runtime/structured-foreground-command-strip.test.ts',
|
||||
'src/main/ssh-expired-lease-pane-readoption.test.ts',
|
||||
'src/main/ssh-reattach-pane-cardinality.test.ts',
|
||||
'src/main/ssh/orcad-migration-cutover-coordinator.test.ts',
|
||||
|
||||
@@ -35,6 +35,23 @@ function command(method: string, id = 'exec', threadId = 'root') {
|
||||
}
|
||||
|
||||
describe('persistent command ownership', () => {
|
||||
it('promotes only the launching turn and recognizes a start delivered after its end', () => {
|
||||
const tracker = new CodexBackgroundCommandTracker('root')
|
||||
tracker.observe(command('item/started'))
|
||||
expect(tracker.tasks()).toEqual([])
|
||||
expect(tracker.endTurn('root', 'another-turn')).toEqual([])
|
||||
expect(tracker.endTurn('child', 'turn')).toEqual([])
|
||||
expect(tracker.tasks()).toEqual([])
|
||||
expect(tracker.endTurn('root', 'turn')).toHaveLength(1)
|
||||
expect(tracker.tasks()).toHaveLength(1)
|
||||
expect(tracker.endTurn('root', 'turn')).toEqual([])
|
||||
tracker.observe(command('item/started', 'late'))
|
||||
expect(tracker.tasks()).toHaveLength(2)
|
||||
tracker.clear()
|
||||
tracker.observe(command('item/started', 'new'))
|
||||
expect(tracker.tasks()).toEqual([])
|
||||
})
|
||||
|
||||
it('preflights finite metadata capacity and admits work again after process completion', () => {
|
||||
const tracker = new CodexBackgroundCommandTracker('root', 700)
|
||||
const first = command('item/started', 'first')
|
||||
@@ -43,12 +60,12 @@ describe('persistent command ownership', () => {
|
||||
tracker.observe(first)
|
||||
expect(tracker.canObserve(second)).toBe(false)
|
||||
expect(() => tracker.observe(second)).toThrow('not admitted')
|
||||
expect(tracker.tasks()).toHaveLength(1)
|
||||
expect(tracker.threadCommands('root')).toHaveLength(1)
|
||||
expect(tracker.retainedMetadataBytes).toBeLessThanOrEqual(700)
|
||||
tracker.observe(command('item/completed', 'first'))
|
||||
expect(tracker.canObserve(second)).toBe(true)
|
||||
tracker.observe(second)
|
||||
expect(tracker.tasks()).toHaveLength(1)
|
||||
expect(tracker.threadCommands('root')).toHaveLength(1)
|
||||
expect(tracker.retainedMetadataBytes).toBeLessThanOrEqual(700)
|
||||
tracker.clear()
|
||||
expect(tracker.retainedMetadataBytes).toBe(0)
|
||||
@@ -73,10 +90,11 @@ describe('persistent command ownership', () => {
|
||||
deliver(command('item/started'))
|
||||
const originalKey = rows.find(({ body }) => body.kind === 'tool-call')?.key
|
||||
deliver(notification('turn/completed', { turn: { id: 'turn' } }))
|
||||
tracker.endTurn('root', 'turn')
|
||||
expect(rows.filter(({ body }) => body.kind === 'tool-call').map(({ body }) => body)).toEqual([
|
||||
expect.objectContaining({ state: 'running' })
|
||||
])
|
||||
expect(tracker.tasks()).toHaveLength(1)
|
||||
expect(tracker.threadCommands('root')).toHaveLength(1)
|
||||
deliver(
|
||||
notification('item/commandExecution/outputDelta', { itemId: 'exec', delta: 'late output' })
|
||||
)
|
||||
@@ -102,6 +120,7 @@ describe('persistent command ownership', () => {
|
||||
})
|
||||
)
|
||||
expect(tracker.tasks(new Set(['child']))).toEqual([])
|
||||
tracker.endTurn('child', 'turn')
|
||||
expect(tracker.tasks()).toEqual([
|
||||
{ id: 'codex-command:thread:child:child-exec', kind: 'command', description: 'sleep 30' }
|
||||
])
|
||||
@@ -119,6 +138,7 @@ describe('persistent command ownership', () => {
|
||||
tracker.observe(command('item/started', `short-${index}`))
|
||||
tracker.observe(command('item/completed', `short-${index}`))
|
||||
}
|
||||
tracker.endTurn('root', 'turn')
|
||||
expect(tracker.tasks()).toEqual([
|
||||
{ id: 'codex-command:primary:long-lived', kind: 'command', description: 'sleep 30' }
|
||||
])
|
||||
|
||||
@@ -8,14 +8,28 @@ import { readRecord, readString } from './codex-item-field-readers'
|
||||
import { readCodexThreadItem } from './codex-structured-item-translation'
|
||||
import { MAX_CODEX_ITEM_STREAM_METADATA_BYTES } from './codex-item-stream-retention'
|
||||
import type { CodexAbandonedCommand } from './codex-prompt-registry'
|
||||
import { readCodexTurnId } from './codex-structured-thread-facts'
|
||||
import { codexPrimaryCommandTaskId } from '../../shared/structured-session-foreground-commands'
|
||||
|
||||
const MAX_SETTLED_COMMANDS = 128
|
||||
|
||||
type Command = { threadId: string; task: AgentSessionBackgroundTask; bytes: number }
|
||||
type Command = {
|
||||
threadId: string
|
||||
turnId: string | null
|
||||
backgrounded: boolean
|
||||
task: AgentSessionBackgroundTask
|
||||
bytes: number
|
||||
}
|
||||
|
||||
/** A command process starting, or ending: it exited, its thread closed, or the session ended. */
|
||||
export type CodexBackgroundCommandChange =
|
||||
| { type: 'started'; threadId: string; task: AgentSessionBackgroundTask }
|
||||
| {
|
||||
type: 'started'
|
||||
threadId: string
|
||||
turnId: string | null
|
||||
backgrounded: boolean
|
||||
task: AgentSessionBackgroundTask
|
||||
}
|
||||
| { type: 'ended'; threadId: string; taskId: string }
|
||||
|
||||
export class CodexBackgroundCommandTracker {
|
||||
@@ -62,7 +76,29 @@ export class CodexBackgroundCommandTracker {
|
||||
this.commands.set(key, command)
|
||||
this.liveBytes += command.bytes
|
||||
this.trimSettled()
|
||||
return { type: 'started', threadId: command.threadId, task: command.task }
|
||||
return this.liveChange(command)
|
||||
}
|
||||
|
||||
/** A surviving process becomes background work when the turn that launched it ends. */
|
||||
endTurn(threadId: string, turnId: string): CodexBackgroundCommandChange[] {
|
||||
const backgrounded: CodexBackgroundCommandChange[] = []
|
||||
for (const command of this.commands.values()) {
|
||||
if (
|
||||
command.threadId === threadId &&
|
||||
(command.turnId === turnId || command.turnId === null) &&
|
||||
!command.backgrounded
|
||||
) {
|
||||
command.backgrounded = true
|
||||
backgrounded.push(this.liveChange(command))
|
||||
}
|
||||
}
|
||||
this.rememberSettled(this.turnKey(threadId, turnId))
|
||||
return backgrounded
|
||||
}
|
||||
|
||||
private liveChange(command: Command): Extract<CodexBackgroundCommandChange, { type: 'started' }> {
|
||||
const { threadId, turnId, backgrounded, task } = command
|
||||
return { type: 'started', threadId, turnId, backgrounded, task }
|
||||
}
|
||||
|
||||
/** The thread closed: Codex stops its processes first, so none of them can report an exit. */
|
||||
@@ -82,7 +118,7 @@ export class CodexBackgroundCommandTracker {
|
||||
childLabel?: (threadId: string) => string | null
|
||||
): AgentSessionBackgroundTask[] {
|
||||
return [...this.commands.values()]
|
||||
.filter((command) => !coveredThreads?.has(command.threadId))
|
||||
.filter((command) => command.backgrounded && !coveredThreads?.has(command.threadId))
|
||||
.map(({ threadId, task }) => {
|
||||
// The agent row carrying the child's name is gone by the time this row shows;
|
||||
// unqualified it reads as a bare shell string with no owner. Resolved on read so
|
||||
@@ -94,11 +130,11 @@ export class CodexBackgroundCommandTracker {
|
||||
})
|
||||
}
|
||||
|
||||
/** The live commands one thread launched, as the strip would publish them. */
|
||||
threadTasks(threadId: string): AgentSessionBackgroundTask[] {
|
||||
/** Live processes, including foreground tools that are absent from the strip. */
|
||||
threadCommands(threadId: string): CodexBackgroundCommandChange[] {
|
||||
return [...this.commands.values()]
|
||||
.filter((command) => command.threadId === threadId)
|
||||
.map((command) => command.task)
|
||||
.map((command) => this.liveChange(command))
|
||||
}
|
||||
|
||||
/** The session ended, and every command with it. */
|
||||
@@ -124,15 +160,26 @@ export class CodexBackgroundCommandTracker {
|
||||
this.liveBytes -= existing.bytes
|
||||
this.commands.delete(key)
|
||||
}
|
||||
this.rememberSettled(key)
|
||||
return existing
|
||||
? { type: 'ended', threadId: existing.threadId, taskId: existing.task.id }
|
||||
: null
|
||||
}
|
||||
|
||||
private turnKey(threadId: string, turnId: string): string {
|
||||
return JSON.stringify(['turn', threadId, turnId])
|
||||
}
|
||||
|
||||
private rememberSettled(key: string): void {
|
||||
if (this.settled.has(key)) {
|
||||
return
|
||||
}
|
||||
const bytes = Buffer.byteLength(key, 'utf8') + 256
|
||||
if (this.liveBytes + bytes <= this.maxMetadataBytes) {
|
||||
this.settled.set(key, bytes)
|
||||
this.settledBytes += bytes
|
||||
}
|
||||
this.trimSettled()
|
||||
return existing
|
||||
? { type: 'ended', threadId: existing.threadId, taskId: existing.task.id }
|
||||
: null
|
||||
}
|
||||
|
||||
private trimSettled(): void {
|
||||
@@ -155,8 +202,7 @@ export class CodexBackgroundCommandTracker {
|
||||
if (event.method !== 'item/started' && event.method !== 'item/completed') {
|
||||
return null
|
||||
}
|
||||
// Any command may outlive its turn; `source` says only how Codex launched it. A stdin write
|
||||
// starts no process: it reaches one already tracked.
|
||||
// Track every process; a stdin write reaches one already tracked.
|
||||
const item = readCodexThreadItem(readRecord(event.params).item)
|
||||
if (item?.type !== 'commandExecution' || item.source === 'unifiedExecInteraction') {
|
||||
return null
|
||||
@@ -166,12 +212,15 @@ export class CodexBackgroundCommandTracker {
|
||||
const description = boundCodexCommandDescription(readString(item, 'command') ?? '')
|
||||
.replace(/\s+/g, ' ')
|
||||
.trim()
|
||||
const turnId = readCodexTurnId(event.params)
|
||||
const value = {
|
||||
threadId: event.threadId,
|
||||
turnId,
|
||||
backgrounded: turnId !== null && this.settled.has(this.turnKey(event.threadId, turnId)),
|
||||
task: {
|
||||
id:
|
||||
event.threadId === this.primaryThreadId
|
||||
? `codex-command:primary:${encodeURIComponent(item.id)}`
|
||||
? codexPrimaryCommandTaskId(item.id)
|
||||
: `codex-command:thread:${encodeURIComponent(event.threadId)}:${encodeURIComponent(item.id)}`,
|
||||
kind: 'command' as const,
|
||||
...(description ? { description } : {})
|
||||
|
||||
@@ -60,7 +60,7 @@ function command(
|
||||
threadId,
|
||||
params: {
|
||||
threadId,
|
||||
turnId: PARENT_TURN,
|
||||
turnId: threadId === PRIMARY ? PARENT_TURN : CHILD_TURN,
|
||||
item: {
|
||||
type: 'commandExecution',
|
||||
id: 'exec-1',
|
||||
@@ -227,13 +227,12 @@ describe('CodexBackgroundTaskTracker child execution ownership', () => {
|
||||
})
|
||||
|
||||
describe('CodexBackgroundTaskTracker command integration', () => {
|
||||
it('keeps a primary shell visible from launch until its own completion', () => {
|
||||
it('reveals a surviving primary shell after its own turn ends', () => {
|
||||
const tracker = new CodexBackgroundTaskTracker(PRIMARY)
|
||||
const shell = [{ id: 'codex-command:primary:exec-1', kind: 'command', description: 'sleep 90' }]
|
||||
tracker.observe(turn('turn/started', PRIMARY, PARENT_TURN))
|
||||
tracker.observe(command())
|
||||
// Visible while the turn that launched it is still running.
|
||||
expect(tracker.state?.tasks).toEqual(shell)
|
||||
expect(tracker.state).toBeNull()
|
||||
tracker.observe(turn('turn/completed', PRIMARY, PARENT_TURN))
|
||||
expect(tracker.state?.tasks).toEqual(shell)
|
||||
// Only the shell's own completion retires the row.
|
||||
|
||||
@@ -44,7 +44,7 @@ export class CodexBackgroundTaskTracker {
|
||||
) {
|
||||
this.commands = new CodexBackgroundCommandTracker(primaryThreadId)
|
||||
this.childWork = new CodexChildWorkEvidence(primaryThreadId, executions, (threadId) =>
|
||||
this.commands.threadTasks(threadId)
|
||||
this.commands.threadCommands(threadId)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -64,15 +64,18 @@ export class CodexBackgroundTaskTracker {
|
||||
): boolean {
|
||||
const itemEvent = event.method === 'item/started' || event.method === 'item/completed'
|
||||
const command = itemEvent ? this.commands.observe(event) : null
|
||||
const frame = readCodexBackgroundTaskFrame(event, this.primaryThreadId)
|
||||
const commands = [
|
||||
...unapproved.flatMap((abandoned) => this.commands.endUnapproved(abandoned) ?? []),
|
||||
...(frame?.kind === 'turn' && frame.state !== 'working'
|
||||
? this.commands.endTurn(frame.threadId, frame.turnId)
|
||||
: []),
|
||||
...(event.method === 'thread/closed'
|
||||
? this.commands.endThread(event.threadId)
|
||||
: command
|
||||
? [command]
|
||||
: [])
|
||||
]
|
||||
const frame = readCodexBackgroundTaskFrame(event, this.primaryThreadId)
|
||||
if (frame?.kind === 'subagents') {
|
||||
for (const child of frame.children) {
|
||||
this.executions.register(
|
||||
|
||||
@@ -411,14 +411,20 @@ describe('Codex child-work evidence', () => {
|
||||
expect.objectContaining({
|
||||
membership: 'live',
|
||||
description: 'npm run dev',
|
||||
residency: 'background',
|
||||
residency: 'foreground',
|
||||
parentChildWorkId: agent!.childWorkId,
|
||||
firstObservedAt: 1_040
|
||||
})
|
||||
])
|
||||
send(turn('turn/completed', CHILD, 'c1'))
|
||||
expect(byKind('agent')[0]).toMatchObject({ membership: 'settled', outcome: 'succeeded' })
|
||||
expect(byKind('command')).toEqual([expect.objectContaining({ membership: 'live' })])
|
||||
expect(byKind('command')).toEqual([
|
||||
expect.objectContaining({
|
||||
membership: 'live',
|
||||
residency: 'background',
|
||||
firstObservedAt: 1_040
|
||||
})
|
||||
])
|
||||
expect(display(agent!.childWorkId)).toBe('monitoring')
|
||||
send(
|
||||
item('item/completed', CHILD, 'c1', {
|
||||
@@ -553,7 +559,7 @@ describe('Codex child-work evidence', () => {
|
||||
expect(records()).toEqual([
|
||||
expect.objectContaining({ kind: 'agent', membership: 'settled', outcome: 'succeeded' })
|
||||
])
|
||||
// One edge when each shell starts and one when it exits, as the strip republishes today.
|
||||
// Raw status evidence keeps every start and exit, including foreground processes.
|
||||
expect(
|
||||
log
|
||||
.slice(before)
|
||||
|
||||
@@ -12,7 +12,6 @@
|
||||
// Codex children outlive the turn that spawned them, so only a child's own turn, or the session,
|
||||
// ends it.
|
||||
|
||||
import type { AgentSessionBackgroundTask } from '../../shared/agent-session-wire'
|
||||
import type {
|
||||
AgentChildWorkEvidence,
|
||||
AgentChildWorkLiveObservation
|
||||
@@ -65,15 +64,18 @@ function ofTurn<T extends { turnId: string | null }>(fact: T | undefined, turnId
|
||||
return fact?.turnId === turnId ? fact : undefined
|
||||
}
|
||||
|
||||
function commandLive(task: AgentSessionBackgroundTask, ownerId: string | null) {
|
||||
function commandLive(
|
||||
command: Extract<CodexBackgroundCommandChange, { type: 'started' }>,
|
||||
ownerId: string | null
|
||||
) {
|
||||
const { task } = command
|
||||
return (observedAt: number): AgentChildWorkEvidence => ({
|
||||
type: 'live',
|
||||
observedAt,
|
||||
child: {
|
||||
handle: { idKind: 'task_id', id: task.id },
|
||||
kind: 'command',
|
||||
// Its own process, not a turn's: no turn ending may settle it.
|
||||
residency: 'background',
|
||||
residency: command.backgrounded ? 'background' : 'foreground',
|
||||
state: 'working',
|
||||
...(task.description ? { description: task.description } : {}),
|
||||
...(ownerId !== null ? { ownerId } : {}),
|
||||
@@ -89,7 +91,7 @@ export class CodexChildWorkEvidence {
|
||||
constructor(
|
||||
private readonly primaryThreadId: string,
|
||||
private readonly executions: CodexSubagentExecutions,
|
||||
private readonly liveCommands: (threadId: string) => readonly AgentSessionBackgroundTask[]
|
||||
private readonly liveCommands: (threadId: string) => readonly CodexBackgroundCommandChange[]
|
||||
) {}
|
||||
|
||||
/** After the tracker applied the frame: which command processes it saw start or stop, and the
|
||||
@@ -142,7 +144,7 @@ export class CodexChildWorkEvidence {
|
||||
for (const command of commands) {
|
||||
if (command.type === 'started') {
|
||||
const ownerId = command.threadId === this.primaryThreadId ? null : command.threadId
|
||||
this.pending.push(commandLive(command.task, ownerId))
|
||||
this.pending.push(commandLive(command, ownerId))
|
||||
continue
|
||||
}
|
||||
const { taskId } = command
|
||||
@@ -277,9 +279,7 @@ export class CodexChildWorkEvidence {
|
||||
/** Work a child launched before the host held its record was admitted with no owner; now that
|
||||
* the owner is recorded, say again whose it is. */
|
||||
private requeueOwnedBy(threadId: string): void {
|
||||
for (const task of this.liveCommands(threadId)) {
|
||||
this.pending.push(commandLive(task, threadId))
|
||||
}
|
||||
this.queueCommands(this.liveCommands(threadId))
|
||||
for (const spawned of this.executions.workingChildren()) {
|
||||
const facts = this.facts.get(spawned.agentThreadId)
|
||||
if (spawned.spawnerThreadId === threadId && facts?.published !== undefined) {
|
||||
|
||||
@@ -503,7 +503,10 @@ describe('a helper whose spawn was never seen', () => {
|
||||
turn('turn/started', HELPER, HELPER_TURN),
|
||||
helperShell('item/started')
|
||||
)
|
||||
expect(run.strip()).toEqual([expect.objectContaining({ kind: 'command' })])
|
||||
expect(run.strip()).toEqual([])
|
||||
expect(run.commands()).toEqual([
|
||||
expect.objectContaining({ membership: 'live', residency: 'foreground' })
|
||||
])
|
||||
|
||||
run.send(sendInput)
|
||||
// No spawn named it, so it reads as any unnamed subagent does.
|
||||
|
||||
@@ -110,11 +110,14 @@ describe('persistent command retention', () => {
|
||||
}).admission
|
||||
).toEqual({ accepted: true })
|
||||
}
|
||||
expect(tracker.tasks()).toEqual([])
|
||||
for (let thread = 0; thread < 7; thread += 1) {
|
||||
const threadId = `thread-${thread}`
|
||||
expect(tracker.threadCommands(threadId)).toHaveLength(64)
|
||||
expect(
|
||||
settleCodexJournalTurn({
|
||||
sessionId: 'session',
|
||||
threadId: `thread-${thread}`,
|
||||
threadId,
|
||||
turnId: 'turn',
|
||||
turnLifecycle: null,
|
||||
completedAt: 1,
|
||||
@@ -125,6 +128,8 @@ describe('persistent command retention', () => {
|
||||
attributionFor: () => ({ turnScope: AGENT_JOURNAL_THREAD_SCOPE })
|
||||
})
|
||||
).toEqual({ accepted: true })
|
||||
expect(tracker.endTurn(threadId, 'turn')).toHaveLength(64)
|
||||
expect(tracker.tasks()).toHaveLength((thread + 1) * 64)
|
||||
}
|
||||
expect(items.activeItems.size).toBe(448)
|
||||
expect(items.streams.persistentCount).toBe(448)
|
||||
|
||||
@@ -211,7 +211,7 @@ describe('the chat strip and the session list read the same host child records',
|
||||
source: 'unifiedExecStartup',
|
||||
status: 'inProgress'
|
||||
})
|
||||
// Every command the child runs is a live record under it, whatever Codex tagged it.
|
||||
// Foreground shell records keep status live; surfaces show the child's tool instead.
|
||||
both([
|
||||
{
|
||||
kind: 'agent',
|
||||
@@ -219,20 +219,6 @@ describe('the chat strip and the session list read the same host child records',
|
||||
state: 'working',
|
||||
membership: 'live',
|
||||
tool: 'Bash: npm run dev'
|
||||
},
|
||||
{
|
||||
kind: 'command',
|
||||
description: 'npm test',
|
||||
state: 'working',
|
||||
membership: 'live',
|
||||
owner: 'review'
|
||||
},
|
||||
{
|
||||
kind: 'command',
|
||||
description: 'npm run dev',
|
||||
state: 'working',
|
||||
membership: 'live',
|
||||
owner: 'review'
|
||||
}
|
||||
])
|
||||
|
||||
@@ -252,13 +238,6 @@ describe('the chat strip and the session list read the same host child records',
|
||||
state: 'working',
|
||||
membership: 'live',
|
||||
tool: 'Bash: npm run dev'
|
||||
},
|
||||
{
|
||||
kind: 'command',
|
||||
description: 'npm run dev',
|
||||
state: 'working',
|
||||
membership: 'live',
|
||||
owner: 'review'
|
||||
}
|
||||
])
|
||||
|
||||
|
||||
@@ -0,0 +1,147 @@
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { AgentSessionBackgroundTaskState } from '../../shared/agent-session-wire'
|
||||
import { AgentHookServer, _internals } from '../agent-hooks/server'
|
||||
import { fakeCodex, THREAD_ID } from '../codex/codex-structured-session-adapter-fixture'
|
||||
import {
|
||||
HOST_TEST_SESSION as SESSION,
|
||||
hostTestAttachParams
|
||||
} from '../native-chat/agent-session-wire/structured-agent-session-host-test-data'
|
||||
import { createStructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger'
|
||||
import {
|
||||
ensureStructuredAgentSessionHost,
|
||||
stopStructuredAgentSessionRuntime
|
||||
} from './structured-agent-session-runtime'
|
||||
|
||||
vi.mock('../telemetry/client', () => ({ track: vi.fn() }))
|
||||
vi.mock('../telemetry/cohort-classifier', () => ({ getCohortAtEmit: vi.fn() }))
|
||||
|
||||
describe('foreground commands in the background-task channel', () => {
|
||||
let root: string
|
||||
let server: AgentHookServer
|
||||
|
||||
beforeEach(async () => {
|
||||
_internals.resetCachesForTests()
|
||||
root = await mkdtemp(join(tmpdir(), 'orca-foreground-command-strip-'))
|
||||
server = new AgentHookServer()
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await stopStructuredAgentSessionRuntime()
|
||||
await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
async function rig() {
|
||||
const codex = fakeCodex({
|
||||
'model/list': () => ({
|
||||
data: [
|
||||
{
|
||||
model: 'gpt-test',
|
||||
displayName: 'GPT Test',
|
||||
hidden: false,
|
||||
supportedReasoningEfforts: [],
|
||||
defaultReasoningEffort: null,
|
||||
isDefault: true
|
||||
}
|
||||
],
|
||||
nextCursor: null
|
||||
})
|
||||
})
|
||||
const host = await ensureStructuredAgentSessionHost({
|
||||
logger: createStructuredAgentSessionLogger(),
|
||||
stateDirectory: root,
|
||||
hostId: 'local',
|
||||
claimKeyId: 'key-1',
|
||||
resolveWorkspacePath: async () => root,
|
||||
resolveLaunchArgs: () => [],
|
||||
resolveClaudeAuthPolicy: () => ({ stripAuthEnv: true }),
|
||||
resolveCodexCommand: () => 'codex',
|
||||
resolveEnvironment: async () => ({ PATH: process.env.PATH }),
|
||||
openCodexConnection: codex.openConnection,
|
||||
readProcessStartTime: async () => 1_700_000_000_000,
|
||||
statusSink: {
|
||||
publish: (summary, subject) => server.ingestStructuredStatus(summary, subject),
|
||||
forget: (subject) => server.dropStructuredStatus(subject),
|
||||
publishChildWork: (subject, evidence, provider) =>
|
||||
server.ingestStructuredChildWork(subject, evidence, provider),
|
||||
readChildWork: (subject) => server.getStructuredChildWorkViews(subject)
|
||||
}
|
||||
})
|
||||
const params = hostTestAttachParams(null, { providerHandle: undefined })
|
||||
params.envelope.clientOperationId = `${Date.now()}-${'1'.padStart(32, '0')}`
|
||||
const attached = await host.attach({ callerKey: 'strip-test' }, params)
|
||||
expect(attached, JSON.stringify(attached)).toMatchObject({ ok: true })
|
||||
const rosters: (AgentSessionBackgroundTaskState | null)[] = []
|
||||
await host.subscribe({
|
||||
id: 'strip',
|
||||
sessionId: SESSION,
|
||||
emit: (event) => {
|
||||
if ('backgroundTasks' in event && event.backgroundTasks !== undefined) {
|
||||
rosters.push(event.backgroundTasks)
|
||||
}
|
||||
}
|
||||
})
|
||||
const settle = async () => {
|
||||
for (let tick = 0; tick < 5; tick += 1) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 0))
|
||||
}
|
||||
}
|
||||
const notify = async (method: string, params: Record<string, unknown>) => {
|
||||
codex.connections[0]?.handlers.onNotification?.(method, params)
|
||||
await settle()
|
||||
}
|
||||
const turn = (method: string, id: string) =>
|
||||
notify(method, { threadId: THREAD_ID, turn: { id, status: 'completed' } })
|
||||
const command = (method: string, id: string, turnId = 'turn-1') =>
|
||||
notify(method, {
|
||||
threadId: THREAD_ID,
|
||||
turnId,
|
||||
item: {
|
||||
type: 'commandExecution',
|
||||
id,
|
||||
command: id === 'pwd' ? 'pwd' : 'git status --short --branch',
|
||||
source: 'unifiedExecStartup',
|
||||
status: method === 'item/completed' ? 'completed' : 'inProgress'
|
||||
}
|
||||
})
|
||||
return { host, rosters, turn, command }
|
||||
}
|
||||
|
||||
it('never publishes a strip between a quick command starting and completing', async () => {
|
||||
const { host, rosters, turn, command } = await rig()
|
||||
await turn('turn/started', 'turn-1')
|
||||
for (const id of ['pwd', 'status']) {
|
||||
await command('item/started', id)
|
||||
const page = await host.history({ sessionId: SESSION, direction: 'tail' })
|
||||
expect(page.page.items).toContainEqual(
|
||||
expect.objectContaining({
|
||||
body: expect.objectContaining({ kind: 'tool-call', callId: id })
|
||||
})
|
||||
)
|
||||
// Observe every publication, including the start frame before the completion arrives.
|
||||
expect(rosters.every((roster) => roster === null)).toBe(true)
|
||||
expect(page.page.backgroundTasks).toBeNull()
|
||||
await command('item/completed', id)
|
||||
}
|
||||
await turn('turn/completed', 'turn-1')
|
||||
expect(rosters.every((roster) => roster === null)).toBe(true)
|
||||
})
|
||||
|
||||
it('shows a surviving command after its own turn ends and throughout the next turn', async () => {
|
||||
const { rosters, turn, command } = await rig()
|
||||
await turn('turn/started', 'turn-1')
|
||||
await command('item/started', 'server')
|
||||
await turn('turn/completed', 'turn-1')
|
||||
expect(rosters.at(-1)?.tasks).toContainEqual(
|
||||
expect.objectContaining({ id: 'codex-command:primary:server', kind: 'command' })
|
||||
)
|
||||
await turn('turn/started', 'turn-2')
|
||||
await command('item/started', 'pwd', 'turn-2')
|
||||
expect(rosters.at(-1)?.tasks?.map((task) => task.id)).toEqual(['codex-command:primary:server'])
|
||||
await command('item/completed', 'pwd', 'turn-2')
|
||||
await command('item/completed', 'server')
|
||||
expect(rosters.at(-1)).toBeNull()
|
||||
})
|
||||
})
|
||||
+10
-4
@@ -5,6 +5,7 @@ import type { StructuredAgentSessionState } from '../../../../shared/structured-
|
||||
import type { StructuredAgentSubagentRoster } from '../../../../shared/structured-agent-session-subagent-roster'
|
||||
import { selectStructuredAgentTurnActivity } from '../../../../shared/native-chat-turn-activity'
|
||||
import { structuredSessionBackgroundTasksView } from '../../../../shared/structured-session-background-tasks-view'
|
||||
import { structuredSessionForegroundCommands } from '../../../../shared/structured-session-foreground-commands'
|
||||
import { useStructuredAgentTurnTiming } from './use-structured-agent-turn-timing'
|
||||
|
||||
const NO_JOURNAL_ITEMS: StructuredAgentSessionState['items'] = []
|
||||
@@ -41,6 +42,14 @@ export function useStructuredAgentSessionTransportState(
|
||||
},
|
||||
turnId
|
||||
)
|
||||
const backgroundTasks = useMemo(() => {
|
||||
const roster = enabled ? state.backgroundTasks : null
|
||||
return structuredSessionBackgroundTasksView(
|
||||
roster,
|
||||
turnId,
|
||||
structuredSessionForegroundCommands(roster, { items: journalItems, latestTurn })
|
||||
)
|
||||
}, [enabled, journalItems, latestTurn, state.backgroundTasks, turnId])
|
||||
return {
|
||||
journalItems,
|
||||
latestTurn,
|
||||
@@ -56,9 +65,6 @@ export function useStructuredAgentSessionTransportState(
|
||||
queuePause: (enabled ? state.queuePause : null) ?? null,
|
||||
/** Working only because the queue is about to send: nothing is in flight to stop yet. */
|
||||
queueSendsNext,
|
||||
backgroundTasks: structuredSessionBackgroundTasksView(
|
||||
enabled ? state.backgroundTasks : null,
|
||||
turnId
|
||||
)
|
||||
backgroundTasks
|
||||
}
|
||||
}
|
||||
|
||||
@@ -66,6 +66,30 @@ function alias(
|
||||
}
|
||||
|
||||
describe('projectAgentChildWorkViews', () => {
|
||||
it('keeps Codex foreground commands in status records but out of all surface views', () => {
|
||||
const foreground = record('foreground', {
|
||||
provider: 'codex',
|
||||
kind: 'command',
|
||||
residency: 'foreground'
|
||||
})
|
||||
const records = [
|
||||
foreground,
|
||||
record('background', { provider: 'codex', kind: 'command', residency: 'background' }),
|
||||
record('claude', { kind: 'command', residency: 'foreground' }),
|
||||
record('agent', { provider: 'codex', residency: 'foreground' })
|
||||
]
|
||||
expect(agentChildWorkLiveness([foreground])).toBe('monitoring')
|
||||
expect(projectAgentChildWorkViews(records, []).map((view) => view.id)).toEqual([
|
||||
'background',
|
||||
'claude',
|
||||
'agent'
|
||||
])
|
||||
expect(
|
||||
projectAgentChildWorkViews([{ ...foreground, residency: 'background' }], [])
|
||||
).toHaveLength(1)
|
||||
expect(foreground.residency).toBe('foreground')
|
||||
})
|
||||
|
||||
it('carries what a surface reads and drops host bookkeeping', () => {
|
||||
const [view] = projectAgentChildWorkViews(
|
||||
[
|
||||
|
||||
@@ -108,14 +108,18 @@ function resolvedOwner(
|
||||
: undefined
|
||||
}
|
||||
|
||||
/** The one record-to-view projection; every surface and every legacy shape starts here. */
|
||||
/** Every surface starts here; foreground Codex tools stay in status records and tool rows. */
|
||||
export function projectAgentChildWorkViews(
|
||||
records: readonly AgentChildWorkInput[],
|
||||
aliases: readonly AgentChildWorkViewAlias[]
|
||||
): AgentChildWorkView[] {
|
||||
const byId = new Map(records.map((record) => [record.childWorkId, record]))
|
||||
const visible = records.filter(
|
||||
(record) =>
|
||||
record.provider !== 'codex' || record.kind !== 'command' || record.residency !== 'foreground'
|
||||
)
|
||||
const byId = new Map(visible.map((record) => [record.childWorkId, record]))
|
||||
const aliasesByChild = groupedBy(aliases, (alias) => alias.childWorkId)
|
||||
return records.map((record) => {
|
||||
return visible.map((record) => {
|
||||
const providerId = providerIdFor(record, aliasesByChild.get(record.childWorkId) ?? [])
|
||||
const owner = resolvedOwner(record, byId)
|
||||
return {
|
||||
|
||||
@@ -17,6 +17,49 @@ function view(membership: 'live' | 'settled', kind: AgentChildWorkView['kind'] =
|
||||
}
|
||||
|
||||
describe('structuredSessionBackgroundTasksView', () => {
|
||||
it('does not mount the strip for a foreground command from an older host', () => {
|
||||
const id = 'codex-command:primary:pwd'
|
||||
const foreground = new Set([id])
|
||||
expect(
|
||||
structuredSessionBackgroundTasksView(
|
||||
{ state: 'monitoring', tasks: [{ id, kind: 'command' }] },
|
||||
'turn-1',
|
||||
foreground
|
||||
)
|
||||
).toMatchObject({ show: false, isMonitoring: false, tasks: [] })
|
||||
})
|
||||
|
||||
it('removes the foreground command from child views and legacy tasks together', () => {
|
||||
const id = 'codex-command:primary:pwd'
|
||||
const shell = { ...view('live', 'command'), providerId: id }
|
||||
const agent = view('live')
|
||||
const tasks = [
|
||||
{ id, kind: 'command' as const },
|
||||
{ id: 'child-agent', kind: 'agent' as const }
|
||||
]
|
||||
const shown = structuredSessionBackgroundTasksView(
|
||||
{ state: 'monitoring', tasks, children: [shell, agent] },
|
||||
'turn-1',
|
||||
new Set([id])
|
||||
)
|
||||
expect(shown).toMatchObject({ show: true, tasks: [tasks[1]], children: [agent] })
|
||||
})
|
||||
|
||||
it('keeps unrelated rosters and commands from earlier turns during the current turn', () => {
|
||||
const tasks = [
|
||||
{ id: 'codex-command:primary:server', kind: 'command' as const },
|
||||
{ id: 'claude-shell', kind: 'command' as const },
|
||||
{ id: 'codex-command:thread:child:pwd', kind: 'command' as const }
|
||||
]
|
||||
const shown = structuredSessionBackgroundTasksView(
|
||||
{ state: 'monitoring', tasks },
|
||||
'turn-2',
|
||||
new Set(['codex-command:primary:pwd'])
|
||||
)
|
||||
expect(shown.show).toBe(true)
|
||||
expect(shown.tasks).toBe(tasks)
|
||||
})
|
||||
|
||||
it('shows finished children until the next turn, but they hold nothing open', () => {
|
||||
const finished = structuredSessionBackgroundTasksView(
|
||||
{ state: 'monitoring', children: [view('settled')] },
|
||||
|
||||
@@ -1,13 +1,7 @@
|
||||
// The background-tasks strip's view of one session's wire state.
|
||||
//
|
||||
// The strip reports work that is IN FLIGHT, whether or not it outlived a turn:
|
||||
// a fan-out's children keep reporting long after the parent settles, and a
|
||||
// foreground fan-out is running work while the turn is still open. It stays
|
||||
// mounted through a running turn — turn state is not a filter on the rows,
|
||||
// because the producers publish only tasks they still have live evidence for.
|
||||
// A host that publishes its child records sends the running ones only, and no
|
||||
// roster once none runs, so the strip hides. Only an idle session with live
|
||||
// work lets the strip animate or speak for itself.
|
||||
// Hosts select background work; newer clients also exclude their current turn's commands
|
||||
// from an older host's roster. Child agents and earlier commands stay visible mid-turn.
|
||||
|
||||
import type {
|
||||
AgentSessionBackgroundTask,
|
||||
@@ -17,7 +11,7 @@ import { agentChildWorkLiveness } from './agent-status-child-work-liveness'
|
||||
import type { AgentChildWorkView } from './agent-status-child-work-view'
|
||||
|
||||
export type StructuredSessionBackgroundTasksView = {
|
||||
/** The strip renders whenever the host reports rows — mid-turn included. */
|
||||
/** Background rows remain visible during a foreground turn. */
|
||||
show: boolean
|
||||
/** Idle-only: gates the animated monitoring indicator and conversation
|
||||
* commands, never the strip itself. A running turn owns the voice. */
|
||||
@@ -32,24 +26,47 @@ export type StructuredSessionBackgroundTasksView = {
|
||||
|
||||
/** One shared empty list: a roster with no rows of a kind keeps the strip's memo on every render. */
|
||||
const NO_TASKS: AgentSessionBackgroundTask[] = []
|
||||
const NO_FOREGROUND_COMMANDS: ReadonlySet<string> = new Set()
|
||||
|
||||
function withoutForeground<T>(
|
||||
rows: T[],
|
||||
foreground: ReadonlySet<string>,
|
||||
id: (row: T) => string | undefined
|
||||
): T[] {
|
||||
const visible = (row: T) => !foreground.has(id(row) ?? '')
|
||||
return rows.every(visible) ? rows : rows.filter(visible)
|
||||
}
|
||||
|
||||
export function structuredSessionBackgroundTasksView(
|
||||
backgroundTasks: AgentSessionBackgroundTaskState | null | undefined,
|
||||
turnId: string | null
|
||||
turnId: string | null,
|
||||
foregroundCommands: ReadonlySet<string> = NO_FOREGROUND_COMMANDS
|
||||
): StructuredSessionBackgroundTasksView {
|
||||
const monitoring = backgroundTasks?.state === 'monitoring'
|
||||
// Decoded once where the frame entered the client's state, so its identity holds between frames.
|
||||
const children = monitoring ? backgroundTasks.children : undefined
|
||||
const children =
|
||||
monitoring && backgroundTasks.children
|
||||
? withoutForeground(backgroundTasks.children, foregroundCommands, (child) => child.providerId)
|
||||
: undefined
|
||||
const tasks = withoutForeground(
|
||||
backgroundTasks?.tasks ?? NO_TASKS,
|
||||
foregroundCommands,
|
||||
(task) => task.id
|
||||
)
|
||||
// Only running children arrive; a roster with none holds nothing open either way.
|
||||
// An older host's roster whose listed rows have all finished shows nothing that runs.
|
||||
const onlyFinished =
|
||||
!children && !backgroundTasks?.tasks?.length && Boolean(backgroundTasks?.settledTasks?.length)
|
||||
const show = monitoring && !onlyFinished
|
||||
const onlyFinished = !children && !tasks.length && Boolean(backgroundTasks?.settledTasks?.length)
|
||||
const onlyForeground =
|
||||
foregroundCommands.size > 0 &&
|
||||
!children?.length &&
|
||||
!tasks.length &&
|
||||
Boolean(backgroundTasks?.children?.length || backgroundTasks?.tasks?.length)
|
||||
const show = monitoring && !onlyFinished && !onlyForeground
|
||||
const liveWork = children ? agentChildWorkLiveness(children) !== null : show
|
||||
return {
|
||||
show,
|
||||
isMonitoring: turnId === null && liveWork,
|
||||
tasks: backgroundTasks?.tasks ?? NO_TASKS,
|
||||
tasks,
|
||||
// Running work only, from any host: an older host's finished rows are not shown either.
|
||||
settledTasks: NO_TASKS,
|
||||
...(children ? { children } : {}),
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { AgentJournalRenderItem } from './agent-session-journal-types'
|
||||
import type { AgentSessionBackgroundTaskState, AgentSessionLatestTurn } from './agent-session-wire'
|
||||
import { structuredSessionForegroundCommands } from './structured-session-foreground-commands'
|
||||
|
||||
const pwd = 'codex-command:primary:pwd'
|
||||
const server = 'codex-command:primary:server'
|
||||
const running = {
|
||||
itemId: 'turn-row-2',
|
||||
observedAt: 20,
|
||||
turn: { turnId: 'turn-2', state: 'running' }
|
||||
} satisfies AgentSessionLatestTurn
|
||||
|
||||
function tool(callId: string, turnItemId: string): AgentJournalRenderItem {
|
||||
return {
|
||||
itemId: `tool-${callId}`,
|
||||
sequence: 2,
|
||||
revision: 1,
|
||||
observedAt: 21,
|
||||
turnScope: { kind: 'turn', turnItemId },
|
||||
body: { kind: 'tool-call', callId, name: 'shell', input: null, state: 'running' }
|
||||
}
|
||||
}
|
||||
|
||||
const roster: AgentSessionBackgroundTaskState = {
|
||||
state: 'monitoring',
|
||||
tasks: [
|
||||
{ id: pwd, kind: 'command' },
|
||||
{ id: server, kind: 'command' },
|
||||
{ id: 'codex-command:thread:child:exec', kind: 'command' },
|
||||
{ id: 'claude-task', kind: 'command' }
|
||||
]
|
||||
}
|
||||
|
||||
describe('structuredSessionForegroundCommands', () => {
|
||||
it('handles a roster arriving before its tool row, without hiding earlier work', () => {
|
||||
expect([
|
||||
...structuredSessionForegroundCommands(roster, {
|
||||
items: [tool('server', 'turn-row-1')],
|
||||
latestTurn: running
|
||||
})
|
||||
]).toEqual([pwd])
|
||||
expect([
|
||||
...structuredSessionForegroundCommands(roster, {
|
||||
items: [tool('server', 'turn-row-1'), tool('pwd', running.itemId)],
|
||||
latestTurn: running
|
||||
})
|
||||
]).toEqual([pwd])
|
||||
})
|
||||
|
||||
it('keeps an earlier command whose journal row is outside the loaded page', () => {
|
||||
const children = [
|
||||
{
|
||||
id: 'child-server',
|
||||
providerId: server,
|
||||
kind: 'command' as const,
|
||||
state: 'working' as const,
|
||||
membership: 'live' as const,
|
||||
firstObservedAt: 10,
|
||||
observedAt: 21,
|
||||
stoppable: false,
|
||||
invocation: { invocationId: server, generation: 1 }
|
||||
}
|
||||
]
|
||||
expect([
|
||||
...structuredSessionForegroundCommands(
|
||||
{ ...roster, children },
|
||||
{
|
||||
items: [],
|
||||
latestTurn: running
|
||||
}
|
||||
)
|
||||
]).toEqual([pwd])
|
||||
})
|
||||
|
||||
it('reveals surviving commands when the source turn ends', () => {
|
||||
expect(
|
||||
structuredSessionForegroundCommands(roster, {
|
||||
items: [],
|
||||
latestTurn: { ...running, turn: { ...running.turn, state: 'completed' } }
|
||||
}).size
|
||||
).toBe(0)
|
||||
})
|
||||
|
||||
it('uses loaded turn rows when an older host has no latest-turn field', () => {
|
||||
const turn: AgentJournalRenderItem = {
|
||||
itemId: running.itemId,
|
||||
sequence: 1,
|
||||
revision: 1,
|
||||
observedAt: 20,
|
||||
body: { kind: 'turn', ...running.turn }
|
||||
}
|
||||
expect([
|
||||
...structuredSessionForegroundCommands(roster, {
|
||||
items: [turn, tool('server', 'turn-row-1')]
|
||||
})
|
||||
]).toEqual([pwd])
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,64 @@
|
||||
import type { AgentJournalRenderItem } from './agent-session-journal-types'
|
||||
import { isRootAgentJournalItem } from './agent-session-journal-producer'
|
||||
import type { AgentSessionBackgroundTaskState, AgentSessionLatestTurn } from './agent-session-wire'
|
||||
import {
|
||||
runningStructuredAgentSessionTurnId,
|
||||
runningStructuredAgentSessionTurnScope
|
||||
} from './structured-agent-session-live-turn'
|
||||
|
||||
const PRIMARY_COMMAND_PREFIX = 'codex-command:primary:'
|
||||
|
||||
export function codexPrimaryCommandTaskId(callId: string): string {
|
||||
return `${PRIMARY_COMMAND_PREFIX}${encodeURIComponent(callId)}`
|
||||
}
|
||||
|
||||
/** Older hosts list foreground shells too; their journal still identifies which turn owns one. */
|
||||
export function structuredSessionForegroundCommands(
|
||||
roster: AgentSessionBackgroundTaskState | null | undefined,
|
||||
source: { items: readonly AgentJournalRenderItem[]; latestTurn?: AgentSessionLatestTurn | null }
|
||||
): ReadonlySet<string> {
|
||||
const foreground = new Set<string>()
|
||||
const turnId = runningStructuredAgentSessionTurnId(source)
|
||||
if (turnId === null || !roster) {
|
||||
return foreground
|
||||
}
|
||||
const scope = runningStructuredAgentSessionTurnScope(source)
|
||||
const owningTurnItems = new Map<string, string>()
|
||||
for (const item of source.items) {
|
||||
if (
|
||||
isRootAgentJournalItem(item) &&
|
||||
item.body.kind === 'tool-call' &&
|
||||
item.body.callId &&
|
||||
item.turnScope?.kind === 'turn'
|
||||
) {
|
||||
owningTurnItems.set(codexPrimaryCommandTaskId(item.body.callId), item.turnScope.turnItemId)
|
||||
}
|
||||
}
|
||||
const firstObserved = new Map(
|
||||
(roster.children ?? []).flatMap((child) =>
|
||||
child.providerId ? [[child.providerId, child.firstObservedAt] as const] : []
|
||||
)
|
||||
)
|
||||
const ids = new Set([
|
||||
...(roster.tasks ?? []).map((task) => task.id),
|
||||
...(roster.children ?? []).flatMap((child) => child.providerId ?? [])
|
||||
])
|
||||
for (const id of ids) {
|
||||
if (!id.startsWith(PRIMARY_COMMAND_PREFIX)) {
|
||||
continue
|
||||
}
|
||||
const observedAt = firstObserved.get(id)
|
||||
const owningTurnItem = owningTurnItems.get(id)
|
||||
const earlierTurn =
|
||||
owningTurnItem !== undefined && scope.kind === 'turn'
|
||||
? owningTurnItem !== scope.turnItemId
|
||||
: observedAt !== undefined &&
|
||||
source.latestTurn != null &&
|
||||
observedAt < source.latestTurn.observedAt
|
||||
// With no ownership yet, keep an older host's start frame in the tool row until the turn ends.
|
||||
if (!earlierTurn) {
|
||||
foreground.add(id)
|
||||
}
|
||||
}
|
||||
return foreground
|
||||
}
|
||||
Reference in New Issue
Block a user