From 61a2b4d96a85ab66f53d218b4ff6d6622fbd0733 Mon Sep 17 00:00:00 2001 From: Juliusicon Date: Thu, 13 Aug 2026 02:19:08 +0200 Subject: [PATCH 1/7] feat(agents): surface managed Codex exec runs --- apps/server/src/auth/RpcAuthorization.ts | 2 + .../Layers/ProviderRuntimeIngestion.ts | 3 + .../orchestration/ManagedCodexExec.test.ts | 137 ++++++++ .../src/orchestration/ManagedCodexExec.ts | 294 ++++++++++++++++++ .../Services/ProviderRuntimeIngestion.ts | 4 + .../src/provider/Layers/CodexAdapter.ts | 14 +- .../provider/Layers/CodexSessionRuntime.ts | 16 + apps/server/src/ws.ts | 19 +- apps/web/src/components/AgentsPanel.tsx | 123 +++++++- .../client-runtime/src/state/orchestration.ts | 10 +- .../src/state/subagentRuntime.test.ts | 57 ++++ .../src/state/subagentRuntime.ts | 40 +++ packages/contracts/src/orchestration.ts | 35 +++ .../contracts/src/providerRuntime.test.ts | 25 ++ packages/contracts/src/providerRuntime.ts | 4 + packages/contracts/src/rpc.ts | 25 ++ 16 files changed, 792 insertions(+), 16 deletions(-) create mode 100644 apps/server/src/orchestration/ManagedCodexExec.test.ts create mode 100644 apps/server/src/orchestration/ManagedCodexExec.ts diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 36f348d6370a..a295aed5d15e 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -22,6 +22,8 @@ type WsRpcMethod = RpcGroup.Rpcs["_tag"]; */ export const RPC_REQUIRED_SCOPES = { [ORCHESTRATION_WS_METHODS.dispatchCommand]: AuthOrchestrationOperateScope, + [ORCHESTRATION_WS_METHODS.launchManagedCodexExec]: AuthOrchestrationOperateScope, + [ORCHESTRATION_WS_METHODS.cancelManagedAgent]: AuthOrchestrationOperateScope, [ORCHESTRATION_WS_METHODS.getWorkflowScript]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.getTurnDiff]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.getFullThreadDiff]: AuthOrchestrationReadScope, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 03253797242e..547ba8af649f 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -346,6 +346,8 @@ function taskLinkageActivityFields(payload: Record): Record worker.enqueue({ source: "runtime", event }), } satisfies ProviderRuntimeIngestionShape; }); diff --git a/apps/server/src/orchestration/ManagedCodexExec.test.ts b/apps/server/src/orchestration/ManagedCodexExec.test.ts new file mode 100644 index 000000000000..147edfd14f8e --- /dev/null +++ b/apps/server/src/orchestration/ManagedCodexExec.test.ts @@ -0,0 +1,137 @@ +import { describe, expect, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { ProjectId, ThreadId, type ProviderRuntimeEvent } from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Sink from "effect/Sink"; +import * as Stream from "effect/Stream"; +import { ChildProcessSpawner } from "effect/unstable/process"; + +import { ManagedCodexExec, layer } from "./ManagedCodexExec.ts"; +import { + ProviderRuntimeIngestionService, + type ProviderRuntimeIngestionShape, +} from "./Services/ProviderRuntimeIngestion.ts"; +import { + ProjectionSnapshotQuery, + type ProjectionSnapshotQueryShape, +} from "./Services/ProjectionSnapshotQuery.ts"; + +describe("ManagedCodexExec", () => { + it.effect("launches an owned codex exec and cancels its exact process handle", () => + Effect.gen(function* () { + const exit = yield* Deferred.make(); + const terminal = yield* Deferred.make(); + const events: ProviderRuntimeEvent[] = []; + let killed = false; + let spawnedCommand: unknown; + const handle = ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(1234), + exitCode: Deferred.await(exit), + isRunning: Effect.sync(() => !killed), + kill: () => + Effect.gen(function* () { + killed = true; + yield* Deferred.succeed(exit, ChildProcessSpawner.ExitCode(143)); + }), + unref: Effect.succeed(Effect.void), + stdin: Sink.drain, + stdout: Stream.empty, + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + }); + const spawner = ChildProcessSpawner.make((command) => + Effect.sync(() => { + spawnedCommand = command; + return handle; + }), + ); + const snapshots = { + getThreadDetailById: () => + Effect.succeed( + Option.some({ + id: ThreadId.make("thread-1"), + projectId: ProjectId.make("project-1"), + worktreePath: "D:/repo/worktree", + }), + ), + getProjectShellById: () => + Effect.succeed( + Option.some({ id: ProjectId.make("project-1"), workspaceRoot: "D:/repo" }), + ), + } as unknown as ProjectionSnapshotQueryShape; + const ingestion = { + start: () => Effect.void, + drain: Effect.void, + ingestRuntimeEvent: (event) => + Effect.sync(() => events.push(event)).pipe( + Effect.andThen( + event.type === "task.completed" + ? Deferred.succeed(terminal, event).pipe(Effect.asVoid) + : Effect.void, + ), + ), + } satisfies ProviderRuntimeIngestionShape; + const dependencies = Layer.mergeAll( + NodeServices.layer, + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), + Layer.succeed(ProjectionSnapshotQuery, snapshots), + Layer.succeed(ProviderRuntimeIngestionService, ingestion), + ); + + const context = yield* Layer.build(layer.pipe(Layer.provide(dependencies))); + const manager = yield* ManagedCodexExec.pipe(Effect.provide(context)); + const launched = yield* manager.launch({ + threadId: ThreadId.make("thread-1"), + prompt: "Review the implementation", + title: "Reviewer", + model: "gpt-5.6-sol", + effort: "high", + sandbox: "workspace-write", + parentAgentId: "native-parent", + }); + expect(launched.agentId).toMatch(/^managed-codex-exec:/); + expect(spawnedCommand).toMatchObject({ + command: "codex", + args: [ + "exec", + "--json", + "--color", + "never", + "-C", + "D:/repo/worktree", + "--model", + "gpt-5.6-sol", + "-c", + "model_reasoning_effort=high", + "--sandbox", + "workspace-write", + "Review the implementation", + ], + }); + expect(events[0]).toMatchObject({ + type: "task.started", + payload: { + taskId: launched.agentId, + parentAgentId: "native-parent", + model: "gpt-5.6-sol", + effort: "high", + cancellationOwner: "t3", + }, + }); + + expect( + yield* manager.cancel({ threadId: ThreadId.make("thread-1"), agentId: launched.agentId }), + ).toEqual({ cancelled: true }); + expect(killed).toBe(true); + expect(yield* Deferred.await(terminal)).toMatchObject({ + type: "task.completed", + payload: { taskId: launched.agentId, status: "stopped" }, + }); + }).pipe(Effect.scoped), + ); +}); diff --git a/apps/server/src/orchestration/ManagedCodexExec.ts b/apps/server/src/orchestration/ManagedCodexExec.ts new file mode 100644 index 000000000000..da8db82c987f --- /dev/null +++ b/apps/server/src/orchestration/ManagedCodexExec.ts @@ -0,0 +1,294 @@ +import { + EventId, + ManagedAgentRunError, + ProviderDriverKind, + RuntimeTaskId, + type ManagedAgentCancelInput, + type ManagedCodexExecLaunchInput, + type ProviderRuntimeEvent, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Scope from "effect/Scope"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; + +import { ProviderRuntimeIngestionService } from "./Services/ProviderRuntimeIngestion.ts"; +import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts"; + +interface ManagedRun { + readonly threadId: string; + readonly child: ChildProcessSpawner.ChildProcessHandle; + cancelled: boolean; +} + +export interface ManagedCodexExecShape { + readonly launch: ( + input: ManagedCodexExecLaunchInput, + ) => Effect.Effect<{ readonly agentId: string }, ManagedAgentRunError>; + readonly cancel: ( + input: ManagedAgentCancelInput, + ) => Effect.Effect<{ readonly cancelled: boolean }, ManagedAgentRunError>; +} + +export class ManagedCodexExec extends Context.Service()( + "t3/orchestration/ManagedCodexExec", +) {} + +function outputSummary(line: string): string | undefined { + const trimmed = line.trim(); + if (!trimmed) return undefined; + const decoded = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown))(trimmed); + if (Exit.isSuccess(decoded) && typeof decoded.value === "object" && decoded.value !== null) { + const value = decoded.value as Record; + const item = + typeof value.item === "object" && value.item !== null + ? (value.item as Record) + : undefined; + return ( + (typeof item?.command === "string" ? item.command : undefined) ?? + (typeof item?.text === "string" ? item.text : undefined) ?? + (typeof value.message === "string" ? value.message : undefined) ?? + (typeof value.type === "string" ? value.type.replaceAll(".", " ") : undefined) + ); + } + return trimmed; +} + +export const layer = Layer.effect( + ManagedCodexExec, + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const crypto = yield* Crypto.Crypto; + const scope = yield* Scope.Scope; + const snapshots = yield* ProjectionSnapshotQuery; + const ingestion = yield* Effect.serviceOption(ProviderRuntimeIngestionService); + const runs = new Map(); + const provider = ProviderDriverKind.make("codex"); + + const emit = ( + threadId: ManagedCodexExecLaunchInput["threadId"], + type: "task.started" | "task.progress" | "task.completed", + payload: ProviderRuntimeEvent["payload"], + ) => + Effect.gen(function* () { + const eventId = EventId.make(yield* crypto.randomUUIDv4); + const createdAt = DateTime.formatIso(yield* DateTime.now); + const ingestionService = Option.getOrUndefined(ingestion); + if (!ingestionService?.ingestRuntimeEvent) { + return yield* new ManagedAgentRunError({ + reason: "spawn-failed", + message: "Managed agent runtime ingestion is unavailable.", + }); + } + yield* ingestionService.ingestRuntimeEvent({ + eventId, + provider, + threadId, + createdAt, + type, + payload, + } as ProviderRuntimeEvent); + }); + + const launch: ManagedCodexExecShape["launch"] = Effect.fn("ManagedCodexExec.launch")( + function* (input) { + const threadOption = yield* snapshots.getThreadDetailById(input.threadId).pipe( + Effect.mapError( + () => + new ManagedAgentRunError({ + reason: "thread-not-found", + message: `Thread ${input.threadId} could not be loaded.`, + }), + ), + ); + const thread = Option.getOrUndefined(threadOption); + if (!thread) { + return yield* new ManagedAgentRunError({ + reason: "thread-not-found", + message: `Thread ${input.threadId} does not exist.`, + }); + } + const projectOption = yield* snapshots.getProjectShellById(thread.projectId).pipe( + Effect.mapError( + () => + new ManagedAgentRunError({ + reason: "thread-not-found", + message: `Project for thread ${input.threadId} could not be loaded.`, + }), + ), + ); + const project = Option.getOrUndefined(projectOption); + if (!project) { + return yield* new ManagedAgentRunError({ + reason: "thread-not-found", + message: `Project for thread ${input.threadId} does not exist.`, + }); + } + + const runUuid = yield* crypto.randomUUIDv4.pipe( + Effect.mapError( + () => + new ManagedAgentRunError({ + reason: "spawn-failed", + message: "Could not allocate a managed agent id.", + }), + ), + ); + const agentId = RuntimeTaskId.make(`managed-codex-exec:${runUuid}`); + const args = [ + "exec", + "--json", + "--color", + "never", + "-C", + thread.worktreePath ?? project.workspaceRoot, + ]; + if (input.model) args.push("--model", input.model); + if (input.effort) args.push("-c", `model_reasoning_effort=${input.effort}`); + if (input.sandbox) args.push("--sandbox", input.sandbox); + args.push(input.prompt); + + const child = yield* spawner + .spawn( + ChildProcess.make("codex", args, { + cwd: thread.worktreePath ?? project.workspaceRoot, + shell: false, + stdout: "pipe", + stderr: "pipe", + forceKillAfter: "3 seconds", + }), + ) + .pipe( + Effect.provideService(Scope.Scope, scope), + Effect.mapError( + (cause) => + new ManagedAgentRunError({ + reason: "spawn-failed", + message: `Could not launch managed Codex exec: ${Cause.pretty(Cause.fail(cause))}`, + }), + ), + ); + + const run: ManagedRun = { threadId: input.threadId, child, cancelled: false }; + runs.set(agentId, run); + const linkage = { + taskId: agentId, + taskType: "managed_codex_exec", + title: input.title, + role: "codex-exec", + ...(input.model ? { model: input.model } : {}), + ...(input.effort ? { effort: input.effort } : {}), + ...(input.parentAgentId ? { parentAgentId: input.parentAgentId } : {}), + agentSource: "managed_codex_exec" as const, + cancellationOwner: "t3" as const, + timelineBypass: true, + }; + yield* emit(input.threadId, "task.started", { + ...linkage, + description: input.title, + }).pipe( + Effect.catch(() => + child.kill().pipe( + Effect.ignore, + Effect.andThen( + Effect.sync(() => { + runs.delete(agentId); + }), + ), + Effect.andThen( + new ManagedAgentRunError({ + reason: "spawn-failed", + message: "Managed Codex exec started but its lifecycle could not be recorded.", + }), + ), + ), + ), + ); + + const reportLines = (stream: Stream.Stream) => + stream.pipe( + Stream.decodeText(), + Stream.splitLines, + Stream.mapEffect((line) => { + const summary = outputSummary(line); + return summary + ? emit(input.threadId, "task.progress", { + ...linkage, + description: input.title, + summary, + }) + : Effect.void; + }), + Stream.runDrain, + Effect.ignore, + ); + + yield* Effect.forkIn( + Effect.gen(function* () { + yield* Effect.all([reportLines(child.stdout), reportLines(child.stderr)], { + concurrency: "unbounded", + }); + const exitCode = Number(yield* child.exitCode); + runs.delete(agentId); + yield* emit(input.threadId, "task.completed", { + ...linkage, + status: run.cancelled ? "stopped" : exitCode === 0 ? "completed" : "failed", + summary: run.cancelled + ? "Cancelled by T3" + : exitCode === 0 + ? "Managed Codex exec completed" + : `Managed Codex exec exited with code ${exitCode}`, + }); + }), + scope, + ); + return { agentId }; + }, + ); + + const cancel: ManagedCodexExecShape["cancel"] = Effect.fn("ManagedCodexExec.cancel")( + function* (input) { + const run = runs.get(input.agentId); + if (!run) { + return yield* new ManagedAgentRunError({ + reason: "run-not-found", + message: `Managed agent ${input.agentId} is not running.`, + }); + } + if (run.threadId !== input.threadId) { + return yield* new ManagedAgentRunError({ + reason: "not-owned", + message: `Managed agent ${input.agentId} does not belong to thread ${input.threadId}.`, + }); + } + run.cancelled = true; + yield* run.child.kill({ killSignal: "SIGTERM", forceKillAfter: "3 seconds" }).pipe( + Effect.mapError( + () => + new ManagedAgentRunError({ + reason: "not-owned", + message: `Managed agent ${input.agentId} could not be cancelled.`, + }), + ), + ); + return { cancelled: true }; + }, + ); + + yield* Effect.addFinalizer(() => + Effect.forEach(runs.values(), (run) => run.child.kill().pipe(Effect.ignore), { + discard: true, + concurrency: "unbounded", + }), + ); + return { launch, cancel }; + }), +); diff --git a/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts index b6fa2711b949..7ebfb622707b 100644 --- a/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts @@ -9,6 +9,7 @@ import * as Context from "effect/Context"; import type * as Effect from "effect/Effect"; import type * as Scope from "effect/Scope"; +import type { ProviderRuntimeEvent } from "@t3tools/contracts"; /** * ProviderRuntimeIngestionShape - Service API for runtime ingestion lifecycle. @@ -30,6 +31,9 @@ export interface ProviderRuntimeIngestionShape { * Intended for test use to replace timing-sensitive sleeps. */ readonly drain: Effect.Effect; + + /** Explicit adapter boundary for T3-owned runtimes that do not use a provider session. */ + readonly ingestRuntimeEvent?: (event: ProviderRuntimeEvent) => Effect.Effect; } /** diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 6b99bf52b1e0..0c4b25d6fab3 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -523,6 +523,8 @@ function mapCollabAgentEvent( const nickname = typeof payload.nickname === "string" ? payload.nickname : undefined; const role = (typeof payload.role === "string" ? payload.role : undefined) ?? pathLeaf ?? "general-purpose"; + const model = typeof payload.model === "string" ? payload.model : undefined; + const effort = typeof payload.effort === "string" ? payload.effort : undefined; // A bare thread id is not a name. Omitting the title lets the client fold // keep the real one from task.started instead of clobbering it (probe // finding: progress rows renamed math_one to its UUID). @@ -535,7 +537,11 @@ function mapCollabAgentEvent( role, ...(knownName ? { title: knownName } : {}), ...(agentPath ? { agentPath } : {}), + ...(model ? { model } : {}), + ...(effort ? { effort } : {}), timelineBypass: true, + agentSource: "provider" as const, + cancellationOwner: "provider" as const, } as const; switch (event.method) { @@ -548,12 +554,10 @@ function mapCollabAgentEvent( taskId, description: title, title, - role, - ...(agentPath ? { agentPath } : {}), + ...statusLinkage, ...(typeof payload.parentThreadId === "string" ? { parentAgentId: payload.parentThreadId } : {}), - timelineBypass: true, }, }, ]; @@ -581,9 +585,7 @@ function mapCollabAgentEvent( taskId, description: title, title, - role, - ...(agentPath ? { agentPath } : {}), - timelineBypass: true, + ...statusLinkage, }, }, ]; diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 58c012bd63ea..13ecc8451b4a 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -624,6 +624,8 @@ interface CollabChildAgentState { readonly agentPath: string | undefined; readonly depth: number | undefined; readonly parentThreadId: string | undefined; + readonly model: string | undefined; + readonly effort: string | undefined; /** * Parent canonical turn active when the child registered. Stamped on every * synthetic collabAgent/* event so clients can batch a fleet by its spawn @@ -640,6 +642,8 @@ function readThreadSpawnSource(thread: { readonly source: unknown }): agentPath: string | undefined; depth: number | undefined; parentThreadId: string | undefined; + model: string | undefined; + effort: string | undefined; } | undefined { const source = thread.source; @@ -655,6 +659,7 @@ function readThreadSpawnSource(thread: { readonly source: unknown }): return undefined; } const record = spawn as Record; + const threadRecord = thread as Record; return { nickname: typeof record.agent_nickname === "string" ? record.agent_nickname : undefined, role: typeof record.agent_role === "string" ? record.agent_role : undefined, @@ -662,6 +667,9 @@ function readThreadSpawnSource(thread: { readonly source: unknown }): depth: typeof record.depth === "number" ? record.depth : undefined, parentThreadId: typeof record.parent_thread_id === "string" ? record.parent_thread_id : undefined, + model: typeof threadRecord.model === "string" ? threadRecord.model : undefined, + effort: + typeof threadRecord.reasoningEffort === "string" ? threadRecord.reasoningEffort : undefined, }; } @@ -1008,6 +1016,8 @@ export const makeCodexSessionRuntime = ( depth: spawn.depth ?? existingChild?.depth, parentThreadId: spawn.parentThreadId ?? thread.parentThreadId ?? existingChild?.parentThreadId, + model: spawn.model ?? existingChild?.model, + effort: spawn.effort ?? existingChild?.effort, spawnTurnId, }; yield* Ref.update(collabChildAgentsRef, (current) => { @@ -1027,6 +1037,8 @@ export const makeCodexSessionRuntime = ( ...(state.agentPath ? { agentPath: state.agentPath } : {}), ...(state.depth !== undefined ? { depth: state.depth } : {}), ...(state.parentThreadId ? { parentThreadId: state.parentThreadId } : {}), + ...(state.model ? { model: state.model } : {}), + ...(state.effort ? { effort: state.effort } : {}), }, }); return true; @@ -1073,6 +1085,8 @@ export const makeCodexSessionRuntime = ( agentPath: existing?.agentPath ?? item.agentPath, depth: existing?.depth, parentThreadId: existing?.parentThreadId, + model: existing?.model, + effort: existing?.effort, spawnTurnId: existing ? existing.spawnTurnId : activitySpawnTurnId, }); return next; @@ -1114,6 +1128,8 @@ export const makeCodexSessionRuntime = ( ...(child.nickname ? { nickname: child.nickname } : {}), ...(child.role ? { role: child.role } : {}), ...(child.agentPath ? { agentPath: child.agentPath } : {}), + ...(child.model ? { model: child.model } : {}), + ...(child.effort ? { effort: child.effort } : {}), }; switch (notification.method) { case "turn/started": { diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 173c89ecabff..be3eec15f548 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -92,6 +92,7 @@ import * as PortScanner from "./preview/PortScanner.ts"; import * as WorkspaceEntries from "./workspace/WorkspaceEntries.ts"; import * as WorkspaceFileSystem from "./workspace/WorkspaceFileSystem.ts"; import { readWorkflowScript } from "./orchestration/workflowScriptQuery.ts"; +import * as ManagedCodexExec from "./orchestration/ManagedCodexExec.ts"; import * as WorkspacePaths from "./workspace/WorkspacePaths.ts"; import * as VcsStatusBroadcaster from "./vcs/VcsStatusBroadcaster.ts"; import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts"; @@ -351,6 +352,7 @@ function toAuthAccessStreamEvent( const makeWsRpcLayer = ( currentSession: EnvironmentAuth.AuthenticatedSession, previewAutomationBroker: PreviewAutomationBroker.PreviewAutomationBroker["Service"], + managedCodexExec: ManagedCodexExec.ManagedCodexExecShape, ) => WsRpcGroup.toLayer( Effect.gen(function* () { @@ -1033,6 +1035,18 @@ const makeWsRpcLayer = ( .pipe(Effect.ignoreCause({ log: true }), Effect.forkDetach, Effect.asVoid); return WsRpcGroup.of({ + [ORCHESTRATION_WS_METHODS.launchManagedCodexExec]: (input) => + observeRpcEffect( + ORCHESTRATION_WS_METHODS.launchManagedCodexExec, + managedCodexExec.launch(input), + { "rpc.aggregate": "thread", threadId: input.threadId }, + ), + [ORCHESTRATION_WS_METHODS.cancelManagedAgent]: (input) => + observeRpcEffect( + ORCHESTRATION_WS_METHODS.cancelManagedAgent, + managedCodexExec.cancel(input), + { "rpc.aggregate": "thread", threadId: input.threadId }, + ), [ORCHESTRATION_WS_METHODS.dispatchCommand]: (command) => observeRpcEffect( ORCHESTRATION_WS_METHODS.dispatchCommand, @@ -2280,6 +2294,7 @@ export const websocketRpcRouteLayer = Layer.unwrap( const previewAutomationBroker = yield* PreviewAutomationBroker.PreviewAutomationBroker; const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate; const pullRequests = yield* PullRequestService.PullRequestService; + const managedCodexExec = yield* ManagedCodexExec.ManagedCodexExec; return HttpRouter.add( "GET", "/ws", @@ -2299,7 +2314,7 @@ export const websocketRpcRouteLayer = Layer.unwrap( disableTracing: true, }).pipe( Effect.provide( - makeWsRpcLayer(session, previewAutomationBroker).pipe( + makeWsRpcLayer(session, previewAutomationBroker, managedCodexExec).pipe( Layer.provideMerge(RpcSerialization.layerJson), Layer.provide(ProviderMaintenanceRunner.layer), Layer.provide(Layer.succeed(ServerSelfUpdate.ServerSelfUpdate, serverSelfUpdate)), @@ -2343,4 +2358,4 @@ export const websocketRpcRouteLayer = Layer.unwrap( ), ); }), -); +).pipe(Layer.provide(ManagedCodexExec.layer)); diff --git a/apps/web/src/components/AgentsPanel.tsx b/apps/web/src/components/AgentsPanel.tsx index 4eeff67ce5f7..82a8eb20c2c1 100644 --- a/apps/web/src/components/AgentsPanel.tsx +++ b/apps/web/src/components/AgentsPanel.tsx @@ -22,12 +22,13 @@ import { formatSubagentTokenCount, } from "@t3tools/client-runtime/state/subagentRuntime"; import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; -import { Bot, Braces, Check, ChevronDown, ChevronRight, X } from "lucide-react"; +import { Bot, Braces, Check, ChevronDown, ChevronRight, Square, X } from "lucide-react"; import { useEffect, useRef, useState } from "react"; import { cn } from "~/lib/utils"; import { orchestrationEnvironment } from "~/state/orchestration"; import { ScrollArea } from "~/components/ui/scroll-area"; +import { useAtomCommand } from "~/state/use-atom-command"; /** * In-flight states all present as Working (one steady state, per the @@ -137,7 +138,13 @@ function agentActivityText(agent: RuntimeSubagent): string | null { } /** Flat, non-interactive agent status line. No unfold. */ -function AgentRow({ agent }: { agent: RuntimeSubagent }) { +function AgentRow({ + agent, + onCancel, +}: { + agent: RuntimeSubagent; + onCancel?: ((agent: RuntimeSubagent) => void) | undefined; +}) { const visuals = STATUS_VISUALS[agent.status]; const activity = agentActivityText(agent); const modelLabel = formatSubagentModelLabel(agent.model, agent.effort); @@ -168,6 +175,21 @@ function AgentRow({ agent }: { agent: RuntimeSubagent }) { + {onCancel && + agent.cancellationOwner === "t3" && + (agent.status === "pending" || + agent.status === "running" || + agent.status === "waiting") ? ( + + ) : null} {agent.status === "completed" ? ( ) : null} @@ -189,6 +211,40 @@ function AgentRow({ agent }: { agent: RuntimeSubagent }) { ); } +function AgentTreeRows({ + agent, + childrenByParentId, + onCancel, + depth = 0, + ancestors = new Set(), +}: { + agent: RuntimeSubagent; + childrenByParentId: AgentPanelModel["childrenByParentId"]; + onCancel?: ((agent: RuntimeSubagent) => void) | undefined; + depth?: number; + ancestors?: ReadonlySet; +}) { + const cyclic = ancestors.has(agent.id); + const nextAncestors = new Set(ancestors).add(agent.id); + return ( +
+ + {!cyclic + ? (childrenByParentId.get(agent.id) ?? []).map((child) => ( + + )) + : null} +
+ ); +} + function workflowIsLive(group: AgentPanelWorkflowGroup): boolean { const status = group.workflow.status; return ( @@ -314,9 +370,13 @@ function WorkflowScriptView({ */ function PhaseSection({ phase, + childrenByParentId, + onCancel, defaultOpen = false, }: { phase: AgentPanelWorkflowGroup["phases"][number]; + childrenByParentId: AgentPanelModel["childrenByParentId"]; + onCancel?: ((agent: RuntimeSubagent) => void) | undefined; defaultOpen?: boolean; }) { const [open, setOpen] = useState(defaultOpen || phase.state === "running"); @@ -366,7 +426,16 @@ function PhaseSection({ ) : null} - {open ? phase.members.map((member) => ) : null} + {open + ? phase.members.map((member) => ( + + )) + : null} ); } @@ -377,11 +446,15 @@ function ExpandedWorkflowSection({ environmentId, threadId, onCollapse, + childrenByParentId, + onCancel, }: { group: AgentPanelWorkflowGroup; environmentId: EnvironmentId | null; threadId: ThreadId | null; onCollapse: () => void; + childrenByParentId: AgentPanelModel["childrenByParentId"]; + onCancel?: ((agent: RuntimeSubagent) => void) | undefined; }) { const [scriptOpen, setScriptOpen] = useState(false); const members = workflowMembers(group); @@ -436,13 +509,24 @@ function ExpandedWorkflowSection({ /> ) : null} {group.phases.map((phase) => ( - + ))} {group.unphasedMembers.map((member) => ( - + ))} {group.phases.length === 0 && group.unphasedMembers.length === 0 ? ( - + ) : null} ); @@ -500,10 +584,14 @@ function WorkflowSection({ group, environmentId, threadId, + childrenByParentId, + onCancel, }: { group: AgentPanelWorkflowGroup; environmentId: EnvironmentId | null; threadId: ThreadId | null; + childrenByParentId: AgentPanelModel["childrenByParentId"]; + onCancel?: ((agent: RuntimeSubagent) => void) | undefined; }) { const [open, setOpen] = useState(() => workflowIsLive(group)); return open ? ( @@ -512,6 +600,8 @@ function WorkflowSection({ environmentId={environmentId} threadId={threadId} onCollapse={() => setOpen(false)} + childrenByParentId={childrenByParentId} + onCancel={onCancel} /> ) : ( setOpen(true)} /> @@ -527,6 +617,18 @@ export function AgentsPanel({ environmentId?: EnvironmentId | null; threadId?: ThreadId | null; }) { + const cancelManagedAgent = useAtomCommand(orchestrationEnvironment.cancelManagedAgent, { + reportFailure: true, + }); + const onCancel = + environmentId !== null && threadId !== null + ? (agent: RuntimeSubagent) => { + void cancelManagedAgent({ + environmentId, + input: { threadId, agentId: agent.id }, + }); + } + : undefined; if (!model.hasAgents) { return (
@@ -550,6 +652,8 @@ export function AgentsPanel({ group={group} environmentId={environmentId} threadId={threadId} + childrenByParentId={model.childrenByParentId} + onCancel={onCancel} /> ))} {model.directAgents.length > 0 ? ( @@ -558,7 +662,12 @@ export function AgentsPanel({ Direct spawns
{model.directAgents.map((agent) => ( - + ))} ) : null} diff --git a/packages/client-runtime/src/state/orchestration.ts b/packages/client-runtime/src/state/orchestration.ts index ba80275bffb3..ffe834c4b9b8 100644 --- a/packages/client-runtime/src/state/orchestration.ts +++ b/packages/client-runtime/src/state/orchestration.ts @@ -1,13 +1,21 @@ import { ORCHESTRATION_WS_METHODS } from "@t3tools/contracts"; import { Atom } from "effect/unstable/reactivity"; -import { createEnvironmentRpcQueryAtomFamily } from "./runtime.ts"; +import { createEnvironmentRpcCommand, createEnvironmentRpcQueryAtomFamily } from "./runtime.ts"; import type { EnvironmentRegistry } from "../connection/registry.ts"; export function createOrchestrationEnvironmentAtoms( runtime: Atom.AtomRuntime, ) { return { + launchManagedCodexExec: createEnvironmentRpcCommand(runtime, { + label: "environment-data:orchestration:launch-managed-codex-exec", + tag: ORCHESTRATION_WS_METHODS.launchManagedCodexExec, + }), + cancelManagedAgent: createEnvironmentRpcCommand(runtime, { + label: "environment-data:orchestration:cancel-managed-agent", + tag: ORCHESTRATION_WS_METHODS.cancelManagedAgent, + }), turnDiff: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:orchestration:turn-diff", tag: ORCHESTRATION_WS_METHODS.getTurnDiff, diff --git a/packages/client-runtime/src/state/subagentRuntime.test.ts b/packages/client-runtime/src/state/subagentRuntime.test.ts index ceb40517550e..a6f1dcb2caea 100644 --- a/packages/client-runtime/src/state/subagentRuntime.test.ts +++ b/packages/client-runtime/src/state/subagentRuntime.test.ts @@ -480,6 +480,34 @@ describe("deriveAgentPanelModel", () => { expect(model.workflows).toHaveLength(0); expect(model.directAgents.map((agent) => agent.id)).toEqual(["gone:wf:0"]); }); + + it("keeps native and managed descendants under their nearest visible parent", () => { + const nested = fold([ + activity("task.started", { taskId: "parent", title: "Parent" }), + activity("task.started", { + taskId: "native-child", + title: "Native child", + parentAgentId: "parent", + }), + activity("task.started", { + taskId: "managed-grandchild", + taskType: "managed_codex_exec", + title: "Managed grandchild", + parentAgentId: "native-child", + agentSource: "managed_codex_exec", + cancellationOwner: "t3", + }), + ]); + + const model = deriveAgentPanelModel({ agents: nested }); + expect(model.directAgents.map((agent) => agent.id)).toEqual(["parent"]); + expect(model.childrenByParentId.get("parent")?.map((agent) => agent.id)).toEqual([ + "native-child", + ]); + expect(model.childrenByParentId.get("native-child")?.map((agent) => agent.id)).toEqual([ + "managed-grandchild", + ]); + }); }); describe("workflowCardMembers", () => { @@ -567,6 +595,35 @@ describe("model and effort attribution", () => { expect(agents[0]!.effort).toBe("high"); }); + it("retains managed-run ownership through a terminal lifecycle row", () => { + const agents = fold([ + activity("task.started", { + taskId: "managed-codex-exec:stable-id", + taskType: "managed_codex_exec", + title: "Review implementation", + model: "gpt-5.6-sol", + effort: "high", + agentSource: "managed_codex_exec", + cancellationOwner: "t3", + }), + activity("task.completed", { + taskId: "managed-codex-exec:stable-id", + status: "stopped", + summary: "Cancelled by T3", + }), + ]); + + expect(agents[0]).toMatchObject({ + id: "managed-codex-exec:stable-id", + model: "gpt-5.6-sol", + effort: "high", + status: "interrupted", + result: "Cancelled by T3", + agentSource: "managed_codex_exec", + cancellationOwner: "t3", + }); + }); + it("formatSubagentModelLabel compacts ids and appends effort", () => { expect(formatSubagentModelLabel("claude-sonnet-5[1m]", "high")).toBe("sonnet-5[1m] ยท high"); expect(formatSubagentModelLabel("claude-opus-4-20250514", null)).toBe("opus-4"); diff --git a/packages/client-runtime/src/state/subagentRuntime.ts b/packages/client-runtime/src/state/subagentRuntime.ts index e5f2b586b8c4..485f99fa1450 100644 --- a/packages/client-runtime/src/state/subagentRuntime.ts +++ b/packages/client-runtime/src/state/subagentRuntime.ts @@ -79,6 +79,8 @@ export interface RuntimeSubagent { readonly workflowName: string | null; readonly phases: ReadonlyArray; readonly runHandles: SubagentRunHandles | null; + readonly agentSource: "provider" | "managed_codex_exec"; + readonly cancellationOwner: "provider" | "t3" | "none"; readonly recentActivity: ReadonlyArray; /** First retained observation, used as the roster's stable display order. */ readonly firstSeenAt: string; @@ -248,6 +250,8 @@ interface MutableAgent { workflowName: string | null; phases: ReadonlyArray; runHandles: SubagentRunHandles | null; + agentSource: RuntimeSubagent["agentSource"]; + cancellationOwner: RuntimeSubagent["cancellationOwner"]; recentActivity: ReadonlyArray; firstSeenAt: string; startedAt: string | null; @@ -302,6 +306,14 @@ function getOrCreate( workflowName: asString(payload.workflowName) ?? null, phases: [], runHandles: null, + agentSource: + asString(payload.agentSource) === "managed_codex_exec" ? "managed_codex_exec" : "provider", + cancellationOwner: + asString(payload.cancellationOwner) === "t3" + ? "t3" + : asString(payload.cancellationOwner) === "provider" + ? "provider" + : "none", recentActivity: [], firstSeenAt: at, startedAt: null, @@ -352,6 +364,18 @@ function fillMetadata(agent: MutableAgent, payload: Record): vo } const outputFile = asString(payload.outputFile); if (outputFile) agent.outputFile = outputFile; + const agentSource = asString(payload.agentSource); + if (agentSource === "provider" || agentSource === "managed_codex_exec") { + agent.agentSource = agentSource; + } + const cancellationOwner = asString(payload.cancellationOwner); + if ( + cancellationOwner === "provider" || + cancellationOwner === "t3" || + cancellationOwner === "none" + ) { + agent.cancellationOwner = cancellationOwner; + } if (Array.isArray(payload.phases)) { const phases: SubagentWorkflowPhase[] = []; for (const entry of payload.phases) { @@ -692,6 +716,7 @@ export interface AgentPanelWorkflowGroup { export interface AgentPanelModel { readonly workflows: ReadonlyArray; readonly directAgents: ReadonlyArray; + readonly childrenByParentId: ReadonlyMap>; readonly runningCount: number; readonly waitingCount: number; readonly idleCount: number; @@ -704,6 +729,7 @@ export interface AgentPanelModel { const EMPTY_PANEL_MODEL: AgentPanelModel = { workflows: [], directAgents: [], + childrenByParentId: new Map(), runningCount: 0, waitingCount: 0, idleCount: 0, @@ -741,7 +767,9 @@ export function deriveAgentPanelModel({ .sort((a, b) => a.firstSeenAt.localeCompare(b.firstSeenAt) || a.id.localeCompare(b.id)); const workflowIds = new Set(workflows.map((workflow) => workflow.id)); const members = new Map(); + const childrenByParentId = new Map(); const direct: RuntimeSubagent[] = []; + const agentIds = new Set(source.map((agent) => agent.id)); for (const agent of source) { if (agent.kind === "workflow") { @@ -751,6 +779,10 @@ export function deriveAgentPanelModel({ const list = members.get(agent.parentAgentId) ?? []; list.push(agent); members.set(agent.parentAgentId, list); + } else if (agent.parentAgentId !== null && agentIds.has(agent.parentAgentId)) { + const list = childrenByParentId.get(agent.parentAgentId) ?? []; + list.push(agent); + childrenByParentId.set(agent.parentAgentId, list); } else { // Orphaned members (coordinator aged out) fall back to the direct list. direct.push(agent); @@ -844,6 +876,14 @@ export function deriveAgentPanelModel({ directAgents: direct .slice() .sort((a, b) => a.firstSeenAt.localeCompare(b.firstSeenAt) || a.id.localeCompare(b.id)), + childrenByParentId: new Map( + Array.from(childrenByParentId, ([parentId, children]) => [ + parentId, + children + .slice() + .sort((a, b) => a.firstSeenAt.localeCompare(b.firstSeenAt) || a.id.localeCompare(b.id)), + ]), + ), runningCount, waitingCount, idleCount, diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 35fef721efa7..85e7bf979e37 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -25,6 +25,8 @@ import { ProviderInstanceId } from "./providerInstance.ts"; export const ORCHESTRATION_WS_METHODS = { dispatchCommand: "orchestration.dispatchCommand", + launchManagedCodexExec: "orchestration.launchManagedCodexExec", + cancelManagedAgent: "orchestration.cancelManagedAgent", getWorkflowScript: "orchestration.getWorkflowScript", getTurnDiff: "orchestration.getTurnDiff", getFullThreadDiff: "orchestration.getFullThreadDiff", @@ -34,6 +36,39 @@ export const ORCHESTRATION_WS_METHODS = { subscribeThread: "orchestration.subscribeThread", } as const; +export const ManagedCodexExecLaunchInput = Schema.Struct({ + threadId: ThreadId, + prompt: TrimmedNonEmptyString, + title: TrimmedNonEmptyString, + model: Schema.optional(TrimmedNonEmptyString), + effort: Schema.optional(TrimmedNonEmptyString), + sandbox: Schema.optional(Schema.Literals(["read-only", "workspace-write", "danger-full-access"])), + parentAgentId: Schema.optional(TrimmedNonEmptyString), +}); +export type ManagedCodexExecLaunchInput = typeof ManagedCodexExecLaunchInput.Type; + +export const ManagedCodexExecLaunchResult = Schema.Struct({ + agentId: TrimmedNonEmptyString, +}); +export type ManagedCodexExecLaunchResult = typeof ManagedCodexExecLaunchResult.Type; + +export const ManagedAgentCancelInput = Schema.Struct({ + threadId: ThreadId, + agentId: TrimmedNonEmptyString, +}); +export type ManagedAgentCancelInput = typeof ManagedAgentCancelInput.Type; + +export const ManagedAgentCancelResult = Schema.Struct({ cancelled: Schema.Boolean }); +export type ManagedAgentCancelResult = typeof ManagedAgentCancelResult.Type; + +export class ManagedAgentRunError extends Schema.TaggedErrorClass()( + "ManagedAgentRunError", + { + reason: Schema.Literals(["thread-not-found", "spawn-failed", "run-not-found", "not-owned"]), + message: TrimmedNonEmptyString, + }, +) {} + export const ProviderApprovalPolicy = Schema.Literals([ "untrusted", "on-failure", diff --git a/packages/contracts/src/providerRuntime.test.ts b/packages/contracts/src/providerRuntime.test.ts index 7563ac155681..fa61f4e375ed 100644 --- a/packages/contracts/src/providerRuntime.test.ts +++ b/packages/contracts/src/providerRuntime.test.ts @@ -181,6 +181,31 @@ describe("ProviderRuntimeEvent", () => { expect(parsed.payload.usage.maxTokens).toBe(200000); expect(parsed.payload.usage.usedTokens).toBe(31251); }); + + it("decodes managed Codex exec linkage and cancellation ownership", () => { + const parsed = decodeRuntimeEvent({ + type: "task.started", + eventId: "event-managed-exec-1", + provider: "codex", + createdAt: "2026-02-28T00:00:05.000Z", + threadId: "thread-1", + payload: { + taskId: "managed-codex-exec:stable-id", + taskType: "managed_codex_exec", + parentAgentId: "native-child-id", + model: "gpt-5.6-sol", + effort: "high", + agentSource: "managed_codex_exec", + cancellationOwner: "t3", + }, + }); + + expect(parsed.type).toBe("task.started"); + if (parsed.type !== "task.started") throw new Error("expected task.started"); + expect(parsed.payload.agentSource).toBe("managed_codex_exec"); + expect(parsed.payload.cancellationOwner).toBe("t3"); + expect(parsed.payload.parentAgentId).toBe("native-child-id"); + }); }); describe("classifyTaskAgentKind", () => { diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index bd525e6542e2..771eacc4718d 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -581,6 +581,10 @@ const taskAgentLinkageFields = { * belongs in the Agents surface, never the parent timeline. */ timelineBypass: Schema.optional(Schema.Boolean), + /** Origin of the agent run. Managed exec runs are launched by this T3 server. */ + agentSource: Schema.optional(Schema.Literals(["provider", "managed_codex_exec"])), + /** Which component owns an actionable cancellation handle for this run. */ + cancellationOwner: Schema.optional(Schema.Literals(["provider", "t3", "none"])), } as const; export const TaskAgentLinkage = Schema.Struct(taskAgentLinkageFields); diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index b5bd91cad59c..84a00e77ca6c 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -65,6 +65,11 @@ import { OrchestrationGetTurnDiffInput, OrchestrationRpcSchemas, OrchestrationGetWorkflowScriptError, + ManagedCodexExecLaunchInput, + ManagedCodexExecLaunchResult, + ManagedAgentCancelInput, + ManagedAgentCancelResult, + ManagedAgentRunError, } from "./orchestration.ts"; import { ProviderInstanceId } from "./providerInstance.ts"; import { @@ -865,6 +870,24 @@ export const WsOrchestrationDispatchCommandRpc = Rpc.make( }, ); +export const WsOrchestrationLaunchManagedCodexExecRpc = Rpc.make( + ORCHESTRATION_WS_METHODS.launchManagedCodexExec, + { + payload: ManagedCodexExecLaunchInput, + success: ManagedCodexExecLaunchResult, + error: Schema.Union([ManagedAgentRunError, EnvironmentAuthorizationError]), + }, +); + +export const WsOrchestrationCancelManagedAgentRpc = Rpc.make( + ORCHESTRATION_WS_METHODS.cancelManagedAgent, + { + payload: ManagedAgentCancelInput, + success: ManagedAgentCancelResult, + error: Schema.Union([ManagedAgentRunError, EnvironmentAuthorizationError]), + }, +); + export const WsOrchestrationGetWorkflowScriptRpc = Rpc.make( ORCHESTRATION_WS_METHODS.getWorkflowScript, { @@ -1062,6 +1085,8 @@ export const WsRpcGroup = RpcGroup.make( WsSubscribeBackgroundPolicyRpc, WsSubscribeResourceTelemetryRpc, WsOrchestrationDispatchCommandRpc, + WsOrchestrationLaunchManagedCodexExecRpc, + WsOrchestrationCancelManagedAgentRpc, WsOrchestrationGetWorkflowScriptRpc, WsOrchestrationGetTurnDiffRpc, WsOrchestrationGetFullThreadDiffRpc, From a84b9eebe73cf0cdbd54a0022e4ce241ae36d906 Mon Sep 17 00:00:00 2001 From: Juliusicon Date: Thu, 13 Aug 2026 02:30:26 +0200 Subject: [PATCH 2/7] fix(server): terminalize signaled Codex exec runs --- .../orchestration/ManagedCodexExec.test.ts | 65 +++++++++++++++++-- .../src/orchestration/ManagedCodexExec.ts | 17 +++-- 2 files changed, 70 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration/ManagedCodexExec.test.ts b/apps/server/src/orchestration/ManagedCodexExec.test.ts index 147edfd14f8e..fd21bd15aac3 100644 --- a/apps/server/src/orchestration/ManagedCodexExec.test.ts +++ b/apps/server/src/orchestration/ManagedCodexExec.test.ts @@ -5,6 +5,8 @@ import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as PlatformError from "effect/PlatformError"; +import * as Queue from "effect/Queue"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; import { ChildProcessSpawner } from "effect/unstable/process"; @@ -20,21 +22,30 @@ import { } from "./Services/ProjectionSnapshotQuery.ts"; describe("ManagedCodexExec", () => { - it.effect("launches an owned codex exec and cancels its exact process handle", () => + it.effect("terminalizes signal exits and releases owned process handles", () => Effect.gen(function* () { - const exit = yield* Deferred.make(); - const terminal = yield* Deferred.make(); + const exit = yield* Deferred.make< + ChildProcessSpawner.ExitCode, + PlatformError.PlatformError + >(); + const terminals = yield* Queue.unbounded(); const events: ProviderRuntimeEvent[] = []; let killed = false; let spawnedCommand: unknown; - const handle = ChildProcessSpawner.makeHandle({ + const signalExitError = PlatformError.systemError({ + _tag: "Unknown", + module: "ChildProcess", + method: "exitCode", + description: "Process interrupted due to receipt of signal: 'SIGTERM'", + }); + const cancelledHandle = ChildProcessSpawner.makeHandle({ pid: ChildProcessSpawner.ProcessId(1234), exitCode: Deferred.await(exit), isRunning: Effect.sync(() => !killed), kill: () => Effect.gen(function* () { killed = true; - yield* Deferred.succeed(exit, ChildProcessSpawner.ExitCode(143)); + yield* Deferred.fail(exit, signalExitError); }), unref: Effect.succeed(Effect.void), stdin: Sink.drain, @@ -44,9 +55,25 @@ describe("ManagedCodexExec", () => { getInputFd: () => Sink.drain, getOutputFd: () => Stream.empty, }); + const unexpectedExitHandle = ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(1235), + exitCode: Effect.fail(signalExitError), + isRunning: Effect.succeed(false), + kill: () => Effect.void, + unref: Effect.succeed(Effect.void), + stdin: Sink.drain, + stdout: Stream.empty, + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + }); + const handles = [cancelledHandle, unexpectedExitHandle]; const spawner = ChildProcessSpawner.make((command) => Effect.sync(() => { spawnedCommand = command; + const handle = handles.shift(); + if (!handle) throw new Error("Unexpected managed Codex spawn"); return handle; }), ); @@ -71,7 +98,7 @@ describe("ManagedCodexExec", () => { Effect.sync(() => events.push(event)).pipe( Effect.andThen( event.type === "task.completed" - ? Deferred.succeed(terminal, event).pipe(Effect.asVoid) + ? Queue.offer(terminals, event).pipe(Effect.asVoid) : Effect.void, ), ), @@ -128,10 +155,34 @@ describe("ManagedCodexExec", () => { yield* manager.cancel({ threadId: ThreadId.make("thread-1"), agentId: launched.agentId }), ).toEqual({ cancelled: true }); expect(killed).toBe(true); - expect(yield* Deferred.await(terminal)).toMatchObject({ + expect(yield* Queue.take(terminals)).toMatchObject({ type: "task.completed", payload: { taskId: launched.agentId, status: "stopped" }, }); + expect( + yield* Effect.result( + manager.cancel({ threadId: ThreadId.make("thread-1"), agentId: launched.agentId }), + ), + ).toMatchObject({ _tag: "Failure", failure: { reason: "run-not-found" } }); + + const unexpectedExit = yield* manager.launch({ + threadId: ThreadId.make("thread-1"), + prompt: "Review another implementation", + title: "Reviewer", + }); + expect(yield* Queue.take(terminals)).toMatchObject({ + type: "task.completed", + payload: { + taskId: unexpectedExit.agentId, + status: "failed", + summary: "Managed Codex exec failed before reporting an exit code", + }, + }); + expect( + yield* Effect.result( + manager.cancel({ threadId: ThreadId.make("thread-1"), agentId: unexpectedExit.agentId }), + ), + ).toMatchObject({ _tag: "Failure", failure: { reason: "run-not-found" } }); }).pipe(Effect.scoped), ); }); diff --git a/apps/server/src/orchestration/ManagedCodexExec.ts b/apps/server/src/orchestration/ManagedCodexExec.ts index da8db82c987f..7508aaecfca7 100644 --- a/apps/server/src/orchestration/ManagedCodexExec.ts +++ b/apps/server/src/orchestration/ManagedCodexExec.ts @@ -15,6 +15,7 @@ import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Result from "effect/Result"; import * as Scope from "effect/Scope"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; @@ -236,16 +237,22 @@ export const layer = Layer.effect( yield* Effect.all([reportLines(child.stdout), reportLines(child.stderr)], { concurrency: "unbounded", }); - const exitCode = Number(yield* child.exitCode); + const exitResult = yield* Effect.result(child.exitCode); runs.delete(agentId); yield* emit(input.threadId, "task.completed", { ...linkage, - status: run.cancelled ? "stopped" : exitCode === 0 ? "completed" : "failed", + status: run.cancelled + ? "stopped" + : Result.isSuccess(exitResult) && Number(exitResult.success) === 0 + ? "completed" + : "failed", summary: run.cancelled ? "Cancelled by T3" - : exitCode === 0 - ? "Managed Codex exec completed" - : `Managed Codex exec exited with code ${exitCode}`, + : Result.isFailure(exitResult) + ? "Managed Codex exec failed before reporting an exit code" + : Number(exitResult.success) === 0 + ? "Managed Codex exec completed" + : `Managed Codex exec exited with code ${Number(exitResult.success)}`, }); }), scope, From 17db9a43c2bc292cc59047dd52d375320460f87d Mon Sep 17 00:00:00 2001 From: Juliusicon Date: Thu, 13 Aug 2026 02:47:31 +0200 Subject: [PATCH 3/7] fix(agents): preserve managed lifecycle identity --- .../orchestration/ManagedCodexExec.test.ts | 51 +++++++- .../src/orchestration/ManagedCodexExec.ts | 49 +++++--- .../src/provider/Layers/CodexAdapter.test.ts | 42 +++++++ .../src/provider/Layers/CodexAdapter.ts | 12 +- .../CodexCollabRuntime.integration.test.ts | 18 ++- .../Layers/CodexSessionRuntime.test.ts | 22 ++++ .../provider/Layers/CodexSessionRuntime.ts | 110 +++++++++++++----- .../codexCollabSpawnMetadata.json | 20 ++++ .../src/state/subagentRuntime.test.ts | 25 ++++ 9 files changed, 298 insertions(+), 51 deletions(-) create mode 100644 apps/server/src/provider/testFixtures/codexCollabSpawnMetadata.json diff --git a/apps/server/src/orchestration/ManagedCodexExec.test.ts b/apps/server/src/orchestration/ManagedCodexExec.test.ts index fd21bd15aac3..1ba08e25c743 100644 --- a/apps/server/src/orchestration/ManagedCodexExec.test.ts +++ b/apps/server/src/orchestration/ManagedCodexExec.test.ts @@ -38,6 +38,12 @@ describe("ManagedCodexExec", () => { method: "exitCode", description: "Process interrupted due to receipt of signal: 'SIGTERM'", }); + const failedKillError = PlatformError.systemError({ + _tag: "Unknown", + module: "ChildProcess", + method: "kill", + description: "Failed to kill child process", + }); const cancelledHandle = ChildProcessSpawner.makeHandle({ pid: ChildProcessSpawner.ProcessId(1234), exitCode: Deferred.await(exit), @@ -68,7 +74,28 @@ describe("ManagedCodexExec", () => { getInputFd: () => Sink.drain, getOutputFd: () => Stream.empty, }); - const handles = [cancelledHandle, unexpectedExitHandle]; + const failedKillExit = yield* Deferred.make< + ChildProcessSpawner.ExitCode, + PlatformError.PlatformError + >(); + const failedKillHandle = ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(1236), + exitCode: Deferred.await(failedKillExit), + isRunning: Effect.succeed(true), + kill: () => + Effect.gen(function* () { + yield* Deferred.fail(failedKillExit, signalExitError); + return yield* failedKillError; + }), + unref: Effect.succeed(Effect.void), + stdin: Sink.drain, + stdout: Stream.empty, + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + }); + const handles = [cancelledHandle, unexpectedExitHandle, failedKillHandle]; const spawner = ChildProcessSpawner.make((command) => Effect.sync(() => { spawnedCommand = command; @@ -183,6 +210,28 @@ describe("ManagedCodexExec", () => { manager.cancel({ threadId: ThreadId.make("thread-1"), agentId: unexpectedExit.agentId }), ), ).toMatchObject({ _tag: "Failure", failure: { reason: "run-not-found" } }); + + const failedCancellation = yield* manager.launch({ + threadId: ThreadId.make("thread-1"), + prompt: "Review cancellation behavior", + title: "Reviewer", + }); + expect( + yield* Effect.result( + manager.cancel({ + threadId: ThreadId.make("thread-1"), + agentId: failedCancellation.agentId, + }), + ), + ).toMatchObject({ _tag: "Failure", failure: { reason: "not-owned" } }); + expect(yield* Queue.take(terminals)).toMatchObject({ + type: "task.completed", + payload: { + taskId: failedCancellation.agentId, + status: "failed", + summary: "Managed Codex exec failed before reporting an exit code", + }, + }); }).pipe(Effect.scoped), ); }); diff --git a/apps/server/src/orchestration/ManagedCodexExec.ts b/apps/server/src/orchestration/ManagedCodexExec.ts index 7508aaecfca7..a1a9bd1a5546 100644 --- a/apps/server/src/orchestration/ManagedCodexExec.ts +++ b/apps/server/src/orchestration/ManagedCodexExec.ts @@ -18,6 +18,7 @@ import * as Option from "effect/Option"; import * as Result from "effect/Result"; import * as Scope from "effect/Scope"; import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; @@ -27,6 +28,7 @@ import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts"; interface ManagedRun { readonly threadId: string; readonly child: ChildProcessSpawner.ChildProcessHandle; + readonly cancellationLock: Semaphore.Semaphore; cancelled: boolean; } @@ -178,7 +180,13 @@ export const layer = Layer.effect( ), ); - const run: ManagedRun = { threadId: input.threadId, child, cancelled: false }; + const cancellationLock = yield* Semaphore.make(1); + const run: ManagedRun = { + threadId: input.threadId, + child, + cancellationLock, + cancelled: false, + }; runs.set(agentId, run); const linkage = { taskId: agentId, @@ -238,15 +246,20 @@ export const layer = Layer.effect( concurrency: "unbounded", }); const exitResult = yield* Effect.result(child.exitCode); - runs.delete(agentId); + const cancelled = yield* run.cancellationLock.withPermit( + Effect.sync(() => { + runs.delete(agentId); + return run.cancelled; + }), + ); yield* emit(input.threadId, "task.completed", { ...linkage, - status: run.cancelled + status: cancelled ? "stopped" : Result.isSuccess(exitResult) && Number(exitResult.success) === 0 ? "completed" : "failed", - summary: run.cancelled + summary: cancelled ? "Cancelled by T3" : Result.isFailure(exitResult) ? "Managed Codex exec failed before reporting an exit code" @@ -276,15 +289,25 @@ export const layer = Layer.effect( message: `Managed agent ${input.agentId} does not belong to thread ${input.threadId}.`, }); } - run.cancelled = true; - yield* run.child.kill({ killSignal: "SIGTERM", forceKillAfter: "3 seconds" }).pipe( - Effect.mapError( - () => - new ManagedAgentRunError({ - reason: "not-owned", - message: `Managed agent ${input.agentId} could not be cancelled.`, - }), - ), + yield* run.cancellationLock.withPermit( + Effect.gen(function* () { + if (runs.get(input.agentId) !== run) { + return yield* new ManagedAgentRunError({ + reason: "run-not-found", + message: `Managed agent ${input.agentId} is not running.`, + }); + } + yield* run.child.kill({ killSignal: "SIGTERM", forceKillAfter: "3 seconds" }).pipe( + Effect.mapError( + () => + new ManagedAgentRunError({ + reason: "not-owned", + message: `Managed agent ${input.agentId} could not be cancelled.`, + }), + ), + ); + run.cancelled = true; + }), ); return { cancelled: true }; }, diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 7b8fbec56660..bf6c1d6d0636 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -512,6 +512,48 @@ function startLifecycleRuntime() { } lifecycleLayer("CodexAdapterLive lifecycle", (it) => { + it.effect("keeps native parent and model metadata on retained terminal agent rows", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-collab-child-closed"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "collabAgent/closed", + threadId: asThreadId("thread-1"), + payload: { + agentThreadId: "native-child", + parentThreadId: "native-parent", + nickname: "reviewer", + role: "reviewer", + model: "gpt-5.6-luna", + effort: "low", + }, + }); + const firstEvent = yield* Fiber.join(firstEventFiber); + + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") return; + NodeAssert.equal(firstEvent.value.type, "task.updated"); + if (firstEvent.value.type !== "task.updated") return; + NodeAssert.deepStrictEqual(firstEvent.value.payload, { + taskId: "native-child", + status: "interrupted", + role: "reviewer", + title: "reviewer", + model: "gpt-5.6-luna", + effort: "low", + parentAgentId: "native-parent", + timelineBypass: true, + agentSource: "provider", + cancellationOwner: "provider", + }); + }), + ); + it.effect("maps completed agent message items to canonical item.completed events", () => Effect.gen(function* () { const { adapter, runtime } = yield* startLifecycleRuntime(); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 0c4b25d6fab3..01a5a13b2df2 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -525,6 +525,8 @@ function mapCollabAgentEvent( (typeof payload.role === "string" ? payload.role : undefined) ?? pathLeaf ?? "general-purpose"; const model = typeof payload.model === "string" ? payload.model : undefined; const effort = typeof payload.effort === "string" ? payload.effort : undefined; + const parentAgentId = + typeof payload.parentThreadId === "string" ? payload.parentThreadId : undefined; // A bare thread id is not a name. Omitting the title lets the client fold // keep the real one from task.started instead of clobbering it (probe // finding: progress rows renamed math_one to its UUID). @@ -539,6 +541,7 @@ function mapCollabAgentEvent( ...(agentPath ? { agentPath } : {}), ...(model ? { model } : {}), ...(effort ? { effort } : {}), + ...(parentAgentId ? { parentAgentId } : {}), timelineBypass: true, agentSource: "provider" as const, cancellationOwner: "provider" as const, @@ -555,9 +558,6 @@ function mapCollabAgentEvent( description: title, title, ...statusLinkage, - ...(typeof payload.parentThreadId === "string" - ? { parentAgentId: payload.parentThreadId } - : {}), }, }, ]; @@ -709,9 +709,8 @@ function mapCollabAgentEvent( payload: { taskId, description: title, - ...(knownName ? { title: knownName } : {}), + ...statusLinkage, typedUsage, - timelineBypass: true, }, }, ]; @@ -741,9 +740,8 @@ function mapCollabAgentEvent( payload: { taskId, description: title, - ...(knownName ? { title: knownName } : {}), + ...statusLinkage, summary, - timelineBypass: true, }, }, ]; diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 38e0e0a7b2c1..e193497b7615 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -20,6 +20,7 @@ import * as Stream from "effect/Stream"; import { assert, describe } from "vite-plus/test"; import wireFixture from "../testFixtures/codexMultiAgentWire.json" with { type: "json" }; +import spawnMetadataFixture from "../testFixtures/codexCollabSpawnMetadata.json" with { type: "json" }; import { makeCodexSessionRuntime } from "./CodexSessionRuntime.ts"; const ROOT = wireFixture.rootThreadId; @@ -66,7 +67,11 @@ function buildScript() { ]; return { rootThreadId: ROOT, - notifications: [...captured.filter((entry) => entry.method !== "turn/completed"), ...extras], + notifications: [ + spawnMetadataFixture, + ...captured.filter((entry) => entry.method !== "turn/completed"), + ...extras, + ], }; } @@ -115,6 +120,12 @@ describe("CodexSessionRuntime collab integration", () => { (event.payload as { agentThreadId?: string }).agentThreadId === CHILD_A, ); assert.isDefined(childTurnCompleted, "child A's turn completion becomes an agent event"); + assert.deepInclude(childTurnCompleted.payload, { + agentThreadId: CHILD_A, + parentThreadId: ROOT, + model: "gpt-5.6-luna", + effort: "low", + }); const childClosed = events.find( (event) => @@ -122,6 +133,11 @@ describe("CodexSessionRuntime collab integration", () => { (event.payload as { agentThreadId?: string }).agentThreadId === CHILD_B, ); assert.isDefined(childClosed, "child B's close becomes an agent event"); + assert.equal( + (childClosed.payload as { parentThreadId?: string }).parentThreadId, + ROOT, + "retained terminal lifecycle keeps the native parent identity", + ); // Parent-owned resolution passes through โ€” not swallowed, not // re-labelled as an agent event. diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index d7346a0e0dbe..51f76fac0334 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -7,6 +7,7 @@ import { describe } from "vite-plus/test"; import { DEFAULT_MODEL, ThreadId } from "@t3tools/contracts"; import * as CodexErrors from "effect-codex-app-server/errors"; import * as CodexRpc from "effect-codex-app-server/rpc"; +import spawnMetadataFixture from "../testFixtures/codexCollabSpawnMetadata.json" with { type: "json" }; import { buildCodexDeveloperInstructions, @@ -19,6 +20,7 @@ import { hasConfiguredMcpServer, isRecoverableThreadResumeError, openCodexThread, + readCollabSpawnMetadata, } from "./CodexSessionRuntime.ts"; const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); @@ -39,6 +41,26 @@ describe("CodexSessionRuntimeIdentifierGenerationError", () => { }); }); +describe("readCollabSpawnMetadata", () => { + it("reads the requested model and effort from the correlated spawn tool call", () => { + NodeAssert.deepStrictEqual( + readCollabSpawnMetadata( + spawnMetadataFixture as unknown as Parameters[0], + ), + [ + [ + "019fcfd6-2883-77e0-9013-4410ede70371", + { + parentThreadId: "019fcfd6-17bb-72f0-ae12-a1f2dee6e3e5", + model: "gpt-5.6-luna", + effort: "low", + }, + ], + ], + ); + }); +}); + function makeThreadOpenResponse( threadId: string, ): CodexRpc.ClientRequestResponsesByMethod["thread/start"] { diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 13ecc8451b4a..583059489fa4 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -635,6 +635,12 @@ interface CollabChildAgentState { readonly spawnTurnId: TurnId | undefined; } +interface CollabSpawnMetadata { + readonly parentThreadId: string; + readonly model: string | undefined; + readonly effort: string | undefined; +} + function readThreadSpawnSource(thread: { readonly source: unknown }): | { nickname: string | undefined; @@ -642,8 +648,6 @@ function readThreadSpawnSource(thread: { readonly source: unknown }): agentPath: string | undefined; depth: number | undefined; parentThreadId: string | undefined; - model: string | undefined; - effort: string | undefined; } | undefined { const source = thread.source; @@ -659,7 +663,6 @@ function readThreadSpawnSource(thread: { readonly source: unknown }): return undefined; } const record = spawn as Record; - const threadRecord = thread as Record; return { nickname: typeof record.agent_nickname === "string" ? record.agent_nickname : undefined, role: typeof record.agent_role === "string" ? record.agent_role : undefined, @@ -667,9 +670,38 @@ function readThreadSpawnSource(thread: { readonly source: unknown }): depth: typeof record.depth === "number" ? record.depth : undefined, parentThreadId: typeof record.parent_thread_id === "string" ? record.parent_thread_id : undefined, - model: typeof threadRecord.model === "string" ? threadRecord.model : undefined, - effort: - typeof threadRecord.reasoningEffort === "string" ? threadRecord.reasoningEffort : undefined, + }; +} + +export function readCollabSpawnMetadata( + notification: CodexServerNotification, +): ReadonlyArray { + if (notification.method !== "item/started" && notification.method !== "item/completed") { + return []; + } + const item = notification.params.item; + if (item.type !== "collabAgentToolCall" || item.tool !== "spawnAgent") { + return []; + } + return item.receiverThreadIds.map((receiverThreadId) => [ + receiverThreadId, + { + parentThreadId: item.senderThreadId, + model: typeof item.model === "string" ? item.model : undefined, + effort: typeof item.reasoningEffort === "string" ? item.reasoningEffort : undefined, + }, + ]); +} + +function collabChildIdentity(child: CollabChildAgentState) { + return { + agentThreadId: child.agentThreadId, + ...(child.nickname ? { nickname: child.nickname } : {}), + ...(child.role ? { role: child.role } : {}), + ...(child.agentPath ? { agentPath: child.agentPath } : {}), + ...(child.parentThreadId ? { parentThreadId: child.parentThreadId } : {}), + ...(child.model ? { model: child.model } : {}), + ...(child.effort ? { effort: child.effort } : {}), }; } @@ -862,6 +894,7 @@ export const makeCodexSessionRuntime = ( const approvalCorrelationsRef = yield* Ref.make(new Map()); const pendingUserInputsRef = yield* Ref.make(new Map()); const collabReceiverTurnsRef = yield* Ref.make(new Map()); + const collabSpawnMetadataRef = yield* Ref.make(new Map()); const collabChildAgentsRef = yield* Ref.make(new Map()); /** Child provider-thread id โ†’ its currently running provider turn id. */ const collabChildLiveTurnsRef = yield* Ref.make(new Map()); @@ -1005,6 +1038,7 @@ export const makeCodexSessionRuntime = ( // child onto a new fleet's CTA (review finding). Only a genuinely // new registration captures the current turn. const existingChild = (yield* Ref.get(collabChildAgentsRef)).get(thread.id); + const correlatedSpawn = (yield* Ref.get(collabSpawnMetadataRef)).get(thread.id); const spawnTurnId = existingChild ? existingChild.spawnTurnId : ((yield* Ref.get(sessionRef)).activeTurnId ?? undefined); @@ -1015,9 +1049,12 @@ export const makeCodexSessionRuntime = ( agentPath: spawn.agentPath ?? existingChild?.agentPath, depth: spawn.depth ?? existingChild?.depth, parentThreadId: - spawn.parentThreadId ?? thread.parentThreadId ?? existingChild?.parentThreadId, - model: spawn.model ?? existingChild?.model, - effort: spawn.effort ?? existingChild?.effort, + spawn.parentThreadId ?? + thread.parentThreadId ?? + existingChild?.parentThreadId ?? + correlatedSpawn?.parentThreadId, + model: existingChild?.model ?? correlatedSpawn?.model, + effort: existingChild?.effort ?? correlatedSpawn?.effort, spawnTurnId, }; yield* Ref.update(collabChildAgentsRef, (current) => { @@ -1031,14 +1068,8 @@ export const makeCodexSessionRuntime = ( method: "collabAgent/started", ...(state.spawnTurnId ? { turnId: state.spawnTurnId } : {}), payload: { - agentThreadId: state.agentThreadId, - ...(state.nickname ? { nickname: state.nickname } : {}), - ...(state.role ? { role: state.role } : {}), - ...(state.agentPath ? { agentPath: state.agentPath } : {}), + ...collabChildIdentity(state), ...(state.depth !== undefined ? { depth: state.depth } : {}), - ...(state.parentThreadId ? { parentThreadId: state.parentThreadId } : {}), - ...(state.model ? { model: state.model } : {}), - ...(state.effort ? { effort: state.effort } : {}), }, }); return true; @@ -1066,6 +1097,7 @@ export const makeCodexSessionRuntime = ( return false; } const activitySpawnTurnId = (yield* Ref.get(sessionRef)).activeTurnId ?? undefined; + const correlatedSpawn = (yield* Ref.get(collabSpawnMetadataRef)).get(item.agentThreadId); yield* Ref.update(collabChildAgentsRef, (current) => { const existing = current.get(item.agentThreadId); const next = new Map(current); @@ -1084,9 +1116,12 @@ export const makeCodexSessionRuntime = ( role: existing?.role, agentPath: existing?.agentPath ?? item.agentPath, depth: existing?.depth, - parentThreadId: existing?.parentThreadId, - model: existing?.model, - effort: existing?.effort, + parentThreadId: + existing?.parentThreadId ?? + correlatedSpawn?.parentThreadId ?? + notification.params.threadId, + model: existing?.model ?? correlatedSpawn?.model, + effort: existing?.effort ?? correlatedSpawn?.effort, spawnTurnId: existing ? existing.spawnTurnId : activitySpawnTurnId, }); return next; @@ -1098,8 +1133,9 @@ export const makeCodexSessionRuntime = ( method: "collabAgent/activity", ...(registeredChild?.spawnTurnId ? { turnId: registeredChild.spawnTurnId } : {}), payload: { - agentThreadId: item.agentThreadId, - agentPath: item.agentPath, + ...(registeredChild + ? collabChildIdentity(registeredChild) + : { agentThreadId: item.agentThreadId, agentPath: item.agentPath }), activityKind: item.kind, }, }); @@ -1123,14 +1159,7 @@ export const makeCodexSessionRuntime = ( if (!child) { return false; } - const childIdentity = { - agentThreadId: child.agentThreadId, - ...(child.nickname ? { nickname: child.nickname } : {}), - ...(child.role ? { role: child.role } : {}), - ...(child.agentPath ? { agentPath: child.agentPath } : {}), - ...(child.model ? { model: child.model } : {}), - ...(child.effort ? { effort: child.effort } : {}), - }; + const childIdentity = collabChildIdentity(child); switch (notification.method) { case "turn/started": { const childTurnId = @@ -1268,6 +1297,7 @@ export const makeCodexSessionRuntime = ( const payload = notification.params; const route = readRouteFields(notification); const collabReceiverTurns = yield* Ref.get(collabReceiverTurnsRef); + const collabSpawnMetadata = yield* Ref.get(collabSpawnMetadataRef); const childParentTurnId = (() => { const providerConversationId = readNotificationThreadId(notification); return providerConversationId @@ -1276,6 +1306,28 @@ export const makeCodexSessionRuntime = ( })(); rememberCollabReceiverTurns(collabReceiverTurns, notification, route.turnId); + const spawnMetadataEntries = readCollabSpawnMetadata(notification); + for (const [childThreadId, metadata] of spawnMetadataEntries) { + collabSpawnMetadata.set(childThreadId, metadata); + } + yield* Ref.set(collabSpawnMetadataRef, collabSpawnMetadata); + if (spawnMetadataEntries.length > 0) { + yield* Ref.update(collabChildAgentsRef, (current) => { + const next = new Map(current); + for (const [childThreadId, metadata] of spawnMetadataEntries) { + const child = next.get(childThreadId); + if (child) { + next.set(childThreadId, { + ...child, + parentThreadId: child.parentThreadId ?? metadata.parentThreadId, + model: child.model ?? metadata.model, + effort: child.effort ?? metadata.effort, + }); + } + } + return next; + }); + } // Interception FIRST: a registered v2 child is usually also in the // receiver-turn map (collabAgentToolCall.receiverThreadIds), and the // legacy suppressor below would drop its lifecycle before it could diff --git a/apps/server/src/provider/testFixtures/codexCollabSpawnMetadata.json b/apps/server/src/provider/testFixtures/codexCollabSpawnMetadata.json new file mode 100644 index 000000000000..a60f400cc0dc --- /dev/null +++ b/apps/server/src/provider/testFixtures/codexCollabSpawnMetadata.json @@ -0,0 +1,20 @@ +{ + "method": "item/completed", + "params": { + "item": { + "type": "collabAgentToolCall", + "id": "call_fixture_spawn_alpha", + "tool": "spawnAgent", + "status": "completed", + "senderThreadId": "019fcfd6-17bb-72f0-ae12-a1f2dee6e3e5", + "receiverThreadIds": ["019fcfd6-2883-77e0-9013-4410ede70371"], + "prompt": "Solve the alpha subtask", + "model": "gpt-5.6-luna", + "reasoningEffort": "low", + "agentsStates": {} + }, + "threadId": "019fcfd6-17bb-72f0-ae12-a1f2dee6e3e5", + "turnId": "019fcfd6-1806-7de1-8564-de69fd55bffb", + "completedAtMs": 1785898346000 + } +} diff --git a/packages/client-runtime/src/state/subagentRuntime.test.ts b/packages/client-runtime/src/state/subagentRuntime.test.ts index a6f1dcb2caea..402c3e592ec7 100644 --- a/packages/client-runtime/src/state/subagentRuntime.test.ts +++ b/packages/client-runtime/src/state/subagentRuntime.test.ts @@ -508,6 +508,31 @@ describe("deriveAgentPanelModel", () => { "managed-grandchild", ]); }); + + it("reconstructs native hierarchy when retention kept only a terminal child row", () => { + const retained = fold([ + activity("task.started", { taskId: "native-parent", title: "Parent" }), + activity("task.updated", { + taskId: "native-child", + title: "Reviewer", + role: "reviewer", + parentAgentId: "native-parent", + status: "interrupted", + agentSource: "provider", + cancellationOwner: "provider", + }), + ]); + + const model = deriveAgentPanelModel({ agents: retained }); + expect(model.directAgents.map((agent) => agent.id)).toEqual(["native-parent"]); + expect(model.childrenByParentId.get("native-parent")?.map((agent) => agent.id)).toEqual([ + "native-child", + ]); + expect(retained.find((agent) => agent.id === "native-child")).toMatchObject({ + parentAgentId: "native-parent", + status: "interrupted", + }); + }); }); describe("workflowCardMembers", () => { From 865817f3327bb36ff2fa79ac375716dae686fb5f Mon Sep 17 00:00:00 2001 From: Juliusicon Date: Thu, 13 Aug 2026 03:05:35 +0200 Subject: [PATCH 4/7] fix(agents): persist late Codex spawn metadata --- .../ProviderRuntimeIngestion.activity.test.ts | 40 +++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 9 ++- .../src/provider/Layers/CodexAdapter.test.ts | 43 ++++++++++++++ .../src/provider/Layers/CodexAdapter.ts | 22 +++++++ .../CodexCollabRuntime.integration.test.ts | 25 +++++++- .../provider/Layers/CodexSessionRuntime.ts | 58 +++++++++++++------ .../src/state/subagentRuntime.test.ts | 28 +++++++++ 7 files changed, 205 insertions(+), 20 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.activity.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.activity.test.ts index 936041038644..fbd82ebea267 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.activity.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.activity.test.ts @@ -52,6 +52,46 @@ describe("runtimeEventToActivities task progress", () => { expect(usagePayload?.usageSnapshot).toBe(true); }); + it("upserts late Codex child identity separately from lifecycle state", () => { + const event = { + ...base, + type: "task.updated", + eventId: EventId.make("evt-late-metadata"), + payload: { + taskId: RuntimeTaskId.make("native-child"), + parentAgentId: "native-parent", + model: "gpt-5.6-luna", + effort: "low", + timelineBypass: true, + agentSource: "provider", + cancellationOwner: "provider", + }, + raw: { + source: "codex.app-server.notification", + method: "collabAgent/metadata", + payload: {}, + }, + } satisfies ProviderRuntimeEvent; + + const [activity] = runtimeEventToActivities(event); + const [duplicateActivity] = runtimeEventToActivities({ + ...event, + eventId: EventId.make("evt-late-metadata-duplicate"), + createdAt: "2026-08-06T00:00:01.000Z", + }); + + expect(activity?.id).toBe("task-identity:thread-1:native-child"); + expect(duplicateActivity?.id).toBe(activity?.id); + expect(activity?.kind).toBe("task.updated"); + expect(activity?.payload).toMatchObject({ + taskId: "native-child", + parentAgentId: "native-parent", + model: "gpt-5.6-luna", + effort: "low", + }); + expect(activity?.payload).not.toHaveProperty("status"); + }); + it("splits combined progress and usage into their independent snapshots", () => { const event = { ...base, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 547ba8af649f..a9eddeec61f0 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -644,9 +644,16 @@ export function runtimeEventToActivities( } case "task.updated": { + const activityId = + event.raw?.method === "collabAgent/metadata" + ? EventId.make(`task-identity:${event.threadId}:${event.payload.taskId}`) + : event.eventId; return [ { - id: event.eventId, + // Provider spawn metadata may arrive after terminal lifecycle and + // may be repeated by item/started + item/completed. Keep one + // independently retained identity snapshot per native child. + id: activityId, createdAt: event.createdAt, tone: event.payload.status === "failed" ? "error" : "info", kind: "task.updated", diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index bf6c1d6d0636..f1c329f6151e 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -554,6 +554,49 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { }), ); + it.effect("maps late native spawn metadata to a correlated identity-only patch", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-collab-child-metadata"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "collabAgent/metadata", + threadId: asThreadId("thread-1"), + turnId: asTurnId("spawn-turn"), + itemId: asItemId("spawn-call"), + payload: { + agentThreadId: "native-child", + parentThreadId: "native-parent", + model: "gpt-5.6-luna", + effort: "low", + }, + }); + const firstEvent = yield* Fiber.join(firstEventFiber); + + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") return; + NodeAssert.equal(firstEvent.value.type, "task.updated"); + if (firstEvent.value.type !== "task.updated") return; + NodeAssert.deepStrictEqual(firstEvent.value.payload, { + taskId: "native-child", + model: "gpt-5.6-luna", + effort: "low", + parentAgentId: "native-parent", + timelineBypass: true, + agentSource: "provider", + cancellationOwner: "provider", + }); + NodeAssert.deepStrictEqual(firstEvent.value.providerRefs, { + providerTurnId: "spawn-turn", + providerItemId: "spawn-call", + }); + }), + ); + it.effect("maps completed agent message items to canonical item.completed events", () => Effect.gen(function* () { const { adapter, runtime } = yield* startLifecycleRuntime(); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 01a5a13b2df2..73184ea2451a 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -548,6 +548,28 @@ function mapCollabAgentEvent( } as const; switch (event.method) { + case "collabAgent/metadata": + // Late spawn metadata is an identity-only patch. Do not synthesize a + // status (which could reopen a terminal child), or fallback identity + // fields that the correlated spawn call did not actually provide. + return [ + { + ...base, + type: "task.updated", + payload: { + taskId, + ...(nickname ? { title: nickname } : pathLeaf ? { title: pathLeaf } : {}), + ...(typeof payload.role === "string" ? { role: payload.role } : {}), + ...(agentPath ? { agentPath } : {}), + ...(model ? { model } : {}), + ...(effort ? { effort } : {}), + ...(parentAgentId ? { parentAgentId } : {}), + timelineBypass: true, + agentSource: "provider", + cancellationOwner: "provider", + }, + }, + ]; case "collabAgent/started": return [ { diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index e193497b7615..7fee9027295f 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -68,9 +68,10 @@ function buildScript() { return { rootThreadId: ROOT, notifications: [ - spawnMetadataFixture, ...captured.filter((entry) => entry.method !== "turn/completed"), ...extras, + // Actual spawn metadata can trail the child's terminal lifecycle. + spawnMetadataFixture, ], }; } @@ -120,12 +121,32 @@ describe("CodexSessionRuntime collab integration", () => { (event.payload as { agentThreadId?: string }).agentThreadId === CHILD_A, ); assert.isDefined(childTurnCompleted, "child A's turn completion becomes an agent event"); - assert.deepInclude(childTurnCompleted.payload, { + assert.deepInclude(childTurnCompleted.payload, { agentThreadId: CHILD_A }); + assert.notProperty( + childTurnCompleted.payload as Record, + "model", + "terminal lifecycle must not invent metadata that has not arrived yet", + ); + + const metadataEnrichment = events.find( + (event) => + event.method === "collabAgent/metadata" && + (event.payload as { agentThreadId?: string }).agentThreadId === CHILD_A, + ); + assert.isDefined(metadataEnrichment, "late spawn metadata becomes an emitted event"); + assert.deepInclude(metadataEnrichment.payload, { agentThreadId: CHILD_A, parentThreadId: ROOT, model: "gpt-5.6-luna", effort: "low", }); + assert.equal(metadataEnrichment.turnId, spawnMetadataFixture.params.turnId); + assert.equal(metadataEnrichment.itemId, spawnMetadataFixture.params.item.id); + assert.isAbove( + events.indexOf(metadataEnrichment), + events.indexOf(childTurnCompleted), + "metadata enrichment preserves its actual after-terminal wire ordering", + ); const childClosed = events.find( (event) => diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 583059489fa4..7600509840df 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -1311,23 +1311,30 @@ export const makeCodexSessionRuntime = ( collabSpawnMetadata.set(childThreadId, metadata); } yield* Ref.set(collabSpawnMetadataRef, collabSpawnMetadata); - if (spawnMetadataEntries.length > 0) { - yield* Ref.update(collabChildAgentsRef, (current) => { - const next = new Map(current); - for (const [childThreadId, metadata] of spawnMetadataEntries) { - const child = next.get(childThreadId); - if (child) { - next.set(childThreadId, { - ...child, - parentThreadId: child.parentThreadId ?? metadata.parentThreadId, - model: child.model ?? metadata.model, - effort: child.effort ?? metadata.effort, - }); - } - } - return next; - }); - } + const enrichedChildren = + spawnMetadataEntries.length > 0 + ? yield* Ref.modify(collabChildAgentsRef, (current) => { + const enriched: Array = []; + const next = new Map(current); + for (const [childThreadId, metadata] of spawnMetadataEntries) { + const child = next.get(childThreadId); + if (child) { + const enrichedChild = { + ...child, + // The correlated spawn tool call is authoritative; the + // activity registration's parent can be only a routing + // fallback captured before actual metadata arrived. + parentThreadId: metadata.parentThreadId, + model: child.model ?? metadata.model, + effort: child.effort ?? metadata.effort, + }; + next.set(childThreadId, enrichedChild); + enriched.push(enrichedChild); + } + } + return [enriched, next]; + }) + : []; // Interception FIRST: a registered v2 child is usually also in the // receiver-turn map (collabAgentToolCall.receiverThreadIds), and the // legacy suppressor below would drop its lifecycle before it could @@ -1434,6 +1441,23 @@ export const makeCodexSessionRuntime = ( : {}), ...(payload !== undefined ? { payload } : {}), }); + // The correlated tool-call itself stays first in the provider stream. + // A separate identity patch then makes late spawn metadata durable: + // ingestion upserts it by canonical thread + child id, while replay + // folds it without changing an already-terminal lifecycle status. + yield* Effect.forEach( + enrichedChildren, + (enrichedChild) => + emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "collabAgent/metadata", + ...(route.turnId ? { turnId: route.turnId } : {}), + ...(route.itemId ? { itemId: route.itemId } : {}), + payload: collabChildIdentity(enrichedChild), + }), + { discard: true }, + ); }); const currentSessionProviderThreadId = Effect.map(Ref.get(sessionRef), currentProviderThreadId); diff --git a/packages/client-runtime/src/state/subagentRuntime.test.ts b/packages/client-runtime/src/state/subagentRuntime.test.ts index 402c3e592ec7..4a7acbd2c19c 100644 --- a/packages/client-runtime/src/state/subagentRuntime.test.ts +++ b/packages/client-runtime/src/state/subagentRuntime.test.ts @@ -745,6 +745,34 @@ describe("terminal robustness", () => { expect(agents[0]!.title).toBe("Late"); }); + it("late metadata enriches a retained terminal row without reopening it", () => { + const agents = fold([ + activity("task.updated", { + taskId: "native-child", + status: "interrupted", + timelineBypass: true, + agentSource: "provider", + }), + activity("task.updated", { + taskId: "native-child", + parentAgentId: "native-parent", + model: "gpt-5.6-luna", + effort: "low", + timelineBypass: true, + agentSource: "provider", + }), + ]); + + expect(agents).toHaveLength(1); + expect(agents[0]).toMatchObject({ + id: "native-child", + status: "interrupted", + parentAgentId: "native-parent", + model: "gpt-5.6-luna", + effort: "low", + }); + }); + it("a completion after a terminal task.updated still enriches result and usage", () => { // Claude commonly emits terminal task.updated before task.completed; // the completion carries the summary and final usage the update lacked. From 4d1acfafd46742c599f9adabeaab7c7563868bef Mon Sep 17 00:00:00 2001 From: Juliusicon Date: Thu, 13 Aug 2026 04:20:26 +0200 Subject: [PATCH 5/7] fix(server): harden managed Codex exec service --- .../Layers/OrchestrationReactor.test.ts | 1 + .../orchestration/ManagedCodexExec.test.ts | 83 ++- .../src/orchestration/ManagedCodexExec.ts | 493 +++++++++--------- .../Services/ProviderRuntimeIngestion.ts | 2 +- apps/server/src/server.test.ts | 22 +- apps/server/src/ws.ts | 2 +- packages/contracts/src/orchestration.ts | 11 +- 7 files changed, 356 insertions(+), 258 deletions(-) diff --git a/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts b/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts index 300d1526bb9a..62e6d7f7e358 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts @@ -35,6 +35,7 @@ describe("OrchestrationReactor", () => { return Effect.void; }, drain: Effect.void, + ingestRuntimeEvent: () => Effect.void, }), ), Layer.provideMerge( diff --git a/apps/server/src/orchestration/ManagedCodexExec.test.ts b/apps/server/src/orchestration/ManagedCodexExec.test.ts index 1ba08e25c743..190d284e0158 100644 --- a/apps/server/src/orchestration/ManagedCodexExec.test.ts +++ b/apps/server/src/orchestration/ManagedCodexExec.test.ts @@ -1,12 +1,18 @@ import { describe, expect, it } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; -import { ProjectId, ThreadId, type ProviderRuntimeEvent } from "@t3tools/contracts"; +import { + ManagedAgentRunError, + ProjectId, + ThreadId, + type ProviderRuntimeEvent, +} from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as PlatformError from "effect/PlatformError"; import * as Queue from "effect/Queue"; +import * as Result from "effect/Result"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; import { ChildProcessSpawner } from "effect/unstable/process"; @@ -22,6 +28,17 @@ import { } from "./Services/ProjectionSnapshotQuery.ts"; describe("ManagedCodexExec", () => { + it("exposes runtime ingestion as a layer composition requirement", () => { + type Requirements = Layer.Services; + const requiresRuntimeIngestion: unknown extends Requirements + ? false + : ProviderRuntimeIngestionService extends Requirements + ? true + : false = true; + + expect(requiresRuntimeIngestion).toBe(true); + }); + it.effect("terminalizes signal exits and releases owned process handles", () => Effect.gen(function* () { const exit = yield* Deferred.make< @@ -234,4 +251,68 @@ describe("ManagedCodexExec", () => { }); }).pipe(Effect.scoped), ); + + it.effect("keeps prompts out of spawn error messages while preserving the cause", () => + Effect.gen(function* () { + const prompt = "TOP_SECRET_MANAGED_CODEX_PROMPT"; + const spawnCause = PlatformError.systemError({ + _tag: "NotFound", + module: "ChildProcess", + method: "spawn", + pathOrDescriptor: `codex exec ${prompt}`, + }); + const spawner = ChildProcessSpawner.make(() => Effect.fail(spawnCause)); + const snapshots = { + getThreadDetailById: () => + Effect.succeed( + Option.some({ + id: ThreadId.make("thread-1"), + projectId: ProjectId.make("project-1"), + worktreePath: "D:/repo/worktree", + }), + ), + getProjectShellById: () => + Effect.succeed( + Option.some({ id: ProjectId.make("project-1"), workspaceRoot: "D:/repo" }), + ), + } as unknown as ProjectionSnapshotQueryShape; + const dependencies = Layer.mergeAll( + NodeServices.layer, + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), + Layer.succeed(ProjectionSnapshotQuery, snapshots), + Layer.succeed(ProviderRuntimeIngestionService, { + start: () => Effect.void, + drain: Effect.void, + ingestRuntimeEvent: () => Effect.void, + }), + ); + + const context = yield* Layer.build(layer.pipe(Layer.provide(dependencies))); + const manager = yield* ManagedCodexExec.pipe(Effect.provide(context)); + const result = yield* Effect.result( + manager.launch({ + threadId: ThreadId.make("thread-1"), + prompt, + title: "Reviewer", + }), + ); + + expect(Result.isFailure(result)).toBe(true); + if (Result.isSuccess(result)) { + return yield* Effect.die("Expected managed Codex spawn to fail"); + } + + const error = result.failure; + expect(error).toBeInstanceOf(ManagedAgentRunError); + expect(error).toMatchObject({ + reason: "spawn-failed", + threadId: ThreadId.make("thread-1"), + cause: spawnCause, + }); + expect(error.message).toBe("Managed agent run failed (spawn-failed): thread-1"); + expect(error.message).not.toContain(prompt); + expect(String(error)).not.toContain(prompt); + expect(error.cause).toBe(spawnCause); + }).pipe(Effect.scoped), + ); }); diff --git a/apps/server/src/orchestration/ManagedCodexExec.ts b/apps/server/src/orchestration/ManagedCodexExec.ts index a1a9bd1a5546..b356231cdd42 100644 --- a/apps/server/src/orchestration/ManagedCodexExec.ts +++ b/apps/server/src/orchestration/ManagedCodexExec.ts @@ -7,7 +7,6 @@ import { type ManagedCodexExecLaunchInput, type ProviderRuntimeEvent, } from "@t3tools/contracts"; -import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; @@ -32,18 +31,17 @@ interface ManagedRun { cancelled: boolean; } -export interface ManagedCodexExecShape { - readonly launch: ( - input: ManagedCodexExecLaunchInput, - ) => Effect.Effect<{ readonly agentId: string }, ManagedAgentRunError>; - readonly cancel: ( - input: ManagedAgentCancelInput, - ) => Effect.Effect<{ readonly cancelled: boolean }, ManagedAgentRunError>; -} - -export class ManagedCodexExec extends Context.Service()( - "t3/orchestration/ManagedCodexExec", -) {} +export class ManagedCodexExec extends Context.Service< + ManagedCodexExec, + { + readonly launch: ( + input: ManagedCodexExecLaunchInput, + ) => Effect.Effect<{ readonly agentId: string }, ManagedAgentRunError>; + readonly cancel: ( + input: ManagedAgentCancelInput, + ) => Effect.Effect<{ readonly cancelled: boolean }, ManagedAgentRunError>; + } +>()("t3/orchestration/ManagedCodexExec") {} function outputSummary(line: string): string | undefined { const trimmed = line.trim(); @@ -65,260 +63,263 @@ function outputSummary(line: string): string | undefined { return trimmed; } -export const layer = Layer.effect( - ManagedCodexExec, - Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const crypto = yield* Crypto.Crypto; - const scope = yield* Scope.Scope; - const snapshots = yield* ProjectionSnapshotQuery; - const ingestion = yield* Effect.serviceOption(ProviderRuntimeIngestionService); - const runs = new Map(); - const provider = ProviderDriverKind.make("codex"); +export const make = Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const crypto = yield* Crypto.Crypto; + const scope = yield* Scope.Scope; + const snapshots = yield* ProjectionSnapshotQuery; + const ingestion = yield* ProviderRuntimeIngestionService; + const runs = new Map(); + const provider = ProviderDriverKind.make("codex"); - const emit = ( - threadId: ManagedCodexExecLaunchInput["threadId"], - type: "task.started" | "task.progress" | "task.completed", - payload: ProviderRuntimeEvent["payload"], - ) => - Effect.gen(function* () { - const eventId = EventId.make(yield* crypto.randomUUIDv4); - const createdAt = DateTime.formatIso(yield* DateTime.now); - const ingestionService = Option.getOrUndefined(ingestion); - if (!ingestionService?.ingestRuntimeEvent) { - return yield* new ManagedAgentRunError({ - reason: "spawn-failed", - message: "Managed agent runtime ingestion is unavailable.", - }); - } - yield* ingestionService.ingestRuntimeEvent({ - eventId, - provider, - threadId, - createdAt, - type, - payload, - } as ProviderRuntimeEvent); - }); + const emit = ( + threadId: ManagedCodexExecLaunchInput["threadId"], + type: "task.started" | "task.progress" | "task.completed", + payload: ProviderRuntimeEvent["payload"], + ) => + Effect.gen(function* () { + const eventId = EventId.make(yield* crypto.randomUUIDv4); + const createdAt = DateTime.formatIso(yield* DateTime.now); + yield* ingestion.ingestRuntimeEvent({ + eventId, + provider, + threadId, + createdAt, + type, + payload, + } as ProviderRuntimeEvent); + }); - const launch: ManagedCodexExecShape["launch"] = Effect.fn("ManagedCodexExec.launch")( - function* (input) { - const threadOption = yield* snapshots.getThreadDetailById(input.threadId).pipe( - Effect.mapError( - () => - new ManagedAgentRunError({ - reason: "thread-not-found", - message: `Thread ${input.threadId} could not be loaded.`, - }), - ), - ); - const thread = Option.getOrUndefined(threadOption); - if (!thread) { - return yield* new ManagedAgentRunError({ - reason: "thread-not-found", - message: `Thread ${input.threadId} does not exist.`, - }); - } - const projectOption = yield* snapshots.getProjectShellById(thread.projectId).pipe( + const launch: ManagedCodexExec["Service"]["launch"] = Effect.fn("ManagedCodexExec.launch")( + function* (input) { + const threadOption = yield* snapshots.getThreadDetailById(input.threadId).pipe( + Effect.mapError( + (cause) => + new ManagedAgentRunError({ + reason: "thread-not-found", + threadId: input.threadId, + cause, + }), + ), + ); + const thread = Option.getOrUndefined(threadOption); + if (!thread) { + return yield* new ManagedAgentRunError({ + reason: "thread-not-found", + threadId: input.threadId, + }); + } + const projectOption = yield* snapshots.getProjectShellById(thread.projectId).pipe( + Effect.mapError( + (cause) => + new ManagedAgentRunError({ + reason: "thread-not-found", + threadId: input.threadId, + cause, + }), + ), + ); + const project = Option.getOrUndefined(projectOption); + if (!project) { + return yield* new ManagedAgentRunError({ + reason: "thread-not-found", + threadId: input.threadId, + }); + } + + const runUuid = yield* crypto.randomUUIDv4.pipe( + Effect.mapError( + (cause) => + new ManagedAgentRunError({ + reason: "spawn-failed", + threadId: input.threadId, + cause, + }), + ), + ); + const agentId = RuntimeTaskId.make(`managed-codex-exec:${runUuid}`); + const args = [ + "exec", + "--json", + "--color", + "never", + "-C", + thread.worktreePath ?? project.workspaceRoot, + ]; + if (input.model) args.push("--model", input.model); + if (input.effort) args.push("-c", `model_reasoning_effort=${input.effort}`); + if (input.sandbox) args.push("--sandbox", input.sandbox); + args.push(input.prompt); + + const child = yield* spawner + .spawn( + ChildProcess.make("codex", args, { + cwd: thread.worktreePath ?? project.workspaceRoot, + shell: false, + stdout: "pipe", + stderr: "pipe", + forceKillAfter: "3 seconds", + }), + ) + .pipe( + Effect.provideService(Scope.Scope, scope), Effect.mapError( - () => + (cause) => new ManagedAgentRunError({ - reason: "thread-not-found", - message: `Project for thread ${input.threadId} could not be loaded.`, + reason: "spawn-failed", + threadId: input.threadId, + cause, }), ), ); - const project = Option.getOrUndefined(projectOption); - if (!project) { - return yield* new ManagedAgentRunError({ - reason: "thread-not-found", - message: `Project for thread ${input.threadId} does not exist.`, - }); - } - const runUuid = yield* crypto.randomUUIDv4.pipe( - Effect.mapError( - () => + const cancellationLock = yield* Semaphore.make(1); + const run: ManagedRun = { + threadId: input.threadId, + child, + cancellationLock, + cancelled: false, + }; + runs.set(agentId, run); + const linkage = { + taskId: agentId, + taskType: "managed_codex_exec", + title: input.title, + role: "codex-exec", + ...(input.model ? { model: input.model } : {}), + ...(input.effort ? { effort: input.effort } : {}), + ...(input.parentAgentId ? { parentAgentId: input.parentAgentId } : {}), + agentSource: "managed_codex_exec" as const, + cancellationOwner: "t3" as const, + timelineBypass: true, + }; + yield* emit(input.threadId, "task.started", { + ...linkage, + description: input.title, + }).pipe( + Effect.catch((cause) => + child.kill().pipe( + Effect.ignore, + Effect.andThen( + Effect.sync(() => { + runs.delete(agentId); + }), + ), + Effect.andThen( new ManagedAgentRunError({ reason: "spawn-failed", - message: "Could not allocate a managed agent id.", + threadId: input.threadId, + agentId, + cause, }), + ), ), + ), + ); + + const reportLines = (stream: Stream.Stream) => + stream.pipe( + Stream.decodeText(), + Stream.splitLines, + Stream.mapEffect((line) => { + const summary = outputSummary(line); + return summary + ? emit(input.threadId, "task.progress", { + ...linkage, + description: input.title, + summary, + }) + : Effect.void; + }), + Stream.runDrain, + Effect.ignore, ); - const agentId = RuntimeTaskId.make(`managed-codex-exec:${runUuid}`); - const args = [ - "exec", - "--json", - "--color", - "never", - "-C", - thread.worktreePath ?? project.workspaceRoot, - ]; - if (input.model) args.push("--model", input.model); - if (input.effort) args.push("-c", `model_reasoning_effort=${input.effort}`); - if (input.sandbox) args.push("--sandbox", input.sandbox); - args.push(input.prompt); - const child = yield* spawner - .spawn( - ChildProcess.make("codex", args, { - cwd: thread.worktreePath ?? project.workspaceRoot, - shell: false, - stdout: "pipe", - stderr: "pipe", - forceKillAfter: "3 seconds", + yield* Effect.forkIn( + Effect.gen(function* () { + yield* Effect.all([reportLines(child.stdout), reportLines(child.stderr)], { + concurrency: "unbounded", + }); + const exitResult = yield* Effect.result(child.exitCode); + const cancelled = yield* run.cancellationLock.withPermit( + Effect.sync(() => { + runs.delete(agentId); + return run.cancelled; }), - ) - .pipe( - Effect.provideService(Scope.Scope, scope), - Effect.mapError( - (cause) => - new ManagedAgentRunError({ - reason: "spawn-failed", - message: `Could not launch managed Codex exec: ${Cause.pretty(Cause.fail(cause))}`, - }), - ), ); + yield* emit(input.threadId, "task.completed", { + ...linkage, + status: cancelled + ? "stopped" + : Result.isSuccess(exitResult) && Number(exitResult.success) === 0 + ? "completed" + : "failed", + summary: cancelled + ? "Cancelled by T3" + : Result.isFailure(exitResult) + ? "Managed Codex exec failed before reporting an exit code" + : Number(exitResult.success) === 0 + ? "Managed Codex exec completed" + : `Managed Codex exec exited with code ${Number(exitResult.success)}`, + }); + }), + scope, + ); + return { agentId }; + }, + ); - const cancellationLock = yield* Semaphore.make(1); - const run: ManagedRun = { + const cancel: ManagedCodexExec["Service"]["cancel"] = Effect.fn("ManagedCodexExec.cancel")( + function* (input) { + const run = runs.get(input.agentId); + if (!run) { + return yield* new ManagedAgentRunError({ + reason: "run-not-found", threadId: input.threadId, - child, - cancellationLock, - cancelled: false, - }; - runs.set(agentId, run); - const linkage = { - taskId: agentId, - taskType: "managed_codex_exec", - title: input.title, - role: "codex-exec", - ...(input.model ? { model: input.model } : {}), - ...(input.effort ? { effort: input.effort } : {}), - ...(input.parentAgentId ? { parentAgentId: input.parentAgentId } : {}), - agentSource: "managed_codex_exec" as const, - cancellationOwner: "t3" as const, - timelineBypass: true, - }; - yield* emit(input.threadId, "task.started", { - ...linkage, - description: input.title, - }).pipe( - Effect.catch(() => - child.kill().pipe( - Effect.ignore, - Effect.andThen( - Effect.sync(() => { - runs.delete(agentId); - }), - ), - Effect.andThen( + agentId: input.agentId, + }); + } + if (run.threadId !== input.threadId) { + return yield* new ManagedAgentRunError({ + reason: "not-owned", + threadId: input.threadId, + agentId: input.agentId, + }); + } + yield* run.cancellationLock.withPermit( + Effect.gen(function* () { + if (runs.get(input.agentId) !== run) { + return yield* new ManagedAgentRunError({ + reason: "run-not-found", + threadId: input.threadId, + agentId: input.agentId, + }); + } + yield* run.child.kill({ killSignal: "SIGTERM", forceKillAfter: "3 seconds" }).pipe( + Effect.mapError( + (cause) => new ManagedAgentRunError({ - reason: "spawn-failed", - message: "Managed Codex exec started but its lifecycle could not be recorded.", + reason: "not-owned", + threadId: input.threadId, + agentId: input.agentId, + cause, }), - ), ), - ), - ); - - const reportLines = (stream: Stream.Stream) => - stream.pipe( - Stream.decodeText(), - Stream.splitLines, - Stream.mapEffect((line) => { - const summary = outputSummary(line); - return summary - ? emit(input.threadId, "task.progress", { - ...linkage, - description: input.title, - summary, - }) - : Effect.void; - }), - Stream.runDrain, - Effect.ignore, ); + run.cancelled = true; + }), + ); + return { cancelled: true }; + }, + ); - yield* Effect.forkIn( - Effect.gen(function* () { - yield* Effect.all([reportLines(child.stdout), reportLines(child.stderr)], { - concurrency: "unbounded", - }); - const exitResult = yield* Effect.result(child.exitCode); - const cancelled = yield* run.cancellationLock.withPermit( - Effect.sync(() => { - runs.delete(agentId); - return run.cancelled; - }), - ); - yield* emit(input.threadId, "task.completed", { - ...linkage, - status: cancelled - ? "stopped" - : Result.isSuccess(exitResult) && Number(exitResult.success) === 0 - ? "completed" - : "failed", - summary: cancelled - ? "Cancelled by T3" - : Result.isFailure(exitResult) - ? "Managed Codex exec failed before reporting an exit code" - : Number(exitResult.success) === 0 - ? "Managed Codex exec completed" - : `Managed Codex exec exited with code ${Number(exitResult.success)}`, - }); - }), - scope, - ); - return { agentId }; - }, - ); + yield* Effect.addFinalizer(() => + Effect.forEach(runs.values(), (run) => run.child.kill().pipe(Effect.ignore), { + discard: true, + concurrency: "unbounded", + }), + ); + return { launch, cancel }; +}); - const cancel: ManagedCodexExecShape["cancel"] = Effect.fn("ManagedCodexExec.cancel")( - function* (input) { - const run = runs.get(input.agentId); - if (!run) { - return yield* new ManagedAgentRunError({ - reason: "run-not-found", - message: `Managed agent ${input.agentId} is not running.`, - }); - } - if (run.threadId !== input.threadId) { - return yield* new ManagedAgentRunError({ - reason: "not-owned", - message: `Managed agent ${input.agentId} does not belong to thread ${input.threadId}.`, - }); - } - yield* run.cancellationLock.withPermit( - Effect.gen(function* () { - if (runs.get(input.agentId) !== run) { - return yield* new ManagedAgentRunError({ - reason: "run-not-found", - message: `Managed agent ${input.agentId} is not running.`, - }); - } - yield* run.child.kill({ killSignal: "SIGTERM", forceKillAfter: "3 seconds" }).pipe( - Effect.mapError( - () => - new ManagedAgentRunError({ - reason: "not-owned", - message: `Managed agent ${input.agentId} could not be cancelled.`, - }), - ), - ); - run.cancelled = true; - }), - ); - return { cancelled: true }; - }, - ); - - yield* Effect.addFinalizer(() => - Effect.forEach(runs.values(), (run) => run.child.kill().pipe(Effect.ignore), { - discard: true, - concurrency: "unbounded", - }), - ); - return { launch, cancel }; - }), -); +export const layer = Layer.effect(ManagedCodexExec, make); diff --git a/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts index 7ebfb622707b..fedaf39ad7bb 100644 --- a/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts @@ -33,7 +33,7 @@ export interface ProviderRuntimeIngestionShape { readonly drain: Effect.Effect; /** Explicit adapter boundary for T3-owned runtimes that do not use a provider session. */ - readonly ingestRuntimeEvent?: (event: ProviderRuntimeEvent) => Effect.Effect; + readonly ingestRuntimeEvent: (event: ProviderRuntimeEvent) => Effect.Effect; } /** diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 3f63eb4dbef7..cfd6368faca5 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -110,6 +110,7 @@ import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import { OrchestrationListenerCallbackError } from "./orchestration/Errors.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; +import { ProviderRuntimeIngestionService } from "./orchestration/Services/ProviderRuntimeIngestion.ts"; import { SqlitePersistenceMemory } from "./persistence/Layers/Sqlite.ts"; import { PersistenceSqlError } from "./persistence/Errors.ts"; import * as ProviderRegistry from "./provider/Services/ProviderRegistry.ts"; @@ -764,13 +765,20 @@ const buildAppUnderTest = (options?: { ), ), Layer.provide( - Layer.mock(OrchestrationEngine.OrchestrationEngineService)({ - readEvents: () => Stream.empty, - dispatch: () => Effect.succeed({ sequence: 0 }), - streamDomainEvents: Stream.empty, - latestSequence: Effect.succeed(0), - ...options?.layers?.orchestrationEngine, - }), + Layer.mergeAll( + Layer.mock(OrchestrationEngine.OrchestrationEngineService)({ + readEvents: () => Stream.empty, + dispatch: () => Effect.succeed({ sequence: 0 }), + streamDomainEvents: Stream.empty, + latestSequence: Effect.succeed(0), + ...options?.layers?.orchestrationEngine, + }), + Layer.succeed(ProviderRuntimeIngestionService, { + start: () => Effect.void, + drain: Effect.void, + ingestRuntimeEvent: () => Effect.void, + }), + ), ), Layer.provide( Layer.mock(ProjectionSnapshotQuery.ProjectionSnapshotQuery)({ diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index be3eec15f548..c1a545e24996 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -352,7 +352,7 @@ function toAuthAccessStreamEvent( const makeWsRpcLayer = ( currentSession: EnvironmentAuth.AuthenticatedSession, previewAutomationBroker: PreviewAutomationBroker.PreviewAutomationBroker["Service"], - managedCodexExec: ManagedCodexExec.ManagedCodexExecShape, + managedCodexExec: ManagedCodexExec.ManagedCodexExec["Service"], ) => WsRpcGroup.toLayer( Effect.gen(function* () { diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 85e7bf979e37..55b8a37418d5 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -65,9 +65,16 @@ export class ManagedAgentRunError extends Schema.TaggedErrorClass Date: Thu, 13 Aug 2026 09:22:32 +0200 Subject: [PATCH 6/7] fix(server): redact managed agent spawn errors --- .../orchestration/ManagedCodexExec.test.ts | 19 ++++++++++++++++--- .../src/orchestration/ManagedCodexExec.ts | 18 ++++++------------ packages/contracts/src/orchestration.ts | 1 - 3 files changed, 22 insertions(+), 16 deletions(-) diff --git a/apps/server/src/orchestration/ManagedCodexExec.test.ts b/apps/server/src/orchestration/ManagedCodexExec.test.ts index 190d284e0158..121059cd8af3 100644 --- a/apps/server/src/orchestration/ManagedCodexExec.test.ts +++ b/apps/server/src/orchestration/ManagedCodexExec.test.ts @@ -4,18 +4,22 @@ import { ManagedAgentRunError, ProjectId, ThreadId, + WsOrchestrationLaunchManagedCodexExecRpc, type ProviderRuntimeEvent, } from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as PlatformError from "effect/PlatformError"; import * as Queue from "effect/Queue"; import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; import { ChildProcessSpawner } from "effect/unstable/process"; +import * as Rpc from "effect/unstable/rpc/Rpc"; import { ManagedCodexExec, layer } from "./ManagedCodexExec.ts"; import { @@ -27,6 +31,12 @@ import { type ProjectionSnapshotQueryShape, } from "./Services/ProjectionSnapshotQuery.ts"; +const encodeManagedCodexLaunchExit = Schema.encodeEffect( + Schema.fromJsonString( + Schema.toCodecJson(Rpc.exitSchema(WsOrchestrationLaunchManagedCodexExecRpc)), + ), +); + describe("ManagedCodexExec", () => { it("exposes runtime ingestion as a layer composition requirement", () => { type Requirements = Layer.Services; @@ -252,7 +262,7 @@ describe("ManagedCodexExec", () => { }).pipe(Effect.scoped), ); - it.effect("keeps prompts out of spawn error messages while preserving the cause", () => + it.effect("redacts spawn causes from the managed Codex launch RPC payload", () => Effect.gen(function* () { const prompt = "TOP_SECRET_MANAGED_CODEX_PROMPT"; const spawnCause = PlatformError.systemError({ @@ -307,12 +317,15 @@ describe("ManagedCodexExec", () => { expect(error).toMatchObject({ reason: "spawn-failed", threadId: ThreadId.make("thread-1"), - cause: spawnCause, }); expect(error.message).toBe("Managed agent run failed (spawn-failed): thread-1"); expect(error.message).not.toContain(prompt); expect(String(error)).not.toContain(prompt); - expect(error.cause).toBe(spawnCause); + + const wirePayload = yield* encodeManagedCodexLaunchExit(Exit.fail(error)); + expect(wirePayload).toContain("spawn-failed"); + expect(wirePayload).toContain("thread-1"); + expect(wirePayload).not.toContain(prompt); }).pipe(Effect.scoped), ); }); diff --git a/apps/server/src/orchestration/ManagedCodexExec.ts b/apps/server/src/orchestration/ManagedCodexExec.ts index b356231cdd42..90c9dd0c9dfd 100644 --- a/apps/server/src/orchestration/ManagedCodexExec.ts +++ b/apps/server/src/orchestration/ManagedCodexExec.ts @@ -94,11 +94,10 @@ export const make = Effect.gen(function* () { function* (input) { const threadOption = yield* snapshots.getThreadDetailById(input.threadId).pipe( Effect.mapError( - (cause) => + () => new ManagedAgentRunError({ reason: "thread-not-found", threadId: input.threadId, - cause, }), ), ); @@ -111,11 +110,10 @@ export const make = Effect.gen(function* () { } const projectOption = yield* snapshots.getProjectShellById(thread.projectId).pipe( Effect.mapError( - (cause) => + () => new ManagedAgentRunError({ reason: "thread-not-found", threadId: input.threadId, - cause, }), ), ); @@ -129,11 +127,10 @@ export const make = Effect.gen(function* () { const runUuid = yield* crypto.randomUUIDv4.pipe( Effect.mapError( - (cause) => + () => new ManagedAgentRunError({ reason: "spawn-failed", threadId: input.threadId, - cause, }), ), ); @@ -164,11 +161,10 @@ export const make = Effect.gen(function* () { .pipe( Effect.provideService(Scope.Scope, scope), Effect.mapError( - (cause) => + () => new ManagedAgentRunError({ reason: "spawn-failed", threadId: input.threadId, - cause, }), ), ); @@ -197,7 +193,7 @@ export const make = Effect.gen(function* () { ...linkage, description: input.title, }).pipe( - Effect.catch((cause) => + Effect.catch(() => child.kill().pipe( Effect.ignore, Effect.andThen( @@ -210,7 +206,6 @@ export const make = Effect.gen(function* () { reason: "spawn-failed", threadId: input.threadId, agentId, - cause, }), ), ), @@ -297,12 +292,11 @@ export const make = Effect.gen(function* () { } yield* run.child.kill({ killSignal: "SIGTERM", forceKillAfter: "3 seconds" }).pipe( Effect.mapError( - (cause) => + () => new ManagedAgentRunError({ reason: "not-owned", threadId: input.threadId, agentId: input.agentId, - cause, }), ), ); diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 55b8a37418d5..832355ea40a8 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -67,7 +67,6 @@ export class ManagedAgentRunError extends Schema.TaggedErrorClass Date: Thu, 13 Aug 2026 09:48:59 +0200 Subject: [PATCH 7/7] fix(server): preserve managed agent failure causes --- .../orchestration/ManagedCodexExec.test.ts | 26 +++++--- .../src/orchestration/ManagedCodexExec.ts | 62 ++++++++++++++----- apps/server/src/ws.ts | 8 ++- 3 files changed, 71 insertions(+), 25 deletions(-) diff --git a/apps/server/src/orchestration/ManagedCodexExec.test.ts b/apps/server/src/orchestration/ManagedCodexExec.test.ts index 121059cd8af3..42e1e8905ba7 100644 --- a/apps/server/src/orchestration/ManagedCodexExec.test.ts +++ b/apps/server/src/orchestration/ManagedCodexExec.test.ts @@ -21,7 +21,7 @@ import * as Stream from "effect/Stream"; import { ChildProcessSpawner } from "effect/unstable/process"; import * as Rpc from "effect/unstable/rpc/Rpc"; -import { ManagedCodexExec, layer } from "./ManagedCodexExec.ts"; +import * as ManagedCodexExec from "./ManagedCodexExec.ts"; import { ProviderRuntimeIngestionService, type ProviderRuntimeIngestionShape, @@ -39,7 +39,7 @@ const encodeManagedCodexLaunchExit = Schema.encodeEffect( describe("ManagedCodexExec", () => { it("exposes runtime ingestion as a layer composition requirement", () => { - type Requirements = Layer.Services; + type Requirements = Layer.Services; const requiresRuntimeIngestion: unknown extends Requirements ? false : ProviderRuntimeIngestionService extends Requirements @@ -164,8 +164,8 @@ describe("ManagedCodexExec", () => { Layer.succeed(ProviderRuntimeIngestionService, ingestion), ); - const context = yield* Layer.build(layer.pipe(Layer.provide(dependencies))); - const manager = yield* ManagedCodexExec.pipe(Effect.provide(context)); + const context = yield* Layer.build(ManagedCodexExec.layer.pipe(Layer.provide(dependencies))); + const manager = yield* ManagedCodexExec.ManagedCodexExec.pipe(Effect.provide(context)); const launched = yield* manager.launch({ threadId: ThreadId.make("thread-1"), prompt: "Review the implementation", @@ -297,8 +297,8 @@ describe("ManagedCodexExec", () => { }), ); - const context = yield* Layer.build(layer.pipe(Layer.provide(dependencies))); - const manager = yield* ManagedCodexExec.pipe(Effect.provide(context)); + const context = yield* Layer.build(ManagedCodexExec.layer.pipe(Layer.provide(dependencies))); + const manager = yield* ManagedCodexExec.ManagedCodexExec.pipe(Effect.provide(context)); const result = yield* Effect.result( manager.launch({ threadId: ThreadId.make("thread-1"), @@ -313,16 +313,24 @@ describe("ManagedCodexExec", () => { } const error = result.failure; - expect(error).toBeInstanceOf(ManagedAgentRunError); + expect(error).toBeInstanceOf(ManagedCodexExec.ManagedCodexExecInternalError); expect(error).toMatchObject({ reason: "spawn-failed", threadId: ThreadId.make("thread-1"), }); - expect(error.message).toBe("Managed agent run failed (spawn-failed): thread-1"); + expect(error.cause).toBe(spawnCause); + expect((error.cause as Error).stack).toBe(spawnCause.stack); + expect(error.message).toBe("Managed Codex exec internal failure (spawn-failed): thread-1"); expect(error.message).not.toContain(prompt); expect(String(error)).not.toContain(prompt); - const wirePayload = yield* encodeManagedCodexLaunchExit(Exit.fail(error)); + const publicError = ManagedCodexExec.toManagedAgentRunError(error); + expect(publicError).toBeInstanceOf(ManagedAgentRunError); + expect(publicError).not.toHaveProperty("cause"); + const domainError = new ManagedAgentRunError({ reason: "run-not-found" }); + expect(ManagedCodexExec.toManagedAgentRunError(domainError)).toBe(domainError); + expect(domainError).not.toHaveProperty("cause"); + const wirePayload = yield* encodeManagedCodexLaunchExit(Exit.fail(publicError)); expect(wirePayload).toContain("spawn-failed"); expect(wirePayload).toContain("thread-1"); expect(wirePayload).not.toContain(prompt); diff --git a/apps/server/src/orchestration/ManagedCodexExec.ts b/apps/server/src/orchestration/ManagedCodexExec.ts index 90c9dd0c9dfd..67188f58376c 100644 --- a/apps/server/src/orchestration/ManagedCodexExec.ts +++ b/apps/server/src/orchestration/ManagedCodexExec.ts @@ -3,6 +3,8 @@ import { ManagedAgentRunError, ProviderDriverKind, RuntimeTaskId, + ThreadId, + TrimmedNonEmptyString, type ManagedAgentCancelInput, type ManagedCodexExecLaunchInput, type ProviderRuntimeEvent, @@ -31,18 +33,44 @@ interface ManagedRun { cancelled: boolean; } +export class ManagedCodexExecInternalError extends Schema.TaggedErrorClass()( + "ManagedCodexExecInternalError", + { + reason: Schema.Literals(["thread-not-found", "spawn-failed", "not-owned"]), + threadId: Schema.optional(ThreadId), + agentId: Schema.optional(TrimmedNonEmptyString), + cause: Schema.Defect(), + }, +) { + override get message(): string { + const target = this.agentId ?? this.threadId ?? "managed agent"; + return `Managed Codex exec internal failure (${this.reason}): ${target}`; + } +} + +type ManagedCodexExecError = ManagedAgentRunError | ManagedCodexExecInternalError; + export class ManagedCodexExec extends Context.Service< ManagedCodexExec, { readonly launch: ( input: ManagedCodexExecLaunchInput, - ) => Effect.Effect<{ readonly agentId: string }, ManagedAgentRunError>; + ) => Effect.Effect<{ readonly agentId: string }, ManagedCodexExecError>; readonly cancel: ( input: ManagedAgentCancelInput, - ) => Effect.Effect<{ readonly cancelled: boolean }, ManagedAgentRunError>; + ) => Effect.Effect<{ readonly cancelled: boolean }, ManagedCodexExecError>; } >()("t3/orchestration/ManagedCodexExec") {} +export function toManagedAgentRunError(error: ManagedCodexExecError): ManagedAgentRunError { + if (error._tag === "ManagedAgentRunError") return error; + return new ManagedAgentRunError({ + reason: error.reason, + ...(error.threadId !== undefined ? { threadId: error.threadId } : {}), + ...(error.agentId !== undefined ? { agentId: error.agentId } : {}), + }); +} + function outputSummary(line: string): string | undefined { const trimmed = line.trim(); if (!trimmed) return undefined; @@ -94,10 +122,11 @@ export const make = Effect.gen(function* () { function* (input) { const threadOption = yield* snapshots.getThreadDetailById(input.threadId).pipe( Effect.mapError( - () => - new ManagedAgentRunError({ + (cause) => + new ManagedCodexExecInternalError({ reason: "thread-not-found", threadId: input.threadId, + cause, }), ), ); @@ -110,10 +139,11 @@ export const make = Effect.gen(function* () { } const projectOption = yield* snapshots.getProjectShellById(thread.projectId).pipe( Effect.mapError( - () => - new ManagedAgentRunError({ + (cause) => + new ManagedCodexExecInternalError({ reason: "thread-not-found", threadId: input.threadId, + cause, }), ), ); @@ -127,10 +157,11 @@ export const make = Effect.gen(function* () { const runUuid = yield* crypto.randomUUIDv4.pipe( Effect.mapError( - () => - new ManagedAgentRunError({ + (cause) => + new ManagedCodexExecInternalError({ reason: "spawn-failed", threadId: input.threadId, + cause, }), ), ); @@ -161,10 +192,11 @@ export const make = Effect.gen(function* () { .pipe( Effect.provideService(Scope.Scope, scope), Effect.mapError( - () => - new ManagedAgentRunError({ + (cause) => + new ManagedCodexExecInternalError({ reason: "spawn-failed", threadId: input.threadId, + cause, }), ), ); @@ -193,7 +225,7 @@ export const make = Effect.gen(function* () { ...linkage, description: input.title, }).pipe( - Effect.catch(() => + Effect.catch((cause) => child.kill().pipe( Effect.ignore, Effect.andThen( @@ -202,10 +234,11 @@ export const make = Effect.gen(function* () { }), ), Effect.andThen( - new ManagedAgentRunError({ + new ManagedCodexExecInternalError({ reason: "spawn-failed", threadId: input.threadId, agentId, + cause, }), ), ), @@ -292,11 +325,12 @@ export const make = Effect.gen(function* () { } yield* run.child.kill({ killSignal: "SIGTERM", forceKillAfter: "3 seconds" }).pipe( Effect.mapError( - () => - new ManagedAgentRunError({ + (cause) => + new ManagedCodexExecInternalError({ reason: "not-owned", threadId: input.threadId, agentId: input.agentId, + cause, }), ), ); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index c1a545e24996..d0c7af1ced6f 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1038,13 +1038,17 @@ const makeWsRpcLayer = ( [ORCHESTRATION_WS_METHODS.launchManagedCodexExec]: (input) => observeRpcEffect( ORCHESTRATION_WS_METHODS.launchManagedCodexExec, - managedCodexExec.launch(input), + managedCodexExec + .launch(input) + .pipe(Effect.mapError(ManagedCodexExec.toManagedAgentRunError)), { "rpc.aggregate": "thread", threadId: input.threadId }, ), [ORCHESTRATION_WS_METHODS.cancelManagedAgent]: (input) => observeRpcEffect( ORCHESTRATION_WS_METHODS.cancelManagedAgent, - managedCodexExec.cancel(input), + managedCodexExec + .cancel(input) + .pipe(Effect.mapError(ManagedCodexExec.toManagedAgentRunError)), { "rpc.aggregate": "thread", threadId: input.threadId }, ), [ORCHESTRATION_WS_METHODS.dispatchCommand]: (command) =>