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;