From 13ef6a406fb5ecc20f3e5f251f052957d7858414 Mon Sep 17 00:00:00 2001 From: Badcuban <108198679+badcuban@users.noreply.github.com> Date: Sat, 15 Aug 2026 00:32:15 -0400 Subject: [PATCH] Promote background codex exec runs into subagent rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Claude thread that launches `codex exec …` as a background Bash command is running a second agent, but the SDK only reports it as a `local_bash` task, so the agents panel showed a bare run row: no model, no effort, no transcript, nothing to click into. The Claude adapter now recognises a `codex exec` invocation from the originating Bash tool call and emits the same `subagent.metadata.updated` events the Codex adapter emits for native subagents. The row appears on the task's start edge with the role, the task description as its goal, and any `-m` / `-c model_reasoning_effort=` the command stated. A scoped poll then correlates the run with the rollout JSONL codex writes under `$CODEX_HOME/sessions/`, matching on the exec originator and a realpath-equal cwd, and upgrades the row with the transcript id, the resolved model and the effort from the file's first `turn_context`. The task's completion settles the row, with codex's own final message as the result. `readSubagentTranscript` serves those rollouts for `codex-exec:` ids, resolving from the id alone so a transcript still opens after a restart. Web-side, a task whose tool call the thread already tracks as a subagent is kept off the run list, and the run it would have been is handed to the matching agent row so the row keeps the same stop arm the run row had. New payload keys on `SubagentMetadataUpdatedPayload` (status, transcriptAgentId, resolvedModel, resultBody, resultCreatedAt) were already read by the subagent projection; they are now declared so emitters can set them. --- .../src/orchestration/projector.test.ts | 21 +- .../Drivers/CodexExecRollouts.test.ts | 396 +++++++++ .../src/provider/Drivers/CodexExecRollouts.ts | 798 ++++++++++++++++++ .../src/provider/Layers/ClaudeAdapter.test.ts | 272 ++++++ .../src/provider/Layers/ClaudeAdapter.ts | 359 ++++++++ apps/web/src/agentsPanelStore.ts | 3 + .../web/src/components/ChatView.logic.test.ts | 44 + apps/web/src/components/ChatView.logic.ts | 50 +- apps/web/src/components/ChatView.tsx | 68 +- apps/web/src/components/chat/AgentsPanel.tsx | 18 +- .../components/chat/agentsPanel.logic.test.ts | 35 + .../src/components/chat/agentsPanel.logic.ts | 17 +- .../routes/_chat.$environmentId.$threadId.tsx | 1 + apps/web/src/session-logic.ts | 7 + packages/contracts/src/providerRuntime.ts | 13 + 15 files changed, 2073 insertions(+), 29 deletions(-) create mode 100644 apps/server/src/provider/Drivers/CodexExecRollouts.test.ts create mode 100644 apps/server/src/provider/Drivers/CodexExecRollouts.ts diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index b6e015002..fe8e8787b 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -154,6 +154,7 @@ describe("orchestration projector", () => { payload: { callId: "spawn-call-1", agentThreadId: "agent-thread-1", + transcriptAgentId: "codex-exec:agent-thread-1", agentPath: "/root/durable_identity", agentNickname: "Identity", model: "gpt-5.6-sol-2026-08-01", @@ -161,6 +162,17 @@ describe("orchestration projector", () => { status: "running", }, }, + // A provider that settles an agent out-of-band states only the spawn call + // and the outcome; everything already known has to survive it. + { + id: "settle-metadata", + payload: { + callId: "spawn-call-1", + status: "completed", + resultBody: "Traced it.", + resultCreatedAt: "2026-08-13T12:00:09.000Z", + }, + }, ]; for (const [index, activity] of metadataActivities.entries()) { @@ -196,7 +208,7 @@ describe("orchestration projector", () => { projectEvent( model, makeEvent({ - sequence: index + 4, + sequence: index + metadataActivities.length + 2, type: "thread.activity-appended", aggregateKind: "thread", aggregateId: "thread-subagents", @@ -211,7 +223,7 @@ describe("orchestration projector", () => { summary: `Filler ${index}`, payload: {}, turnId: null, - sequence: index + 4, + sequence: index + metadataActivities.length + 2, createdAt: "2026-08-13T12:01:00.000Z", }, }, @@ -229,16 +241,19 @@ describe("orchestration projector", () => { expect.objectContaining({ id: "agent-thread-1", agentThreadId: "agent-thread-1", + transcriptAgentId: "codex-exec:agent-thread-1", spawnCallId: "spawn-call-1", nickname: "Identity", role: "durable_identity", objective: "Trace durable identity", - status: "running", + status: "completed", requestedModel: "gpt-5.6-sol", resolvedModel: "gpt-5.6-sol-2026-08-01", reasoningEffort: "high", modelProvenance: "explicit", reasoningEffortProvenance: "explicit", + resultBody: "Traced it.", + resultCreatedAt: "2026-08-13T12:00:09.000Z", }), ]); }); diff --git a/apps/server/src/provider/Drivers/CodexExecRollouts.test.ts b/apps/server/src/provider/Drivers/CodexExecRollouts.test.ts new file mode 100644 index 000000000..f09afc4e4 --- /dev/null +++ b/apps/server/src/provider/Drivers/CodexExecRollouts.test.ts @@ -0,0 +1,396 @@ +// @effect-diagnostics nodeBuiltinImport:off +import { mkdirSync, mkdtempSync, rmSync, utimesSync, writeFileSync } from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Path from "effect/Path"; + +import { + codexExecSessionsRoot, + findCodexExecRollout, + locateCodexExecRolloutBySessionId, + mapCodexExecRolloutTranscript, + matchCodexExecCommand, + parseCodexExecAgentId, + readCodexExecFinalMessage, + readCodexExecRolloutHead, + readCodexExecTurnContext, +} from "./CodexExecRollouts.ts"; + +const SESSION_A = "019fea48-4154-7982-876b-43e4c551eb65"; +const SESSION_B = "019fea48-4154-7982-876b-43e4c551eb66"; + +function line(record: unknown): string { + return JSON.stringify(record); +} + +function sessionMeta(input: { + readonly sessionId: string; + readonly cwd: string; + readonly originator?: string; + readonly source?: string; + readonly startedAt?: string; +}): string { + return line({ + timestamp: input.startedAt ?? "2026-08-10T06:07:11.861Z", + type: "session_meta", + payload: { + session_id: input.sessionId, + cwd: input.cwd, + originator: input.originator ?? "codex_exec", + source: input.source ?? "exec", + cli_version: "0.147.0", + }, + }); +} + +function turnContext(model: string, effort: string): string { + return line({ + timestamp: "2026-08-10T06:07:12.000Z", + type: "turn_context", + payload: { turn_id: "turn-1", cwd: "/tmp/x", model, effort }, + }); +} + +function makeSessionsRoot(): { readonly root: string; readonly cleanup: () => void } { + const home = mkdtempSync(path.join(os.tmpdir(), "codex-exec-rollouts-")); + const root = path.join(home, "sessions"); + mkdirSync(root, { recursive: true }); + return { root, cleanup: () => rmSync(home, { recursive: true, force: true }) }; +} + +function writeRollout(input: { + readonly root: string; + readonly date: Date; + readonly sessionId: string; + readonly lines: ReadonlyArray; + readonly mtimeMs?: number; +}): string { + const year = String(input.date.getFullYear()); + const month = String(input.date.getMonth() + 1).padStart(2, "0"); + const day = String(input.date.getDate()).padStart(2, "0"); + const directory = path.join(input.root, year, month, day); + mkdirSync(directory, { recursive: true }); + const filePath = path.join( + directory, + `rollout-${year}-${month}-${day}T00-00-00-${input.sessionId}.jsonl`, + ); + writeFileSync(filePath, `${input.lines.join("\n")}\n`); + if (input.mtimeMs !== undefined) { + const seconds = input.mtimeMs / 1_000; + utimesSync(filePath, seconds, seconds); + } + return filePath; +} + +describe("matchCodexExecCommand", () => { + it("recognises codex exec invocations and reads their explicit flags", () => { + assert.deepEqual(matchCodexExecCommand(`codex exec "do the thing"`), { + prompt: "do the thing", + }); + assert.deepEqual(matchCodexExecCommand(`codex e -m gpt-5.6-sol 'review this'`), { + model: "gpt-5.6-sol", + prompt: "review this", + }); + assert.deepEqual(matchCodexExecCommand(`CODEX_HOME=/x codex exec "go"`), { prompt: "go" }); + assert.deepEqual(matchCodexExecCommand(`timeout 600 codex exec "go"`), { prompt: "go" }); + assert.deepEqual( + matchCodexExecCommand(`codex exec -c model_reasoning_effort="high" --model=gpt-5.6-sol "go"`), + { model: "gpt-5.6-sol", reasoningEffort: "high", prompt: "go" }, + ); + assert.deepEqual(matchCodexExecCommand(`cd /tmp/work && /usr/local/bin/codex exec "go"`), { + prompt: "go", + }); + }); + + it("rejects commands that only mention codex exec", () => { + for (const command of [ + "codex review", + "codex apply", + "echo codex exec", + "vp run codex-things", + `python -c "subprocess.run(['codex', 'exec', 'go'])"`, + `echo "codex exec go" > /tmp/note`, + "npx codex exec go", + "", + ]) { + assert.equal(matchCodexExecCommand(command), null, command); + } + }); +}); + +describe("parseCodexExecAgentId", () => { + it("accepts only prefixed uuids", () => { + assert.equal(parseCodexExecAgentId(`codex-exec:${SESSION_A}`), SESSION_A); + assert.equal(parseCodexExecAgentId("codex-exec:../../etc/passwd"), null); + assert.equal(parseCodexExecAgentId("agent-123"), null); + }); +}); + +describe("readCodexExecRolloutHead", () => { + it("accepts exec rollouts and rejects app-server ones", () => { + assert.deepEqual( + readCodexExecRolloutHead(sessionMeta({ sessionId: SESSION_A, cwd: "/tmp/work" })), + { sessionId: SESSION_A, cwd: "/tmp/work", startedAt: "2026-08-10T06:07:11.861Z" }, + ); + assert.equal( + readCodexExecRolloutHead( + sessionMeta({ + sessionId: SESSION_A, + cwd: "/tmp/work", + originator: "threadlines_desktop", + source: "vscode", + }), + ), + null, + ); + assert.equal(readCodexExecRolloutHead("not json"), null); + }); +}); + +describe("readCodexExecTurnContext", () => { + it("reads model and effort from the first turn context", () => { + assert.deepEqual( + readCodexExecTurnContext([ + sessionMeta({ sessionId: SESSION_A, cwd: "/tmp/work" }), + turnContext("gpt-5.6-sol", "medium"), + turnContext("gpt-5.6-terra", "high"), + ]), + { model: "gpt-5.6-sol", reasoningEffort: "medium" }, + ); + assert.equal( + readCodexExecTurnContext([sessionMeta({ sessionId: SESSION_A, cwd: "/tmp/work" })]), + null, + ); + }); +}); + +describe("mapCodexExecRolloutTranscript", () => { + it("maps prompts, reasoning, tool calls and their output into entries", () => { + const transcript = mapCodexExecRolloutTranscript([ + sessionMeta({ sessionId: SESSION_A, cwd: "/tmp/work" }), + turnContext("gpt-5.6-sol", "medium"), + line({ type: "world_state", payload: { anything: true } }), + line({ + timestamp: "2026-08-10T06:07:13.000Z", + type: "event_msg", + payload: { type: "user_message", message: "Verify the usage page" }, + }), + line({ + type: "response_item", + payload: { type: "reasoning", summary: [{ type: "summary_text", text: "Plan the check" }] }, + }), + line({ + type: "response_item", + payload: { + type: "custom_tool_call", + name: "exec", + call_id: "call-1", + input: "ls /tmp/work", + }, + }), + line({ + type: "response_item", + payload: { + type: "custom_tool_call_output", + call_id: "call-1", + output: [{ type: "input_text", text: "usage.tsx" }], + }, + }), + line({ + type: "event_msg", + payload: { type: "agent_message", message: "The page renders." }, + }), + "not-json", + ]); + + assert.equal(transcript.model, "gpt-5.6-sol"); + assert.equal(transcript.cwd, "/tmp/work"); + assert.equal(transcript.prompt, "Verify the usage page"); + assert.deepEqual( + transcript.entries.map((entry) => ({ + role: entry.role, + text: entry.text, + tools: entry.toolUses.map((tool) => `${tool.name}:${tool.summary}`), + ...(entry.outputPreview ? { outputPreview: entry.outputPreview } : {}), + })), + [ + { role: "user", text: "Verify the usage page", tools: [] }, + { role: "thinking", text: "Plan the check", tools: [] }, + { + role: "assistant", + text: "", + tools: ["exec:ls /tmp/work"], + outputPreview: "usage.tsx", + }, + { role: "assistant", text: "The page renders.", tools: [] }, + ], + ); + assert.equal(transcript.entries[0]?.at, "2026-08-10T06:07:13.000Z"); + }); +}); + +describe("readCodexExecFinalMessage", () => { + it("prefers the newest completion message", () => { + assert.equal( + readCodexExecFinalMessage([ + line({ type: "event_msg", payload: { type: "agent_message", message: "working" } }), + line({ + type: "event_msg", + payload: { type: "task_complete", last_agent_message: "all done" }, + }), + ]), + "all done", + ); + assert.equal(readCodexExecFinalMessage(["not json"]), null); + }); +}); + +describe("codexExecSessionsRoot", () => { + it.effect("follows CODEX_HOME when set, else the default codex home", () => + Effect.gen(function* () { + const pathService = yield* Path.Path; + assert.equal( + codexExecSessionsRoot({ CODEX_HOME: "/tmp/codex-work" }, pathService), + path.join("/tmp/codex-work", "sessions"), + ); + assert.equal( + codexExecSessionsRoot({}, pathService), + path.join(os.homedir(), ".codex", "sessions"), + ); + }).pipe(Effect.provide(NodeServices.layer)), + ); +}); + +describe("findCodexExecRollout", () => { + const cwd = os.tmpdir(); + + it.effect("claims the earliest matching rollout and never reuses a claimed one", () => + Effect.gen(function* () { + const { root, cleanup } = makeSessionsRoot(); + try { + const startedAtMs = Date.now(); + const startedAtIso = new Date(startedAtMs).toISOString(); + const first = writeRollout({ + root, + date: new Date(startedAtMs), + sessionId: SESSION_A, + lines: [ + sessionMeta({ sessionId: SESSION_A, cwd, startedAt: startedAtIso }), + turnContext("gpt-5.6-sol", "medium"), + ], + mtimeMs: startedAtMs + 1_000, + }); + const second = writeRollout({ + root, + date: new Date(startedAtMs), + sessionId: SESSION_B, + lines: [sessionMeta({ sessionId: SESSION_B, cwd, startedAt: startedAtIso })], + mtimeMs: startedAtMs + 2_000, + }); + // An exec run that began ten minutes before this task, still writing: + // its mtime is fresh, so only its recorded start time rules it out. + // Earliest-mtime ordering would otherwise hand it every match. + writeRollout({ + root, + date: new Date(startedAtMs), + sessionId: "019fea48-4154-7982-876b-43e4c551eb69", + lines: [ + sessionMeta({ + sessionId: "019fea48-4154-7982-876b-43e4c551eb69", + cwd, + startedAt: new Date(startedAtMs - 600_000).toISOString(), + }), + ], + mtimeMs: startedAtMs + 500, + }); + // Same directory, but written by the desktop app-server rather than an + // exec run, and for an unrelated directory. + writeRollout({ + root, + date: new Date(startedAtMs), + sessionId: "019fea48-4154-7982-876b-43e4c551eb67", + lines: [ + sessionMeta({ + sessionId: "019fea48-4154-7982-876b-43e4c551eb67", + cwd, + originator: "threadlines_desktop", + source: "vscode", + }), + ], + mtimeMs: startedAtMs, + }); + writeRollout({ + root, + date: new Date(startedAtMs), + sessionId: "019fea48-4154-7982-876b-43e4c551eb68", + lines: [ + sessionMeta({ + sessionId: "019fea48-4154-7982-876b-43e4c551eb68", + cwd: path.join(cwd, "somewhere-else"), + }), + ], + mtimeMs: startedAtMs, + }); + + const claimed = new Set(); + const firstMatch = yield* findCodexExecRollout({ + sessionsRoot: root, + cwd, + notBeforeMs: startedAtMs, + claimedPaths: claimed, + }); + assert.deepEqual(firstMatch, { rolloutPath: first, sessionId: SESSION_A }); + + claimed.add(first); + const secondMatch = yield* findCodexExecRollout({ + sessionsRoot: root, + cwd, + notBeforeMs: startedAtMs, + claimedPaths: claimed, + }); + assert.deepEqual(secondMatch, { rolloutPath: second, sessionId: SESSION_B }); + + claimed.add(second); + assert.equal( + yield* findCodexExecRollout({ + sessionsRoot: root, + cwd, + notBeforeMs: startedAtMs, + claimedPaths: claimed, + }), + null, + ); + } finally { + cleanup(); + } + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("resolves a rollout from its session id alone", () => + Effect.gen(function* () { + const { root, cleanup } = makeSessionsRoot(); + try { + const rolloutPath = writeRollout({ + root, + date: new Date("2026-08-10T12:00:00Z"), + sessionId: SESSION_A, + lines: [sessionMeta({ sessionId: SESSION_A, cwd })], + }); + assert.equal( + yield* locateCodexExecRolloutBySessionId({ sessionsRoot: root, sessionId: SESSION_A }), + rolloutPath, + ); + assert.equal( + yield* locateCodexExecRolloutBySessionId({ sessionsRoot: root, sessionId: SESSION_B }), + null, + ); + } finally { + cleanup(); + } + }).pipe(Effect.provide(NodeServices.layer)), + ); +}); diff --git a/apps/server/src/provider/Drivers/CodexExecRollouts.ts b/apps/server/src/provider/Drivers/CodexExecRollouts.ts new file mode 100644 index 000000000..5176aa9a5 --- /dev/null +++ b/apps/server/src/provider/Drivers/CodexExecRollouts.ts @@ -0,0 +1,798 @@ +/** + * CodexExecRollouts — recognise `codex exec` runs and read the rollout files + * they leave behind. + * + * A Claude thread that shells out to `codex exec …` in the background is, from + * the user's point of view, running a second agent. The Claude SDK only reports + * it as a `local_bash` task with a command string, so everything else that + * makes an agent row rich — which model ran, at what effort, and what it + * actually did — has to come from the rollout JSONL every `codex exec` + * invocation writes under `$CODEX_HOME/sessions/YYYY/MM/DD/`. + * + * This module owns the three jobs that needs: deciding whether a shell command + * really is a `codex exec` invocation, correlating a started task with the + * rollout file it produced, and mapping that file into transcript entries. + * + * @module provider/Drivers/CodexExecRollouts + */ +// @effect-diagnostics nodeBuiltinImport:off +import { homedir } from "node:os"; + +import type { ProviderSubagentTranscriptEntry } from "@threadlines/contracts"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import type * as PlatformError from "effect/PlatformError"; + +import { expandHomePath } from "../../pathExpansion.ts"; + +/** Prefix that marks a subagent id as a `codex exec` rollout rather than a + * provider-native agent thread. The remainder is the rollout's session id. */ +export const CODEX_EXEC_AGENT_ID_PREFIX = "codex-exec:"; + +const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/iu; + +const TRANSCRIPT_TEXT_MAX_CHARS = 4_000; +const TRANSCRIPT_OUTPUT_PREVIEW_MAX_CHARS = 2_000; +/** Head lines scanned for the first `turn_context`. It lands within the first + * handful of records; a cap keeps a huge rollout from being read whole. */ +const HEAD_SCAN_MAX_LINES = 64; +/** Date directories walked when resolving a rollout from its id alone. Codex + * files one directory per day, so this is years of history. */ +const SESSION_DATE_DIRECTORY_SCAN_LIMIT = 90; +const READ_CHUNK_BYTES = 64 * 1_024; +/** Ceiling on the `session_meta` line, past which the file is not one of ours. */ +const FIRST_LINE_MAX_CHARS = 512 * 1_024; + +export interface CodexExecInvocation { + /** Model from an explicit `-m`/`--model` flag. */ + readonly model?: string; + /** Effort from an explicit `-c model_reasoning_effort=…`. */ + readonly reasoningEffort?: string; + /** The prompt argument, when it is a plain positional. */ + readonly prompt?: string; +} + +/** `codex-exec:` → ``; anything else is not one of ours. */ +export function parseCodexExecAgentId(agentId: string): string | null { + if (!agentId.startsWith(CODEX_EXEC_AGENT_ID_PREFIX)) { + return null; + } + const sessionId = agentId.slice(CODEX_EXEC_AGENT_ID_PREFIX.length); + return UUID_PATTERN.test(sessionId) ? sessionId.toLowerCase() : null; +} + +export function codexExecAgentId(sessionId: string): string { + return `${CODEX_EXEC_AGENT_ID_PREFIX}${sessionId}`; +} + +interface ShellToken { + readonly text: string; + /** False when any part of the token came from inside quotes. A quoted word + * can never be the command being run, which is what keeps + * `python -c "codex exec …"` from reading as a codex invocation. */ + readonly bare: boolean; +} + +/** + * Split a command line into words the way a shell would, keeping track of + * which words were quoted. Operators that start a new command + * (`&&`, `||`, `;`, `|`, newline) become their own segment boundary. + */ +function tokenizeShellSegments(command: string): ReadonlyArray> { + const segments: Array> = [[]]; + let current: string | null = null; + let bare = true; + + const pushToken = () => { + if (current !== null) { + segments[segments.length - 1]!.push({ text: current, bare }); + current = null; + bare = true; + } + }; + const pushSegment = () => { + pushToken(); + segments.push([]); + }; + + for (let index = 0; index < command.length; index += 1) { + const char = command[index]!; + if (char === "\\" && index + 1 < command.length) { + // A line continuation joins the next line; any other escape contributes + // its literal character to the current word. + const next = command[index + 1]!; + index += 1; + if (next !== "\n") { + current = (current ?? "") + next; + } + continue; + } + if (char === `"` || char === "'") { + const closing = command.indexOf(char, index + 1); + const body = closing === -1 ? command.slice(index + 1) : command.slice(index + 1, closing); + current = (current ?? "") + body; + bare = false; + index = closing === -1 ? command.length : closing; + continue; + } + if (char === "\n" || char === ";") { + pushSegment(); + continue; + } + if (char === "&" || char === "|") { + // `cmd &` backgrounds rather than chains, but either way the next word + // starts a fresh command position, which is all we need. + if (command[index + 1] === char) { + index += 1; + } + pushSegment(); + continue; + } + if (char === " " || char === "\t" || char === "\r") { + pushToken(); + continue; + } + current = (current ?? "") + char; + } + pushToken(); + return segments.filter((segment) => segment.length > 0); +} + +function commandBasename(value: string): string { + const normalized = value.replace(/\\/gu, "/"); + const lastSlash = normalized.lastIndexOf("/"); + return lastSlash === -1 ? normalized : normalized.slice(lastSlash + 1); +} + +/** `NAME=value`, the only thing allowed to precede the program name. */ +function isEnvironmentAssignment(token: ShellToken): boolean { + return token.bare && /^[A-Za-z_][A-Za-z0-9_]*=/u.test(token.text); +} + +/** Wrappers that run their tail as a command, so `codex` may still follow. */ +const COMMAND_WRAPPERS = new Set(["env", "nohup", "timeout", "gtimeout", "stdbuf", "nice"]); + +function skipWrappers(tokens: ReadonlyArray): number { + let index = 0; + for (let guard = 0; guard < 8; guard += 1) { + while (index < tokens.length && isEnvironmentAssignment(tokens[index]!)) { + index += 1; + } + const token = tokens[index]; + if (!token?.bare || !COMMAND_WRAPPERS.has(commandBasename(token.text))) { + return index; + } + const wrapper = commandBasename(token.text); + index += 1; + // Wrapper options and their operands sit between the wrapper and the real + // program. `timeout`'s duration is a bare positional, so it is consumed + // explicitly; everything else stops at the first non-flag word. + while (index < tokens.length && tokens[index]!.text.startsWith("-")) { + index += 1; + } + if ( + (wrapper === "timeout" || wrapper === "gtimeout") && + index < tokens.length && + /^\d+(?:\.\d+)?[smhd]?$/u.test(tokens[index]!.text) + ) { + index += 1; + } + } + return index; +} + +function unquotedFlagValue(value: string): string | undefined { + const trimmed = value.trim(); + return trimmed.length > 0 ? trimmed : undefined; +} + +/** + * Whether this shell command runs `codex exec`, and what it asked for. + * + * Deliberately strict: the program word must be a bare `codex` (possibly + * path-qualified, possibly behind env assignments or a wrapper like + * `timeout`) at the head of a command position, and the very next word must be + * the `exec`/`e` subcommand. Anything mentioning "codex exec" as an argument, + * inside quotes, or under another program is not a match — a false positive + * would promote an unrelated background command into an agent row, which is + * worse than leaving a real one as a plain run. + */ +export function matchCodexExecCommand(command: string): CodexExecInvocation | null { + if (!/\bcodex\b/u.test(command)) { + return null; + } + for (const tokens of tokenizeShellSegments(command)) { + const start = skipWrappers(tokens); + const program = tokens[start]; + if (!program?.bare || commandBasename(program.text) !== "codex") { + continue; + } + const subcommand = tokens[start + 1]; + if (!subcommand?.bare || (subcommand.text !== "exec" && subcommand.text !== "e")) { + continue; + } + return readCodexExecFlags(tokens.slice(start + 2)); + } + return null; +} + +function readCodexExecFlags(tokens: ReadonlyArray): CodexExecInvocation { + let model: string | undefined; + let reasoningEffort: string | undefined; + let prompt: string | undefined; + + for (let index = 0; index < tokens.length; index += 1) { + const token = tokens[index]!; + const text = token.text; + if (text === "-m" || text === "--model") { + model = unquotedFlagValue(tokens[index + 1]?.text ?? ""); + index += 1; + continue; + } + if (text.startsWith("--model=")) { + model = unquotedFlagValue(text.slice("--model=".length)); + continue; + } + if (text === "-c" || text === "--config") { + const override = tokens[index + 1]?.text ?? ""; + reasoningEffort = readReasoningEffortOverride(override) ?? reasoningEffort; + index += 1; + continue; + } + if (text.startsWith("--config=")) { + reasoningEffort = + readReasoningEffortOverride(text.slice("--config=".length)) ?? reasoningEffort; + continue; + } + if (text.startsWith("-")) { + continue; + } + if (prompt === undefined && text.trim().length > 0) { + prompt = text.trim(); + } + } + + return { + ...(model ? { model } : {}), + ...(reasoningEffort ? { reasoningEffort } : {}), + ...(prompt ? { prompt } : {}), + }; +} + +function readReasoningEffortOverride(override: string): string | undefined { + const match = /^model_reasoning_effort\s*=\s*(.+)$/u.exec(override.trim()); + if (!match) { + return undefined; + } + return unquotedFlagValue(match[1]!.replace(/^["']|["']$/gu, "")); +} + +/** `$CODEX_HOME/sessions`, else `~/.codex/sessions`. */ +export function codexExecSessionsRoot(environment: NodeJS.ProcessEnv, path: Path.Path): string { + const configured = environment.CODEX_HOME?.trim(); + const home = + configured && configured.length > 0 + ? expandHomePath(configured) + : path.join(homedir(), ".codex"); + return path.resolve(path.join(home, "sessions")); +} + +export interface CodexExecRolloutHead { + readonly sessionId: string; + readonly cwd: string | null; + readonly startedAt: string | null; +} + +function readRecord(value: unknown): Record | null { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? (value as Record) + : null; +} + +function readText(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + +function parseJsonRecord(line: string): Record | null { + const trimmed = line.trim(); + if (trimmed.length === 0 || !trimmed.startsWith("{")) { + return null; + } + try { + return readRecord(JSON.parse(trimmed)); + } catch { + return null; + } +} + +/** + * The `session_meta` first line, but only for rollouts an `exec` run wrote. + * App-server sessions share the same directory and filename shape, so the + * originator is what separates them. + */ +export function readCodexExecRolloutHead(firstLine: string): CodexExecRolloutHead | null { + const record = parseJsonRecord(firstLine); + if (!record || record.type !== "session_meta") { + return null; + } + const payload = readRecord(record.payload); + if (!payload) { + return null; + } + if (readText(payload.originator) !== "codex_exec" && readText(payload.source) !== "exec") { + return null; + } + const sessionId = readText(payload.session_id) ?? readText(payload.id); + if (!sessionId) { + return null; + } + return { + sessionId, + cwd: readText(payload.cwd), + startedAt: readText(payload.timestamp) ?? readText(record.timestamp), + }; +} + +export interface CodexExecTurnContext { + readonly model?: string; + readonly reasoningEffort?: string; +} + +/** Model and effort from the first `turn_context` in the file's head. */ +export function readCodexExecTurnContext(lines: Iterable): CodexExecTurnContext | null { + let scanned = 0; + for (const line of lines) { + if (scanned >= HEAD_SCAN_MAX_LINES) { + break; + } + scanned += 1; + const record = parseJsonRecord(line); + if (!record || record.type !== "turn_context") { + continue; + } + const payload = readRecord(record.payload); + if (!payload) { + continue; + } + const model = readText(payload.model); + const effort = + readText(payload.effort) ?? + readText(readRecord(readRecord(payload.collaboration_mode)?.settings)?.reasoning_effort); + if (!model && !effort) { + continue; + } + return { + ...(model ? { model } : {}), + ...(effort ? { reasoningEffort: effort } : {}), + }; + } + return null; +} + +/** Codex's own last word on the run, for the agent row's result line. */ +export function readCodexExecFinalMessage(lines: Iterable): string | null { + let latest: string | null = null; + for (const line of lines) { + const record = parseJsonRecord(line); + if (!record) { + continue; + } + const payload = readRecord(record.payload); + if (!payload) { + continue; + } + if (record.type === "event_msg" && payload.type === "task_complete") { + latest = readText(payload.last_agent_message) ?? latest; + continue; + } + if (record.type === "event_msg" && payload.type === "agent_message") { + latest = readText(payload.message) ?? latest; + } + } + return latest; +} + +function capText(value: string, maxChars: number): string { + const trimmed = value.trim(); + return trimmed.length > maxChars ? trimmed.slice(0, maxChars) : trimmed; +} + +function readContentText(value: unknown): string { + if (typeof value === "string") { + return value; + } + if (!Array.isArray(value)) { + return ""; + } + return value + .map((part) => { + const record = readRecord(part); + return record && typeof record.text === "string" ? record.text : ""; + }) + .filter((text) => text.length > 0) + .join("\n"); +} + +export interface CodexExecTranscript { + readonly entries: ReadonlyArray; + readonly model?: string; + /** Where the run worked, for the transcript header when there is no prompt. */ + readonly cwd?: string; + /** The operator's own instruction, which reads better than the cwd. */ + readonly prompt?: string; +} + +/** + * Map a rollout JSONL into renderable transcript entries. + * + * Codex records the same turn twice — once as raw `response_item`s for the + * model and once as `event_msg` summaries for the UI — so this reads the + * summaries for prose (they carry the user's actual prompt and the agent's + * finished messages, without the developer-instruction preamble) and the + * response items for the work (reasoning summaries and tool calls). Tool + * output is folded back into the call it belongs to by `call_id`. + */ +export function mapCodexExecRolloutTranscript(lines: Iterable): CodexExecTranscript { + const entries: Array< + ProviderSubagentTranscriptEntry & { toolUses: Array<{ name: string; summary: string }> } + > = []; + const toolEntryByCallId = new Map(); + let model: string | undefined; + let cwd: string | undefined; + let prompt: string | undefined; + + const push = ( + entry: Omit & { + readonly toolUses?: ReadonlyArray<{ name: string; summary: string }>; + }, + ): number => { + entries.push({ ...entry, toolUses: [...(entry.toolUses ?? [])] }); + return entries.length - 1; + }; + + for (const line of lines) { + const record = parseJsonRecord(line); + if (!record) { + continue; + } + const payload = readRecord(record.payload); + if (!payload) { + continue; + } + const at = readText(record.timestamp) ?? undefined; + const withAt = (entry: T) => ({ ...entry, ...(at ? { at } : {}) }); + + if (record.type === "session_meta") { + cwd = readText(payload.cwd) ?? cwd; + continue; + } + if (record.type === "turn_context") { + model = readText(payload.model) ?? model; + continue; + } + if (record.type === "event_msg") { + if (payload.type === "user_message") { + const text = readText(payload.message); + if (text) { + prompt = prompt ?? text; + push(withAt({ role: "user" as const, text: capText(text, TRANSCRIPT_TEXT_MAX_CHARS) })); + } + continue; + } + if (payload.type === "agent_message") { + const text = readText(payload.message); + if (text) { + push( + withAt({ role: "assistant" as const, text: capText(text, TRANSCRIPT_TEXT_MAX_CHARS) }), + ); + } + continue; + } + if (payload.type === "agent_reasoning") { + const text = readText(payload.text); + if (text) { + push( + withAt({ role: "thinking" as const, text: capText(text, TRANSCRIPT_TEXT_MAX_CHARS) }), + ); + } + continue; + } + if (payload.type === "web_search_end") { + const query = readText(payload.query); + if (query) { + push( + withAt({ + role: "assistant" as const, + text: "", + toolUses: [{ name: "web_search", summary: query }], + }), + ); + } + continue; + } + if (payload.type === "patch_apply_end") { + const changes = readRecord(payload.changes); + const summary = changes ? Object.keys(changes).join(", ") : null; + if (summary) { + push( + withAt({ + role: "assistant" as const, + text: "", + toolUses: [{ name: "apply_patch", summary }], + }), + ); + } + continue; + } + continue; + } + if (record.type !== "response_item") { + continue; + } + + switch (payload.type) { + case "reasoning": { + const summary = Array.isArray(payload.summary) + ? payload.summary + .map((part) => { + const partRecord = readRecord(part); + return partRecord && typeof partRecord.text === "string" ? partRecord.text : ""; + }) + .filter((text) => text.length > 0) + .join("\n") + : ""; + if (summary.trim().length > 0) { + push( + withAt({ + role: "thinking" as const, + text: capText(summary, TRANSCRIPT_TEXT_MAX_CHARS), + }), + ); + } + break; + } + case "custom_tool_call": + case "function_call": { + const name = readText(payload.name) ?? "tool"; + const rawInput = + typeof payload.input === "string" + ? payload.input + : typeof payload.arguments === "string" + ? payload.arguments + : ""; + const index = push( + withAt({ + role: "assistant" as const, + text: "", + toolUses: [{ name, summary: capText(rawInput, TRANSCRIPT_OUTPUT_PREVIEW_MAX_CHARS) }], + }), + ); + const callId = readText(payload.call_id); + if (callId) { + toolEntryByCallId.set(callId, index); + } + break; + } + case "custom_tool_call_output": + case "function_call_output": { + const callId = readText(payload.call_id); + const index = callId === null ? undefined : toolEntryByCallId.get(callId); + const preview = capText( + readContentText(payload.output), + TRANSCRIPT_OUTPUT_PREVIEW_MAX_CHARS, + ); + if (index === undefined || preview.length === 0) { + break; + } + entries[index] = { ...entries[index]!, outputPreview: preview }; + break; + } + default: + break; + } + } + + return { + entries, + ...(model ? { model } : {}), + ...(cwd ? { cwd } : {}), + ...(prompt ? { prompt } : {}), + }; +} + +function readFileLines( + fileSystem: FileSystem.FileSystem, + filePath: string, +): Effect.Effect, PlatformError.PlatformError> { + return fileSystem.readFileString(filePath).pipe(Effect.map((text) => text.split("\n"))); +} + +/** + * The rollout's `session_meta` line, read without pulling in the rest of the + * file. Every candidate in a day's directory gets opened during correlation + * and a long transcript runs to megabytes, so this stops at the first newline. + * `session_meta` carries the full base instructions and is itself tens of + * kilobytes, hence the generous cap. + */ +function readFirstLine(fileSystem: FileSystem.FileSystem, filePath: string) { + return Effect.scoped( + Effect.gen(function* () { + const file = yield* fileSystem.open(filePath); + const decoder = new TextDecoder(); + let head = ""; + while (head.length < FIRST_LINE_MAX_CHARS) { + const chunk = yield* file.readAlloc(READ_CHUNK_BYTES); + if (Option.isNone(chunk) || chunk.value.length === 0) { + break; + } + head += decoder.decode(chunk.value, { stream: true }); + const newline = head.indexOf("\n"); + if (newline !== -1) { + return head.slice(0, newline); + } + } + return head; + }), + ).pipe(Effect.catch(() => Effect.succeed(""))); +} + +function dateDirectorySegments(date: Date): ReadonlyArray { + const year = String(date.getFullYear()); + const month = String(date.getMonth() + 1).padStart(2, "0"); + const day = String(date.getDate()).padStart(2, "0"); + return [year, month, day]; +} + +function realPath(fileSystem: FileSystem.FileSystem, candidate: string) { + return fileSystem.realPath(candidate).pipe(Effect.catch(() => Effect.succeed(candidate))); +} + +export interface CodexExecRolloutMatch { + readonly rolloutPath: string; + readonly sessionId: string; +} + +export interface FindCodexExecRolloutInput { + readonly sessionsRoot: string; + /** Working directory of the Claude session that launched the run. */ + readonly cwd: string; + /** Epoch millis the task started, minus whatever slack the caller wants. */ + readonly notBeforeMs: number; + /** Rollout paths already attributed to another run in this session. */ + readonly claimedPaths: ReadonlySet; +} + +/** + * Find the rollout an in-flight `codex exec` run is writing. + * + * Candidates are narrowed by date directory and mtime before any file is + * opened, then confirmed by reading only the `session_meta` line: it must be + * an exec-originated rollout whose recorded cwd resolves to the same real path + * as the Claude session's. The earliest unclaimed match wins, so two + * concurrent runs in one thread settle onto different files. + */ +export const findCodexExecRollout = Effect.fn("findCodexExecRollout")(function* ( + input: FindCodexExecRolloutInput, +): Effect.fn.Return< + CodexExecRolloutMatch | null, + PlatformError.PlatformError, + FileSystem.FileSystem | Path.Path +> { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const expectedCwd = yield* realPath(fileSystem, input.cwd); + + // A run started just before midnight writes into the previous day's + // directory, so both are searched. + const startedAt = new Date(input.notBeforeMs); + const directories = [ + path.join(input.sessionsRoot, ...dateDirectorySegments(startedAt)), + path.join( + input.sessionsRoot, + ...dateDirectorySegments(new Date(input.notBeforeMs + 86_400_000)), + ), + ]; + + const candidates: Array<{ readonly filePath: string; readonly mtimeMs: number }> = []; + for (const directory of new Set(directories)) { + const entries = yield* fileSystem + .readDirectory(directory) + .pipe(Effect.catch(() => Effect.succeed>([]))); + for (const entry of entries) { + if (!entry.startsWith("rollout-") || !entry.endsWith(".jsonl")) { + continue; + } + const filePath = path.join(directory, entry); + if (input.claimedPaths.has(filePath)) { + continue; + } + const info = yield* fileSystem.stat(filePath).pipe(Effect.catch(() => Effect.succeed(null))); + if (info === null) { + continue; + } + const modifiedAt = Option.getOrUndefined(info.mtime)?.getTime(); + if (modifiedAt !== undefined && modifiedAt < input.notBeforeMs) { + continue; + } + candidates.push({ filePath, mtimeMs: modifiedAt ?? input.notBeforeMs }); + } + } + + for (const candidate of candidates.toSorted((left, right) => { + const byTime = left.mtimeMs - right.mtimeMs; + return byTime !== 0 ? byTime : left.filePath.localeCompare(right.filePath); + })) { + const head = readCodexExecRolloutHead(yield* readFirstLine(fileSystem, candidate.filePath)); + if (!head?.cwd) { + continue; + } + // The mtime gate above cannot exclude an older run that is still writing — + // its file stays fresh for as long as it runs. The recorded start time can: + // a rollout that began before this task did belongs to some other run. + const headStartedAtMs = head.startedAt === null ? Number.NaN : Date.parse(head.startedAt); + if (!Number.isNaN(headStartedAtMs) && headStartedAtMs < input.notBeforeMs) { + continue; + } + const candidateCwd = yield* realPath(fileSystem, head.cwd); + if (candidateCwd !== expectedCwd) { + continue; + } + return { rolloutPath: candidate.filePath, sessionId: head.sessionId }; + } + return null; +}); + +/** + * Resolve a rollout from its session id alone, so a transcript stays readable + * after the server restarts and nothing remembers where the file was. + */ +export const locateCodexExecRolloutBySessionId = Effect.fn("locateCodexExecRolloutBySessionId")( + function* (input: { + readonly sessionsRoot: string; + readonly sessionId: string; + }): Effect.fn.Return< + string | null, + PlatformError.PlatformError, + FileSystem.FileSystem | Path.Path + > { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const suffix = `-${input.sessionId}.jsonl`; + const readSorted = (directory: string) => + fileSystem + .readDirectory(directory) + .pipe(Effect.catch(() => Effect.succeed>([]))) + .pipe(Effect.map((entries) => entries.toSorted().toReversed())); + + let scanned = 0; + for (const year of yield* readSorted(input.sessionsRoot)) { + for (const month of yield* readSorted(path.join(input.sessionsRoot, year))) { + for (const day of yield* readSorted(path.join(input.sessionsRoot, year, month))) { + if (scanned >= SESSION_DATE_DIRECTORY_SCAN_LIMIT) { + return null; + } + scanned += 1; + const directory = path.join(input.sessionsRoot, year, month, day); + for (const entry of yield* readSorted(directory)) { + if (entry.startsWith("rollout-") && entry.endsWith(suffix)) { + return path.join(directory, entry); + } + } + } + } + } + return null; + }, +); + +/** Read a rollout's lines, or an empty list when it cannot be read. */ +export const readCodexExecRolloutLines = Effect.fn("readCodexExecRolloutLines")(function* ( + rolloutPath: string, +): Effect.fn.Return, never, FileSystem.FileSystem> { + const fileSystem = yield* FileSystem.FileSystem; + return yield* readFileLines(fileSystem, rolloutPath).pipe( + Effect.catch(() => Effect.succeed>([])), + ); +}); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 2b8d86930..359a218b1 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -7393,3 +7393,275 @@ describe("ClaudeAdapterLive", () => { ); }); }); + +/** + * A Claude thread that shells out to `codex exec` in the background is running + * a second agent, and the panel should say so. These cover the promotion arc: + * the row appears on the task edge, gains the rollout's identity and settings + * once codex has written them, and settles with the task. + */ +describe("ClaudeAdapterLive codex exec promotion", () => { + const CODEX_SESSION_ID = "019fea48-4154-7982-876b-43e4c551eb65"; + + function makeCodexHome(): { readonly codexHome: string; readonly cleanup: () => void } { + const codexHome = mkdtempSync(path.join(os.tmpdir(), "claude-adapter-codex-home-")); + return { codexHome, cleanup: () => rmSync(codexHome, { recursive: true, force: true }) }; + } + + function seedRollout(codexHome: string, cwd: string): void { + const now = new Date(); + const directory = path.join( + codexHome, + "sessions", + String(now.getFullYear()), + String(now.getMonth() + 1).padStart(2, "0"), + String(now.getDate()).padStart(2, "0"), + ); + mkdirSync(directory, { recursive: true }); + writeFileSync( + path.join(directory, `rollout-now-${CODEX_SESSION_ID}.jsonl`), + [ + JSON.stringify({ + // A just-started run's recorded start time; the correlation gate + // rejects rollouts that began before the task did. + timestamp: now.toISOString(), + type: "session_meta", + payload: { + session_id: CODEX_SESSION_ID, + cwd, + originator: "codex_exec", + source: "exec", + }, + }), + JSON.stringify({ + type: "turn_context", + payload: { model: "gpt-5.6-sol", effort: "medium" }, + }), + JSON.stringify({ + type: "event_msg", + payload: { type: "task_complete", last_agent_message: "Reviewed the adapter." }, + }), + "", + ].join("\n"), + ); + } + + function emitBashToolUse( + harness: ReturnType, + input: { readonly toolUseId: string; readonly command: string }, + ): void { + harness.query.emit({ + type: "stream_event", + session_id: "sdk-session-codex-exec", + uuid: `${input.toolUseId}-start`, + parent_tool_use_id: null, + event: { + type: "content_block_start", + index: 0, + content_block: { + type: "tool_use", + id: input.toolUseId, + name: "Bash", + input: { command: input.command, run_in_background: true }, + }, + }, + } as unknown as SDKMessage); + } + + function emitMarkerTask(harness: ReturnType): void { + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-marker", + description: "Marker task", + session_id: "sdk-session-codex-exec", + uuid: "task-marker-started", + } as unknown as SDKMessage); + } + + const subagentMetadata = (events: ReadonlyArray) => + events.flatMap((event) => (event.type === "subagent.metadata.updated" ? [event.payload] : [])); + + const collectUntilMarker = (adapter: ClaudeAdapterShape) => + adapter.streamEvents.pipe( + Stream.takeUntil( + (event) => event.type === "task.started" && event.payload.taskId === "task-marker", + ), + Stream.runCollect, + Effect.forkChild, + ); + + function makeCodexExecHarness(): { + readonly harness: ReturnType; + readonly codexHome: string; + readonly cwd: string; + readonly cleanup: () => void; + } { + const { codexHome, cleanup } = makeCodexHome(); + const cwd = mkdtempSync(path.join(os.tmpdir(), "claude-adapter-codex-cwd-")); + return { + harness: makeHarness({ + cwd, + environment: { ...missingClaudeConfigEnvironment(), CODEX_HOME: codexHome }, + }), + codexHome, + cwd, + cleanup: () => { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + }, + }; + } + + it.effect("promotes a background codex exec run and settles it with the task", () => { + const fixture = makeCodexExecHarness(); + seedRollout(fixture.codexHome, fixture.cwd); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEventsFiber = yield* collectUntilMarker(adapter); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + cwd: fixture.cwd, + runtimeMode: "full-access", + }); + + emitBashToolUse(fixture.harness, { + toolUseId: "tool-codex-exec", + command: `codex exec -m gpt-5.6-sol "Review the adapter"`, + }); + fixture.harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-codex-exec", + description: "Review the adapter", + task_type: "local_bash", + tool_use_id: "tool-codex-exec", + session_id: "sdk-session-codex-exec", + uuid: "task-codex-exec-started", + } as unknown as SDKMessage); + fixture.harness.query.emit({ + type: "system", + subtype: "task_notification", + task_id: "task-codex-exec", + tool_use_id: "tool-codex-exec", + status: "completed", + summary: "codex exec finished", + session_id: "sdk-session-codex-exec", + uuid: "task-codex-exec-completed", + } as unknown as SDKMessage); + emitMarkerTask(fixture.harness); + + const metadata = subagentMetadata(Array.from(yield* Fiber.join(runtimeEventsFiber))); + // Row first, identity second, settlement last. + assert.deepEqual(metadata[0], { + callId: "tool-codex-exec", + status: "running", + agentRole: "codex", + objective: "Review the adapter", + model: "gpt-5.6-sol", + modelSource: "explicit", + }); + assert.deepEqual(metadata[1], { + callId: "tool-codex-exec", + agentThreadId: `codex-exec:${CODEX_SESSION_ID}`, + transcriptAgentId: `codex-exec:${CODEX_SESSION_ID}`, + status: "running", + model: "gpt-5.6-sol", + resolvedModel: "gpt-5.6-sol", + modelSource: "provider", + reasoningEffort: "medium", + reasoningEffortSource: "provider", + }); + assert.equal(metadata.length, 3); + assert.equal(metadata[2]?.callId, "tool-codex-exec"); + assert.equal(metadata[2]?.status, "completed"); + assert.equal(metadata[2]?.resultBody, "Reviewed the adapter."); + + // The rollout the row was linked to also serves its transcript. + const transcript = yield* adapter.readSubagentTranscript!(THREAD_ID, { + threadId: THREAD_ID, + agentId: `codex-exec:${CODEX_SESSION_ID}`, + }); + assert.equal(transcript.agent?.model, "gpt-5.6-sol"); + assert.equal(transcript.agent?.agentType, "codex"); + }).pipe(Effect.provide(fixture.harness.layer), Effect.ensuring(Effect.sync(fixture.cleanup))); + }); + + it.effect("leaves ordinary background commands alone", () => { + const fixture = makeCodexExecHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEventsFiber = yield* collectUntilMarker(adapter); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + cwd: fixture.cwd, + runtimeMode: "full-access", + }); + + emitBashToolUse(fixture.harness, { + toolUseId: "tool-dev-server", + command: `echo "codex exec go" && pnpm dev`, + }); + fixture.harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-dev-server", + description: "Run the dev server", + task_type: "local_bash", + tool_use_id: "tool-dev-server", + session_id: "sdk-session-codex-exec", + uuid: "task-dev-server-started", + } as unknown as SDKMessage); + emitMarkerTask(fixture.harness); + + assert.deepEqual(subagentMetadata(Array.from(yield* Fiber.join(runtimeEventsFiber))), []); + }).pipe(Effect.provide(fixture.harness.layer), Effect.ensuring(Effect.sync(fixture.cleanup))); + }); + + it.effect("settles a promoted row even when no rollout is ever found", () => { + const fixture = makeCodexExecHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEventsFiber = yield* collectUntilMarker(adapter); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + cwd: fixture.cwd, + runtimeMode: "full-access", + }); + + emitBashToolUse(fixture.harness, { + toolUseId: "tool-codex-exec", + command: `codex exec "Review the adapter"`, + }); + fixture.harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-codex-exec", + description: "Review the adapter", + task_type: "local_bash", + tool_use_id: "tool-codex-exec", + session_id: "sdk-session-codex-exec", + uuid: "task-codex-exec-started", + } as unknown as SDKMessage); + fixture.harness.query.emit({ + type: "system", + subtype: "task_updated", + task_id: "task-codex-exec", + patch: { status: "failed", error: "codex exited 1" }, + session_id: "sdk-session-codex-exec", + uuid: "task-codex-exec-failed", + } as unknown as SDKMessage); + emitMarkerTask(fixture.harness); + + const metadata = subagentMetadata(Array.from(yield* Fiber.join(runtimeEventsFiber))); + assert.equal(metadata.length, 2); + assert.deepEqual(metadata[1], { callId: "tool-codex-exec", status: "failed" }); + }).pipe(Effect.provide(fixture.harness.layer), Effect.ensuring(Effect.sync(fixture.cleanup))); + }); +}); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 627bb3060..8710f0838 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -49,6 +49,7 @@ import { type ProviderUserInputAnswers, type RuntimeSessionExitKind, type ServerProviderModel, + type SubagentMetadataUpdatedPayload, type RuntimeContentStreamKind, RuntimeItemId, RuntimeRequestId, @@ -65,6 +66,7 @@ import { import { renderThreadContextSeed, withContextSeedPreamble } from "@threadlines/shared/contextSeed"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; import * as Deferred from "effect/Deferred"; import { randomUUIDv4 } from "@threadlines/shared/uuid"; import * as Effect from "effect/Effect"; @@ -91,6 +93,19 @@ import { locateClaudeSubagentTranscript, readClaudeSubagentTranscriptEntries, } from "../Drivers/ClaudeSubagentTranscripts.ts"; +import { + codexExecAgentId, + codexExecSessionsRoot, + findCodexExecRollout, + locateCodexExecRolloutBySessionId, + mapCodexExecRolloutTranscript, + matchCodexExecCommand, + parseCodexExecAgentId, + readCodexExecFinalMessage, + readCodexExecRolloutLines, + readCodexExecTurnContext, + type CodexExecInvocation, +} from "../Drivers/CodexExecRollouts.ts"; import { addProviderAuthHint, isProviderAuthErrorMessage } from "../providerAuthHints.ts"; import { claudeModelSupportsAutoRuntimeMode, @@ -263,6 +278,17 @@ interface ToolInFlight { readonly lastEmittedInputFingerprint?: string; } +/** A background `codex exec` run promoted to a subagent row, from the moment + * its task starts until its rollout is found and its task settles. */ +interface CodexExecRun { + readonly taskId: string; + readonly toolUseId: string; + readonly startedAtMs: number; + readonly cwd: string; + rolloutPath: string | null; + settled: boolean; +} + type ClaudeTaskStatus = "pending" | "running" | "completed" | "failed" | "killed" | "paused"; interface ClaudeTaskSnapshot { @@ -343,6 +369,15 @@ interface ClaudeSessionContext { * Consumed when the matching tool_result is emitted. */ readonly fileChangeStatsByToolUseId: Map; readonly tasks: Map; + /** Shell commands of in-flight Bash tool calls, keyed by tool_use id. A + * `task_started` names the tool that launched it but not what it runs, so + * the command has to be remembered from the tool call itself. Bounded FIFO. */ + readonly bashCommandsByToolUseId: Map; + /** Background `codex exec` runs promoted to subagent rows, keyed by task id. */ + readonly codexExecRuns: Map; + /** Rollout files already attributed to a run in this session, so two + * concurrent `codex exec` calls never settle onto the same file. */ + readonly claimedCodexExecRolloutPaths: Set; /** Task ids whose lifecycle start edge was already emitted in this Claude * process. Kept separate from task metadata so reordered progress cannot * suppress the real start edge. */ @@ -1950,6 +1985,54 @@ const SUBAGENT_AGENT_ID_PATTERN = /^[A-Za-z0-9_-]+$/; * past any real session's subagent count while bounding a runaway. */ const SUBAGENT_SPAWN_ANCESTRY_MAX_ENTRIES = 1_024; +/** Role recorded on a promoted `codex exec` row. The agents panel renders the + * role as the row's name, so this is what the row reads as. */ +const CODEX_EXEC_SUBAGENT_ROLE = "codex"; +/** Shell commands remembered per session for task correlation; oldest evict + * first. A turn issues a handful of Bash calls, so this is generous. */ +const CODEX_EXEC_COMMAND_MAX_ENTRIES = 32; +/** Gaps between rollout lookups, i.e. 1s/3s/7s/15s/30s after the task starts. + * Codex writes `session_meta` on startup, so the first attempt usually wins; + * the tail covers a cold binary or a loaded machine. */ +const CODEX_EXEC_ROLLOUT_POLL_DELAYS_MS = [1_000, 2_000, 4_000, 8_000, 15_000] as const; +/** How far before the task's start a rollout may have been last written and + * still be this run's. Covers clock skew between the task edge and the file. */ +const CODEX_EXEC_ROLLOUT_START_SLACK_MS = 10_000; + +/** + * Runs a `codex exec` promotion step for effect, swallowing every outcome. + * + * Promotion is a presentation nicety layered onto ordinary task handling: a + * missing rollout directory, an unreadable file or an unexpected record shape + * must leave the task stream exactly as it was, with the run shown the old way. + */ +function guardCodexExecFailure( + label: string, + effect: Effect.Effect, +): Effect.Effect { + return effect.pipe( + Effect.asVoid, + Effect.catchCause((cause) => Effect.logDebug(label, { cause })), + ); +} + +/** Remembers a Bash-shaped tool call's command line so a later `task_started` + * that names only the tool_use id can tell what was launched. */ +function rememberShellCommand(context: ClaudeSessionContext, tool: ToolInFlight): void { + const command = tool.input.command; + if (typeof command !== "string" || command.trim().length === 0) { + return; + } + context.bashCommandsByToolUseId.set(tool.itemId, command); + while (context.bashCommandsByToolUseId.size > CODEX_EXEC_COMMAND_MAX_ENTRIES) { + const oldest = context.bashCommandsByToolUseId.keys().next(); + if (oldest.done) { + return; + } + context.bashCommandsByToolUseId.delete(oldest.value); + } +} + function capTranscriptText(value: string, maxChars: number): string { const trimmed = value.trim(); return trimmed.length > maxChars ? trimmed.slice(0, maxChars) : trimmed; @@ -3439,6 +3522,9 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ? toolInputFingerprint(parsedInput) : undefined; context.inFlightTools.set(event.index, nextTool); + if (parsedInput) { + rememberShellCommand(context, nextTool); + } if ( !parsedInput || @@ -3569,6 +3655,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ...(inputFingerprint ? { lastEmittedInputFingerprint: inputFingerprint } : {}), }; context.inFlightTools.set(index, tool); + rememberShellCommand(context, tool); const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ @@ -3712,6 +3799,13 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( payload: message, }, }); + // A `codex exec` launched into the background is a second agent, not a + // shell command; promoting it here gives it the same row every other + // subagent gets. Never allowed to disturb the task stream above. + yield* guardCodexExecFailure( + "claude.codex-exec.promote-failed", + promoteCodexExecTask(context, task), + ); return true; }); @@ -3759,9 +3853,223 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( payload: message, }, }); + // Settles alongside the single task completion, so a promoted row leaves + // "running" exactly when the task does — including when its rollout was + // never found and the row has nothing else to settle on. + yield* guardCodexExecFailure( + "claude.codex-exec.settle-failed", + settleCodexExecRun(context, { taskId: task.taskId, status: task.status }), + ); return true; }); + const emitSubagentMetadata = ( + context: ClaudeSessionContext, + payload: SubagentMetadataUpdatedPayload, + ) => + Effect.gen(function* () { + const stamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + type: "subagent.metadata.updated", + eventId: stamp.eventId, + provider: PROVIDER, + createdAt: stamp.createdAt, + threadId: context.session.threadId, + ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), + payload, + providerRefs: nativeProviderRefs(context), + }); + }); + + /** + * Attribute a started `codex exec` run to the rollout file it is writing and + * report what that file says about the run. + * + * Returns false while the file has not appeared yet, which is the normal + * state for the first second or two of a run. + */ + const resolveCodexExecRollout = Effect.fn("resolveCodexExecRollout")(function* ( + context: ClaudeSessionContext, + run: CodexExecRun, + ) { + const match = yield* findCodexExecRollout({ + sessionsRoot: codexExecSessionsRoot(claudeEnvironment, path), + cwd: run.cwd, + notBeforeMs: run.startedAtMs - CODEX_EXEC_ROLLOUT_START_SLACK_MS, + claimedPaths: context.claimedCodexExecRolloutPaths, + }).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + Effect.catch(() => Effect.succeed(null)), + ); + if (!match) { + return false; + } + context.claimedCodexExecRolloutPaths.add(match.rolloutPath); + run.rolloutPath = match.rolloutPath; + + const lines = yield* readCodexExecRolloutLines(match.rolloutPath).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + ); + const turnContext = readCodexExecTurnContext(lines); + const agentId = codexExecAgentId(match.sessionId); + yield* emitSubagentMetadata(context, { + callId: run.toolUseId, + agentThreadId: agentId, + transcriptAgentId: agentId, + status: "running", + ...(turnContext?.model + ? { + model: turnContext.model, + resolvedModel: turnContext.model, + modelSource: "provider" as const, + } + : {}), + ...(turnContext?.reasoningEffort + ? { + reasoningEffort: turnContext.reasoningEffort, + reasoningEffortSource: "provider" as const, + } + : {}), + }); + return true; + }); + + /** Backs off while the rollout has not appeared. Forked as a child of the + * session's stream fiber, so it dies when the session does. */ + const pollCodexExecRollout = Effect.fn("pollCodexExecRollout")(function* ( + context: ClaudeSessionContext, + run: CodexExecRun, + ) { + for (const delay of CODEX_EXEC_ROLLOUT_POLL_DELAYS_MS) { + yield* Effect.sleep(Duration.millis(delay)); + if (run.settled || run.rolloutPath !== null || context.stopped) { + return; + } + if (yield* resolveCodexExecRollout(context, run)) { + return; + } + } + }); + + /** + * Promote a background `codex exec` command into a subagent row. + * + * The row is emitted before anything is known beyond the command line, so it + * appears the moment the task starts; the rollout poll fills in the model, + * the effort and the transcript id as soon as codex has written them. + */ + const startCodexExecRun = Effect.fn("startCodexExecRun")(function* ( + context: ClaudeSessionContext, + input: { + readonly taskId: string; + readonly toolUseId: string; + readonly description?: string; + readonly invocation: CodexExecInvocation; + }, + ) { + const cwd = context.session.cwd; + if (!cwd || context.codexExecRuns.has(input.taskId)) { + return; + } + const run: CodexExecRun = { + taskId: input.taskId, + toolUseId: input.toolUseId, + startedAtMs: Date.now(), + cwd, + rolloutPath: null, + settled: false, + }; + context.codexExecRuns.set(input.taskId, run); + + const objective = input.description ?? input.invocation.prompt; + const { model, reasoningEffort } = input.invocation; + yield* emitSubagentMetadata(context, { + callId: input.toolUseId, + status: "running", + agentRole: CODEX_EXEC_SUBAGENT_ROLE, + ...(objective ? { objective } : {}), + ...(model ? { model, modelSource: "explicit" as const } : {}), + ...(reasoningEffort ? { reasoningEffort, reasoningEffortSource: "explicit" as const } : {}), + }); + yield* Effect.forkChild(pollCodexExecRollout(context, run)); + }); + + /** Settles a promoted run's row when its task ends, with codex's own last + * message as the result when the rollout can be read. */ + const settleCodexExecRun = Effect.fn("settleCodexExecRun")(function* ( + context: ClaudeSessionContext, + input: { + readonly taskId: string; + readonly status: "completed" | "failed" | "stopped"; + }, + ) { + const run = context.codexExecRuns.get(input.taskId); + if (!run || run.settled) { + return; + } + run.settled = true; + context.codexExecRuns.delete(input.taskId); + // Codex writes its final message before the shell exits, so the file that + // never showed up during the poll is usually there by now. + if (run.rolloutPath === null) { + yield* resolveCodexExecRollout(context, run); + } + const resultBody = + run.rolloutPath === null + ? null + : readCodexExecFinalMessage( + yield* readCodexExecRolloutLines(run.rolloutPath).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + ), + ); + const createdAt = yield* nowIso; + yield* emitSubagentMetadata(context, { + callId: run.toolUseId, + status: + input.status === "failed" + ? "failed" + : input.status === "stopped" + ? "interrupted" + : "completed", + ...(resultBody ? { resultBody, resultCreatedAt: createdAt } : {}), + }); + }); + + /** + * Whether a started background task is a `codex exec` run, and if so, the + * row it should be promoted to. + * + * Every failure here degrades to today's behavior — a plain background run + * row — rather than disturbing task handling, which is why the whole path is + * guarded rather than allowed to fail the message. + */ + const promoteCodexExecTask = Effect.fn("promoteCodexExecTask")(function* ( + context: ClaudeSessionContext, + task: { + readonly taskId: string; + readonly description?: string; + readonly toolUseId?: string; + readonly subagentType?: string; + readonly taskType?: string; + }, + ) { + if (task.taskType !== "local_bash" || task.subagentType !== undefined || !task.toolUseId) { + return; + } + const command = context.bashCommandsByToolUseId.get(task.toolUseId); + const invocation = command ? matchCodexExecCommand(command) : null; + if (!invocation) { + return; + } + yield* startCodexExecRun(context, { + taskId: task.taskId, + toolUseId: task.toolUseId, + ...(task.description ? { description: task.description } : {}), + invocation, + }); + }); + // A background subagent's Task tool_result is only a launch acknowledgment; // the agent's real final message arrives later inside a // user message. Replay it as a completion of the originating tool item so @@ -5747,6 +6055,9 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( subagentSpawnAncestry: new Map(), fileChangeStatsByToolUseId, tasks: new Map(), + bashCommandsByToolUseId: new Map(), + codexExecRuns: new Map(), + claimedCodexExecRolloutPaths: new Set(), startedTaskIds: new Set(), backgroundTaskSnapshotObserved: false, planTracker: new Map(), @@ -6093,6 +6404,47 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, ); + const readCodexExecTranscript = Effect.fn("readCodexExecTranscript")(function* ( + sessionId: string, + input: { + readonly limit?: number | undefined; + readonly offset?: number | undefined; + readonly fromEnd?: boolean | undefined; + }, + requestError: (detail: string) => ProviderAdapterRequestError, + ): Effect.fn.Return { + const rolloutPath = yield* locateCodexExecRolloutBySessionId({ + sessionsRoot: codexExecSessionsRoot(claudeEnvironment, path), + sessionId, + }).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + Effect.catch(() => Effect.succeed(null)), + ); + if (rolloutPath === null) { + return yield* requestError("No transcript found for this codex run yet."); + } + const transcript = mapCodexExecRolloutTranscript( + yield* readCodexExecRolloutLines(rolloutPath).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + ), + ); + const description = transcript.prompt ?? transcript.cwd; + return { + ...pageClaudeSubagentTranscriptEntries(transcript.entries, { + ...(input.limit !== undefined ? { limit: input.limit } : {}), + ...(input.offset !== undefined ? { offset: input.offset } : {}), + ...(input.fromEnd !== undefined ? { fromEnd: input.fromEnd } : {}), + }), + agent: { + id: codexExecAgentId(sessionId), + agentType: CODEX_EXEC_SUBAGENT_ROLE, + ...(description ? { description } : {}), + ...(transcript.model ? { model: transcript.model } : {}), + }, + }; + }); + const readSubagentTranscript: NonNullable = Effect.fn("readSubagentTranscript")(function* (threadId, input) { const requestError = (detail: string) => @@ -6101,6 +6453,13 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( method: "readSubagentTranscript", detail, }); + // A promoted `codex exec` run's transcript is the rollout codex wrote, + // not a Claude subagent file. It resolves from the id alone, so it keeps + // working across a server restart, when nothing remembers the run. + const codexExecSessionId = parseCodexExecAgentId(input.agentId); + if (codexExecSessionId !== null) { + return yield* readCodexExecTranscript(codexExecSessionId, input, requestError); + } if (!SUBAGENT_AGENT_ID_PATTERN.test(input.agentId)) { return yield* requestError(`Invalid subagent id '${input.agentId}'.`); } diff --git a/apps/web/src/agentsPanelStore.ts b/apps/web/src/agentsPanelStore.ts index a76981007..8332d2a17 100644 --- a/apps/web/src/agentsPanelStore.ts +++ b/apps/web/src/agentsPanelStore.ts @@ -23,6 +23,9 @@ export interface AgentsPanelSource { threadId: ThreadId; subagents: ReadonlyArray; backgroundRuns: ReadonlyArray; + /** Runs the panel lists as subagents rather than as runs, keyed by the tool + * call that launched them. They carry the stop handle those agent rows use. */ + subagentRuns: ReadonlyMap; /** Every agent the thread has run, live or long finished. Published alongside * the live items so the panel and the conversation's receipts resolve the * same set of agents. */ diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index 00771d498..1f4acf77c 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -1398,6 +1398,50 @@ describe("deriveProviderBackgroundRuns", () => { expect(detectionSeeds.commandHints[0]).toContain("scripts/dev-runner.ts"); expect(detectionSeeds.commandHints[0]).toContain("threadlines-activity-preview-280"); }); + + it("moves a task the thread already tracks as a subagent off the run list", () => { + const activities = [ + taskActivity( + "task.started", + { + taskId: "task-codex-exec", + taskType: "local_bash", + description: "Review the adapter", + toolUseId: "tool-codex-exec", + detail: "codex exec running at http://localhost:5959", + }, + 1, + ), + taskActivity("task.progress", { taskId: "task-codex-exec", detail: "Still reviewing" }, 2), + ]; + + const promoted = deriveProviderBackgroundRuns({ + activities, + messages: [], + pendingBackgroundTaskCount: 1, + activeSubagentCount: 1, + subagentSpawnCallIds: new Set(["tool-codex-exec"]), + }); + + // The row is gone from the run list, but its stop handles survive: the + // agent row it moved to has no other way to stop the process. + expect(promoted.runs).toEqual([]); + expect(promoted.promotedSubagentRuns.get("tool-codex-exec")).toMatchObject({ + id: "provider:task-codex-exec", + label: "Still reviewing", + urls: ["http://localhost:5959"], + }); + expect(promoted.detectionSeeds.urls).toEqual(["http://localhost:5959"]); + + // Without the subagent record it is an ordinary background run. + const plain = deriveProviderBackgroundRuns({ + activities, + messages: [], + pendingBackgroundTaskCount: 1, + }); + expect(plain.runs).toHaveLength(1); + expect(plain.promotedSubagentRuns.size).toBe(0); + }); }); describe("backgroundRunCommandsMatch", () => { diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index 493de6c79..2796e9ddb 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -276,6 +276,14 @@ export interface BackgroundRunDetectionSeeds { export interface ProviderBackgroundRunsResult { /** Provider-tracked background runs that are safe to render in the header. */ runs: ProviderBackgroundRunState[]; + /** + * Runs that are already shown as subagent rows, keyed by the tool call that + * launched them. They are kept out of `runs` so the same work is never + * listed twice, but the agents panel still needs them: the subagent row has + * no stop handle of its own, and this is where the command hints and PIDs + * that resolve one live. + */ + promotedSubagentRuns: ReadonlyMap; /** * URLs, PIDs, and command hints used to seed server-side process detection. * Includes unconfirmed prose mentions: those feed detection so a real @@ -987,10 +995,14 @@ export function deriveProviderBackgroundRuns(input: { pendingBackgroundTaskCount: number; activeSubagentCount?: number | undefined; activeCommandTurnId?: TurnId | null | undefined; + /** Tool calls the thread already tracks as subagents, live or in history. A + * task launched by one of these is that agent, not a separate run. */ + subagentSpawnCallIds?: ReadonlySet | undefined; }): ProviderBackgroundRunsResult { const activeRunsByTaskId = new Map(); const activeCommandDetectionRunsByKey = new Map(); const suppressedSubagentTaskIds = new Set(); + const promotedToolUseIdByTaskId = new Map(); const activeTaskScopesByTaskId = new Map< string, { @@ -1024,6 +1036,7 @@ export function deriveProviderBackgroundRuns(input: { if (activity.kind === "task.completed") { activeRunsByTaskId.delete(taskId); suppressedSubagentTaskIds.delete(taskId); + promotedToolUseIdByTaskId.delete(taskId); activeTaskScopesByTaskId.delete(taskId); continue; } @@ -1059,6 +1072,14 @@ export function deriveProviderBackgroundRuns(input: { } activeTaskScopesByTaskId.set(taskId, taskScope); + // A task launched by a tool call the thread already tracks as a subagent + // is that agent's own work. Sticky per task, because a progress activity + // may omit the tool_use id the start edge carried. + const promotedToolUseId = asBackgroundRunString(payload?.toolUseId); + if (promotedToolUseId && input.subagentSpawnCallIds?.has(promotedToolUseId)) { + promotedToolUseIdByTaskId.set(taskId, promotedToolUseId); + } + const previous = activeRunsByTaskId.get(taskId); const description = asBackgroundRunString(payload?.description); const detail = @@ -1170,7 +1191,21 @@ export function deriveProviderBackgroundRuns(input: { }, ]; } - const taskBackedRuns = runs; + // Split out the runs the agents panel is already showing as subagent rows. + // They stay in the detection seeds below — the subagent row's stop arm goes + // through the same pid/command resolution a run row's does. + const promotedSubagentRuns = new Map(); + const promotedRunIdsByToolUseId = new Map( + [...promotedToolUseIdByTaskId].map(([taskId, toolUseId]) => [`provider:${taskId}`, toolUseId]), + ); + const taskBackedRuns = runs.filter((run) => { + const toolUseId = promotedRunIdsByToolUseId.get(run.id); + if (toolUseId === undefined) { + return true; + } + promotedSubagentRuns.set(toolUseId, run); + return false; + }); const commandDetectionRuns = [...activeCommandDetectionRunsByKey.values()].filter( (commandRun) => !taskBackedRuns.some( @@ -1181,7 +1216,7 @@ export function deriveProviderBackgroundRuns(input: { ); runs = taskBackedRuns; const hiddenSubagentTaskCount = Math.max( - suppressedSubagentTaskIds.size, + suppressedSubagentTaskIds.size + promotedSubagentRuns.size, input.activeSubagentCount ?? 0, ); const missingProviderCount = Math.max( @@ -1211,12 +1246,13 @@ export function deriveProviderBackgroundRuns(input: { // confirmed listener is promoted to a stoppable `detected` run elsewhere. We // deliberately never render the mention itself, because an unverified // localhost reference in prose is not a live background process. + const seededRuns = [...runs, ...promotedSubagentRuns.values()]; const knownUrls = new Set([ - ...runs.flatMap((run) => run.urls), + ...seededRuns.flatMap((run) => run.urls), ...commandDetectionRuns.flatMap((run) => run.urls), ]); const knownPids = new Set([ - ...runs.flatMap((run) => run.pids), + ...seededRuns.flatMap((run) => run.pids), ...commandDetectionRuns.flatMap((run) => run.pids), ]); const previewMessage = input.messages @@ -1244,15 +1280,15 @@ export function deriveProviderBackgroundRuns(input: { pids: [...new Set([...knownPids, ...mentionedPids])], commandHints: [ ...new Set([ - ...runs.flatMap((run) => run.commandHints), + ...seededRuns.flatMap((run) => run.commandHints), ...commandDetectionRuns.flatMap((run) => run.commandHints), - ...(runs.length > 0 ? commandActivityHints : []), + ...(seededRuns.length > 0 ? commandActivityHints : []), ...mentionedCommandHints, ]), ], }; - return { runs, detectionSeeds }; + return { runs, promotedSubagentRuns, detectionSeeds }; } function isSubagentProviderTask(input: { diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 240817682..77282fc07 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -250,6 +250,7 @@ import { deriveProviderSendPreflight, type ProviderSendPreflightPrompt, filterUnresolvedProviderBackgroundRuns, + type ProviderBackgroundRunState, hasServerAcknowledgedLocalDispatch, isRetryableThreadError, isScrollMetricsAtEnd, @@ -344,6 +345,33 @@ const EMPTY_PROVIDERS: ServerProvider[] = []; const EMPTY_PROVIDER_SKILLS: ServerProvider["skills"] = []; const EMPTY_PENDING_USER_INPUT_ANSWERS: Record = {}; const EMPTY_SUBAGENT_ITEMS: ReadonlyArray = []; +/** A provider-tracked run as the run list renders it. The provider never gives + * us a pid or a terminal for one, so stopping it goes through the command and + * URL hints — which is also what decides whether it can be stopped at all. */ +function toThreadBackgroundRunItem( + run: Run, +): Run & { + terminalId: null; + pid: null; + port: null; + elapsed: null; + canStop: boolean; + cwd: null; +} { + return { + ...run, + terminalId: null, + pid: null, + port: null, + elapsed: null, + canStop: + run.command !== null || + run.commandHints.length > 0 || + run.urls.length > 0 || + run.pids.length > 0, + cwd: null, + }; +} const CODEX_PROVIDER_DRIVER = ProviderDriverKind.make("codex"); const CLAUDE_PROVIDER_DRIVER = ProviderDriverKind.make("claudeAgent"); const LAYOUT_STICK_TO_BOTTOM_FRAME_COUNT = 4; @@ -2956,6 +2984,19 @@ export default function ChatView(props: ChatViewProps) { }, [performCloseTerminal], ); + // A tool call the thread already tracks as an agent is not a separate + // background run. History is consulted alongside the live turn so a promoted + // agent stays promoted after its spawning turn settles. + const subagentSpawnCallIds = useMemo(() => { + const ids = new Set(); + for (const item of subagentProgress?.items ?? []) { + if (item.spawnCallId) ids.add(item.spawnCallId); + } + for (const entry of subagentHistory) { + if (entry.item.spawnCallId) ids.add(entry.item.spawnCallId); + } + return ids; + }, [subagentHistory, subagentProgress?.items]); const providerBackgroundSnapshot = useMemo( () => deriveProviderBackgroundRuns({ @@ -2966,6 +3007,7 @@ export default function ChatView(props: ChatViewProps) { activeCommandTurnId: activeTurnInProgress ? (activeThread?.session?.activeTurnId ?? activeLatestTurn?.turnId ?? null) : null, + subagentSpawnCallIds, }), [ activeLatestTurn?.turnId, @@ -2973,25 +3015,23 @@ export default function ChatView(props: ChatViewProps) { activeThread?.session?.activeTurnId, activeTurnInProgress, subagentProgress?.activeCount, + subagentSpawnCallIds, threadActivities, timelineMessages, ], ); const providerBackgroundRuns = useMemo( + () => providerBackgroundSnapshot.runs.map(toThreadBackgroundRunItem), + [providerBackgroundSnapshot], + ); + const promotedSubagentRuns = useMemo( () => - providerBackgroundSnapshot.runs.map((run) => ({ - ...run, - terminalId: null, - pid: null, - port: null, - elapsed: null, - canStop: - run.command !== null || - run.commandHints.length > 0 || - run.urls.length > 0 || - run.pids.length > 0, - cwd: null, - })), + new Map( + [...providerBackgroundSnapshot.promotedSubagentRuns].map(([spawnCallId, run]) => [ + spawnCallId, + toThreadBackgroundRunItem(run), + ]), + ), [providerBackgroundSnapshot], ); const backgroundRunDetectionUrls = useMemo( @@ -3308,6 +3348,7 @@ export default function ChatView(props: ChatViewProps) { threadId: activeThreadId, subagents: subagentProgress?.items ?? EMPTY_SUBAGENT_ITEMS, backgroundRuns, + subagentRuns: promotedSubagentRuns, history: subagentHistory, workEntries: workLogEntries, providerLabel: activeProviderDriver, @@ -3323,6 +3364,7 @@ export default function ChatView(props: ChatViewProps) { backgroundRuns, environmentId, gitCwd, + promotedSubagentRuns, stopBackgroundRun, subagentHistory, subagentProgress?.items, diff --git a/apps/web/src/components/chat/AgentsPanel.tsx b/apps/web/src/components/chat/AgentsPanel.tsx index 2466bbf68..dd9f6aa2c 100644 --- a/apps/web/src/components/chat/AgentsPanel.tsx +++ b/apps/web/src/components/chat/AgentsPanel.tsx @@ -32,6 +32,10 @@ export interface AgentsPanelProps { threadId: ThreadId; subagents: ReadonlyArray; backgroundRuns: ReadonlyArray; + /** Background runs already listed above as subagents, keyed by the tool call + * that launched them. They are not rendered as rows; they only give the + * matching agent row a stop handle. */ + subagentRuns?: ReadonlyMap | undefined; /** Every agent the thread has run, from its durable activity projection. Keeps * the panel populated (and receipts resolvable) after the turn ends. */ history?: ReadonlyArray | undefined; @@ -143,7 +147,9 @@ function BranchRow({ ? `Open ${branch.name} transcript` : `${branch.terminalVisible ? "Close" : "Open"} ${branch.name} terminal`; - const canStop = branch.kind === "run" && branch.run.canStop; + // An agent the provider launched as a background shell command borrows the + // run rows' stop arm — same control, same placement, same behavior. + const canStop = branch.kind === "run" ? branch.run.canStop : branch.stoppableRun !== null; const flat = variant === "flat"; // Flat rows carry no trunk to hang a status dot off, and a filed-away agent // that simply finished has nothing to say with one. Anything else does. @@ -295,6 +301,7 @@ export const AgentsPanel = memo(function AgentsPanel({ threadId, subagents, backgroundRuns, + subagentRuns, history, workEntries = EMPTY_WORK_ENTRIES, providerLabel, @@ -308,8 +315,8 @@ export const AgentsPanel = memo(function AgentsPanel({ const selectedAgentId = useSelectedAgentId(); const view = useMemo( - () => buildAgentsPanelView({ subagents, backgroundRuns, history, providerLabel }), - [backgroundRuns, history, providerLabel, subagents], + () => buildAgentsPanelView({ subagents, backgroundRuns, subagentRuns, history, providerLabel }), + [backgroundRuns, history, providerLabel, subagentRuns, subagents], ); const headerMeta = useMemo(() => formatAgentsHeaderMeta({ subagents }), [subagents]); const headerSummary = useMemo( @@ -345,8 +352,9 @@ export const AgentsPanel = memo(function AgentsPanel({ const handleStop = useCallback( (branch: AgentBranch) => { - if (branch.kind === "run") { - onStopBackgroundRun(branch.run); + const run = branch.kind === "run" ? branch.run : branch.stoppableRun; + if (run) { + onStopBackgroundRun(run); } }, [onStopBackgroundRun], diff --git a/apps/web/src/components/chat/agentsPanel.logic.test.ts b/apps/web/src/components/chat/agentsPanel.logic.test.ts index d29303e6f..d1ae8adb7 100644 --- a/apps/web/src/components/chat/agentsPanel.logic.test.ts +++ b/apps/web/src/components/chat/agentsPanel.logic.test.ts @@ -104,6 +104,41 @@ describe("buildAgentBranches", () => { ]); }); + it("lends a live agent the stop handle of the run it was promoted from", () => { + const run = buildRun({ id: "provider:task-codex-exec", source: "provider" }); + const branches = buildAgentBranches({ + subagents: [ + buildSubagent({ id: "live", spawnCallId: "tool-codex-exec" }), + buildSubagent({ + id: "settled", + status: "completed", + statusLabel: "Done", + spawnCallId: "tool-codex-exec-done", + }), + buildSubagent({ id: "native", spawnCallId: "tool-native" }), + ], + backgroundRuns: [], + subagentRuns: new Map([ + ["tool-codex-exec", run], + // A settled agent's process is already gone, so its row never offers + // a stop even when a run is still listed against it. + ["tool-codex-exec-done", buildRun({ id: "provider:task-done", source: "provider" })], + ]), + }); + + const stoppable = Object.fromEntries( + branches.map((branch) => [ + branch.key, + branch.kind === "subagent" ? (branch.stoppableRun?.id ?? null) : null, + ]), + ); + expect(stoppable).toEqual({ + "subagent:live": "provider:task-codex-exec", + "subagent:native": null, + "subagent:settled": null, + }); + }); + it("treats a started agent as running and an interrupted one as failed", () => { const branches = buildAgentBranches({ subagents: [ diff --git a/apps/web/src/components/chat/agentsPanel.logic.ts b/apps/web/src/components/chat/agentsPanel.logic.ts index 834a707d4..02da4eecb 100644 --- a/apps/web/src/components/chat/agentsPanel.logic.ts +++ b/apps/web/src/components/chat/agentsPanel.logic.ts @@ -59,6 +59,11 @@ export interface AgentSubagentBranch extends AgentBranchBase { readonly item: SubagentProgressItem; /** Only a subagent the provider can serve a transcript for can drill in. */ readonly transcriptAvailable: boolean; + /** The background run this agent is, for agents the provider launched as a + * shell command (a `codex exec` in the background). It carries the only + * stop handle the row has, so the row lends it the run rows' stop arm. + * Null once the agent has settled, and for every provider-native agent. */ + readonly stoppableRun: ThreadBackgroundRunItem | null; } export interface AgentRunBranch extends AgentBranchBase { @@ -146,10 +151,13 @@ function agentIdentityMetaParts(item: SubagentProgressItem): ReadonlyArray, ): AgentSubagentBranch { const status = subagentBranchStatus(item.status); const details = deriveSubagentDisplayDetails(item); const live = isLiveAgentBranchStatus(status); + const stoppableRun = + live && item.spawnCallId ? (runsBySpawnCallId?.get(item.spawnCallId) ?? null) : null; return { kind: "subagent", key: `subagent:${item.id}`, @@ -179,6 +187,7 @@ function subagentBranch( depth: Math.min(Math.max(item.treeDepth ?? 0, 0), MAX_BRANCH_DEPTH), item, transcriptAvailable: item.agentThreadId !== null, + stoppableRun: stoppableRun !== null && stoppableRun.canStop ? stoppableRun : null, }; } @@ -219,12 +228,15 @@ function runBranch( export function buildAgentBranches(input: { readonly subagents: ReadonlyArray; readonly backgroundRuns: ReadonlyArray; + /** Background runs that are already listed as subagents, keyed by the tool + * call that launched them. Lends each matching row its stop handle. */ + readonly subagentRuns?: ReadonlyMap | undefined; readonly providerLabel?: string | null | undefined; readonly nowMs?: number | undefined; }): ReadonlyArray { const branches: Array<{ branch: AgentBranch; startedAtMs: number | null }> = [ ...input.subagents.map((item) => ({ - branch: subagentBranch(item, input.nowMs) as AgentBranch, + branch: subagentBranch(item, input.nowMs, input.subagentRuns) as AgentBranch, startedAtMs: parseTimestamp(item.createdAt), })), ...input.backgroundRuns.map((run) => ({ @@ -294,6 +306,8 @@ function historyBranch( depth: Math.min(Math.max(item.treeDepth ?? 0, 0), MAX_BRANCH_DEPTH), item, transcriptAvailable: item.agentThreadId !== null, + // A filed-away agent is not running, so there is nothing to stop. + stoppableRun: null, }; } @@ -361,6 +375,7 @@ function providerDisplayLabel(providerLabel: string | null | undefined): string export function buildAgentsPanelView(input: { readonly subagents: ReadonlyArray; readonly backgroundRuns: ReadonlyArray; + readonly subagentRuns?: ReadonlyMap | undefined; readonly history?: ReadonlyArray | undefined; readonly providerLabel?: string | null | undefined; readonly nowMs?: number | undefined; diff --git a/apps/web/src/routes/_chat.$environmentId.$threadId.tsx b/apps/web/src/routes/_chat.$environmentId.$threadId.tsx index 0fe45da65..f99961194 100644 --- a/apps/web/src/routes/_chat.$environmentId.$threadId.tsx +++ b/apps/web/src/routes/_chat.$environmentId.$threadId.tsx @@ -440,6 +440,7 @@ function ChatThreadRouteView() { threadId={threadRef.threadId} subagents={agentsSource?.subagents ?? EMPTY_SUBAGENTS} backgroundRuns={agentsSource?.backgroundRuns ?? EMPTY_BACKGROUND_RUNS} + subagentRuns={agentsSource?.subagentRuns} history={agentsSource?.history ?? EMPTY_SUBAGENT_HISTORY} workEntries={agentsSource?.workEntries} providerLabel={agentsSource?.providerLabel} diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index ae83039a3..41aadedd6 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -218,6 +218,11 @@ export interface SubagentProgressItem { * from the spawning tool call id this record is keyed by; Codex addresses * transcripts by the child thread id, which is already the record id. */ transcriptAgentId: string | null; + /** Tool call that spawned this agent. Providers that launch an agent through + * an ordinary tool (a background `codex exec`) also report it on the task + * stream under the same id, which is how the plain run row is matched to + * this record. */ + spawnCallId?: string | null; /** Stable V2 hierarchy path (for example `/root/research/database`). */ agentPath?: string | null; parentAgentPath?: string | null; @@ -1143,6 +1148,7 @@ function collectSubagentActivityRecords( id: agentId, agentThreadId: subagent.agentThreadId, transcriptAgentId: subagent.transcriptAgentId, + spawnCallId: subagent.spawnCallId, agentPath: subagent.agentPath, parentAgentPath: subagent.parentAgentPath, treeDepth: subagent.treeDepth, @@ -1353,6 +1359,7 @@ function collectSubagentActivityRecords( transcriptAgentId: pendingAgent ? null : (taskIdByToolUseId.get(agentId) ?? previous?.transcriptAgentId ?? agentId), + spawnCallId: toolCallId, agentPath: pathMetadata?.agentPath ?? null, parentAgentPath: pathMetadata?.parentAgentPath ?? null, treeDepth: pathMetadata?.treeDepth ?? 0, diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index dae552065..35e31996b 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -14,6 +14,7 @@ import { TurnId, } from "./baseSchemas.ts"; import { ProviderInstanceId, ProviderDriverKind } from "./providerInstance.ts"; +import { OrchestrationSubagentStatus } from "./orchestration.ts"; const TrimmedNonEmptyStringSchema = TrimmedNonEmptyString; const UnknownRecordSchema = Schema.Record(Schema.String, Schema.Unknown); @@ -345,15 +346,27 @@ export const SubagentMetadataUpdatedPayload = Schema.Struct({ callId: Schema.optional(TrimmedNonEmptyStringSchema), agentThreadId: Schema.optional(TrimmedNonEmptyStringSchema), parentAgentThreadId: Schema.optional(TrimmedNonEmptyStringSchema), + /** Provider id the transcript route addresses this agent by, when it differs + * from `agentThreadId`. */ + transcriptAgentId: Schema.optional(TrimmedNonEmptyStringSchema), agentPath: Schema.optional(TrimmedNonEmptyStringSchema), agentNickname: Schema.optional(TrimmedNonEmptyStringSchema), agentRole: Schema.optional(TrimmedNonEmptyStringSchema), taskName: Schema.optional(TrimmedNonEmptyStringSchema), objective: Schema.optional(TrimmedNonEmptyStringSchema), + /** Lifecycle state, for providers whose subagents have no separate activity + * stream to infer it from. Absent leaves the projected status untouched. */ + status: Schema.optional(OrchestrationSubagentStatus), model: Schema.optional(TrimmedNonEmptyStringSchema), + /** The model the provider actually ran, when it is known separately from the + * requested one. */ + resolvedModel: Schema.optional(TrimmedNonEmptyStringSchema), reasoningEffort: Schema.optional(TrimmedNonEmptyStringSchema), modelSource: Schema.optional(SubagentMetadataProvenance), reasoningEffortSource: Schema.optional(SubagentMetadataProvenance), + /** The agent's final report, when the provider can recover one. */ + resultBody: Schema.optional(Schema.String), + resultCreatedAt: Schema.optional(TrimmedNonEmptyStringSchema), }); export type SubagentMetadataUpdatedPayload = typeof SubagentMetadataUpdatedPayload.Type;