diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index ce34855a3194..ad7046d620f7 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -14,6 +14,7 @@ import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as MonitorSession from "../src/mcp/MonitorSession.ts"; import * as ManagedRuntime from "effect/ManagedRuntime"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; @@ -411,6 +412,7 @@ export const makeOrchestrationIntegrationHarness = ( Layer.provideMerge(runtimeServicesLayer), Layer.provideMerge(orchestrationReactorLayer), Layer.provideMerge(providerRegistryLayer), + Layer.provideMerge(MonitorSession.layer), Layer.provide(persistenceLayer), Layer.provideMerge(RepositoryIdentityResolver.layer), Layer.provideMerge(ServerSettingsService.layerTest()), diff --git a/apps/server/src/mcp/McpHttpServer.test.ts b/apps/server/src/mcp/McpHttpServer.test.ts index fa2880f9c364..f506c54df738 100644 --- a/apps/server/src/mcp/McpHttpServer.test.ts +++ b/apps/server/src/mcp/McpHttpServer.test.ts @@ -4,10 +4,14 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { EnvironmentId, PreviewTabId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { McpProtocol, McpSchema, McpServer } from "effect/unstable/ai"; import { HttpBody, HttpClient, HttpRouter, HttpServerResponse } from "effect/unstable/http"; +import * as McpSessionRegistry from "./McpSessionRegistry.ts"; +import * as MonitorSession from "./MonitorSession.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as McpHttpServer from "./McpHttpServer.ts"; import * as McpInvocationContext from "./McpInvocationContext.ts"; import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts"; @@ -286,3 +290,82 @@ it.effect("registers annotated tools and preserves authenticated request context }), ).pipe(Effect.provide(TestLayer)), ); + +it.effect("HTTP tool discovery only advertises monitors to monitoring credentials", () => + Effect.gen(function* () { + yield* HttpRouter.serve(McpHttpServer.layer, { + disableListenLog: true, + disableLogger: true, + }).pipe(Layer.build); + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const httpClient = yield* HttpClient.HttpClient; + for (const capabilities of [["preview"], ["monitor"], []] as const) { + const { config } = yield* registry.issue({ + threadId, + providerInstanceId: ProviderInstanceId.make("test"), + capabilities, + }); + const headers = { + authorization: config.authorizationHeader, + accept: "application/json, text/event-stream", + }; + const initialized = yield* httpClient.post("/mcp", { + headers, + body: HttpBody.text( + '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"mcp-test","version":"1.0.0"}}}', + "application/json", + ), + }); + const sessionId = initialized.headers["mcp-session-id"]!; + expect(initialized.status).toBe(200); + yield* initialized.text; + const listed = yield* httpClient.post("/mcp", { + headers: { ...headers, "mcp-session-id": sessionId, "mcp-protocol-version": "2025-06-18" }, + body: HttpBody.text( + '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}', + "application/json", + ), + }); + const decoded = yield* listed.json.pipe( + Effect.flatMap( + Schema.decodeUnknownEffect( + Schema.Struct({ + result: Schema.Struct({ + tools: Schema.Array(Schema.Struct({ name: Schema.String })), + }), + }), + ), + ), + ); + const monitorNames = decoded.result.tools + .map((tool) => tool.name) + .filter((name) => name.startsWith("monitor_")); + expect(monitorNames).toEqual( + capabilities.some((capability) => capability === "monitor") + ? ["monitor_start", "monitor_unsubscribe"] + : [], + ); + } + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + McpSessionRegistry.layer, + PreviewAutomationBroker.layer, + MonitorSession.layer, + ).pipe( + Layer.provide( + Layer.succeed( + ServerEnvironment.ServerEnvironment, + ServerEnvironment.ServerEnvironment.of({ + getEnvironmentId: Effect.succeed(environmentId), + getDescriptor: Effect.die("unused"), + }), + ), + ), + Layer.provideMerge(NodeHttpServer.layerTest), + Layer.provideMerge(NodeServices.layer), + ), + ), + ), +); diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 44ca928e63bb..45fe09dd99c5 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -23,6 +23,13 @@ import { PreviewStandardToolkit, } from "./toolkits/preview/tools.ts"; +import { MonitorToolkit } from "./toolkits/monitor/tools.ts"; +import { MonitorToolkitHandlersLive } from "./toolkits/monitor/handlers.ts"; + +export const MonitorToolkitRegistrationLive = McpServer.toolkit(MonitorToolkit).pipe( + Layer.provide(MonitorToolkitHandlersLive), +); + const unauthorized = HttpServerResponse.jsonUnsafe( { error: "invalid_mcp_credential", @@ -222,4 +229,7 @@ const McpTransportLive = McpServer.layerHttp({ protocols: [McpProtocol.v2025_06_18], }).pipe(Layer.provide(McpAuthMiddlewareLive)); -export const layer = PreviewToolkitRegistrationLive.pipe(Layer.provideMerge(McpTransportLive)); +export const layer = Layer.mergeAll( + PreviewToolkitRegistrationLive, + MonitorToolkitRegistrationLive, +).pipe(Layer.provideMerge(McpTransportLive)); diff --git a/apps/server/src/mcp/McpInvocationContext.ts b/apps/server/src/mcp/McpInvocationContext.ts index 49273485a44d..7fe650d77e03 100644 --- a/apps/server/src/mcp/McpInvocationContext.ts +++ b/apps/server/src/mcp/McpInvocationContext.ts @@ -7,7 +7,7 @@ import { import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; -export type McpCapability = "preview"; +export type McpCapability = "preview" | "monitor"; export interface McpInvocationScope { readonly environmentId: EnvironmentId; @@ -24,7 +24,7 @@ export class McpInvocationContext extends Context.Service< >()("t3/mcp/McpInvocationContext") {} export const requireMcpCapability = Effect.fn("mcp.requireCapability")(function* ( - capability: McpCapability, + capability: "preview", ) { const invocation = yield* McpInvocationContext; if (!invocation.capabilities.has(capability)) { diff --git a/apps/server/src/mcp/McpProviderSession.ts b/apps/server/src/mcp/McpProviderSession.ts index d5dc582046c1..9d5e78bad203 100644 --- a/apps/server/src/mcp/McpProviderSession.ts +++ b/apps/server/src/mcp/McpProviderSession.ts @@ -1,10 +1,12 @@ import type { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import type { McpCapability } from "./McpInvocationContext.ts"; export interface McpProviderSessionConfig { readonly environmentId: EnvironmentId; readonly threadId: ThreadId; readonly providerSessionId: string; readonly providerInstanceId: ProviderInstanceId; + readonly capabilities?: ReadonlyArray; readonly endpoint: string; readonly authorizationHeader: string; } diff --git a/apps/server/src/mcp/McpSessionRegistry.test.ts b/apps/server/src/mcp/McpSessionRegistry.test.ts index 1d8aead99d0d..37ea5dfa7b88 100644 --- a/apps/server/src/mcp/McpSessionRegistry.test.ts +++ b/apps/server/src/mcp/McpSessionRegistry.test.ts @@ -127,3 +127,19 @@ it.effect("does not keep credentials of other threads alive", () => expect(yield* registry.resolve(token)).toBeUndefined(); }), ); + +it.effect("preserves the explicitly granted toolkit capabilities", () => + Effect.gen(function* () { + const registry = yield* makeRegistry(() => 1_000); + const issued = yield* registry.issue({ + threadId: ThreadId.make("monitor-only"), + providerInstanceId: ProviderInstanceId.make("codex"), + capabilities: ["monitor"], + }); + const resolved = yield* registry.resolve( + issued.config.authorizationHeader.replace(/^Bearer\s+/, ""), + ); + expect(Array.from(resolved!.capabilities)).toEqual(["monitor"]); + expect(issued.config.capabilities).toEqual(["monitor"]); + }), +); diff --git a/apps/server/src/mcp/McpSessionRegistry.ts b/apps/server/src/mcp/McpSessionRegistry.ts index f19a4f4e8c49..4b06138aa3c4 100644 --- a/apps/server/src/mcp/McpSessionRegistry.ts +++ b/apps/server/src/mcp/McpSessionRegistry.ts @@ -12,6 +12,7 @@ import * as McpInvocationContext from "./McpInvocationContext.ts"; import * as McpProviderSession from "./McpProviderSession.ts"; export interface McpCredentialRequest { + readonly capabilities?: ReadonlyArray; readonly threadId: ThreadId; readonly providerInstanceId: ProviderInstanceId; } @@ -128,7 +129,7 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( threadId: ThreadId.make(request.threadId), providerSessionId, providerInstanceId: ProviderInstanceId.make(request.providerInstanceId), - capabilities: new Set(["preview"]), + capabilities: new Set(request.capabilities ?? ["preview"]), issuedAt, }; yield* SynchronizedRef.update(state, ({ records }) => { @@ -139,6 +140,7 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( return { config: { environmentId, + capabilities: Array.from(scope.capabilities), threadId: scope.threadId, providerSessionId, providerInstanceId: scope.providerInstanceId, diff --git a/apps/server/src/mcp/MonitorSession.test.ts b/apps/server/src/mcp/MonitorSession.test.ts new file mode 100644 index 000000000000..49ab50b41037 --- /dev/null +++ b/apps/server/src/mcp/MonitorSession.test.ts @@ -0,0 +1,140 @@ +import { expect, it } from "@effect/vitest"; +import { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import { McpSchema, McpServer } from "effect/unstable/ai"; +import { McpInvocationContext, requireMcpCapability } from "./McpInvocationContext.ts"; +import { MonitorToolkitRegistrationLive } from "./McpHttpServer.ts"; +import * as MonitorSession from "./MonitorSession.ts"; +import { CodexBackgroundTasks } from "../provider/Layers/CodexBackgroundTasks.ts"; + +const scope = { + environmentId: EnvironmentId.make("monitor-test"), + threadId: ThreadId.make("monitor-test"), + providerSessionId: "monitor-session", + providerInstanceId: ProviderInstanceId.make("codex"), + capabilities: new Set(["monitor"] as const), + issuedAt: 1, +}; +const client = McpSchema.McpServerClient.of({ + clientId: 1, + protocolVersion: "2025-06-18", + initializePayload: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "monitor-test", version: "1.0.0" }, + }, + getClient: Effect.die("unused"), +}); +const TestLayer = MonitorToolkitRegistrationLive.pipe( + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provideMerge(MonitorSession.layer), +); + +it.effect("MCP subscription enables wakes and unsubscribe discards queued events", () => + Effect.gen(function* () { + const server = yield* McpServer.McpServer; + const tasks = new CodexBackgroundTasks(); + tasks.started({ + id: "watch", + processId: "42", + source: "unifiedExecStartup", + command: "watch-ci", + }); + yield* (yield* MonitorSession.MonitorSessions).register(scope.providerSessionId, { + start: () => + Effect.sync(() => { + tasks.subscribe("42"); + return { monitorId: "42", status: "scheduled" as const }; + }), + subscribe: (id) => + Effect.sync(() => { + expect(tasks.subscribe(id)).toBe(true); + }), + unsubscribe: (id) => Effect.sync(() => tasks.unsubscribe(id)), + }); + const call = (name: string) => server.callTool({ name, arguments: { processId: "42" } }); + expect( + (yield* server.callTool({ name: "monitor_start", arguments: { command: ["watch-ci"] } })) + .isError, + ).toBe(false); + tasks.output("watch", "first event\n"); + expect(tasks.takeWake()?.output).toContain("first event"); + tasks.output("watch", "queued event\n"); + expect((yield* call("monitor_unsubscribe")).isError).toBe(false); + tasks.output("watch", "later event\n"); + expect(tasks.takeWake()).toBeUndefined(); + }).pipe( + Effect.scoped, + Effect.provideService(McpInvocationContext, scope), + Effect.provideService(McpSchema.McpServerClient, client), + Effect.provide(TestLayer), + ), +); + +it.effect("MCP tools reject other sessions, missing capability, and a closed runtime", () => + Effect.gen(function* () { + const server = yield* McpServer.McpServer; + let subscribed = false; + const call = server.callTool({ name: "monitor_start", arguments: { command: ["watch-ci"] } }); + yield* Effect.gen(function* () { + yield* (yield* MonitorSession.MonitorSessions).register(scope.providerSessionId, { + start: () => + Effect.sync(() => { + subscribed = true; + return { monitorId: "42", status: "scheduled" as const }; + }), + subscribe: () => + Effect.sync(() => { + subscribed = true; + }), + unsubscribe: () => Effect.void, + }); + expect( + (yield* call.pipe( + Effect.provideService(McpInvocationContext, { + ...scope, + providerSessionId: "other-session", + }), + )).isError, + ).toBe(true); + expect( + (yield* call.pipe( + Effect.provideService(McpInvocationContext, { + ...scope, + capabilities: new Set(["preview"] as const), + }), + )).isError, + ).toBe(true); + expect(subscribed).toBe(false); + }).pipe(Effect.scoped); + expect((yield* call).isError).toBe(true); + }).pipe( + Effect.provideService(McpInvocationContext, scope), + Effect.provideService(McpSchema.McpServerClient, client), + Effect.provide(TestLayer), + ), +); + +it.effect("separately constructed registries isolate the same provider session ID", () => + Effect.gen(function* () { + const first = yield* MonitorSession.make; + const second = yield* MonitorSession.make; + yield* first.register("same-session", { + start: () => Effect.succeed({ monitorId: "42", status: "scheduled" as const }), + subscribe: () => Effect.void, + unsubscribe: () => Effect.void, + }); + expect((yield* first.invoke("same-session", "subscribe", "42")).subscribed).toBe(true); + const missing = yield* second.invoke("same-session", "subscribe", "42").pipe(Effect.result); + expect(missing._tag).toBe("Failure"); + }).pipe(Effect.scoped), +); + +it.effect("a monitoring credential cannot invoke preview tools", () => + requireMcpCapability("preview").pipe( + Effect.result, + Effect.tap((result) => Effect.sync(() => expect(result._tag).toBe("Failure"))), + Effect.provideService(McpInvocationContext, scope), + ), +); diff --git a/apps/server/src/mcp/MonitorSession.ts b/apps/server/src/mcp/MonitorSession.ts new file mode 100644 index 000000000000..3d95c91eeb40 --- /dev/null +++ b/apps/server/src/mcp/MonitorSession.ts @@ -0,0 +1,125 @@ +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; +import * as Context from "effect/Context"; +import * as Layer from "effect/Layer"; +import type * as Scope from "effect/Scope"; + +export class MonitorUnavailableError extends Schema.TaggedErrorClass()( + "MonitorUnavailableError", + { sessionId: Schema.String }, +) { + override get message() { + return `Monitoring requires an active Codex 0.153.2 or later session (${this.sessionId}).`; + } +} + +export class MonitorCapabilityError extends Schema.TaggedErrorClass()( + "MonitorCapabilityError", + {}, +) { + override get message() { + return "Monitoring is unavailable for this provider session."; + } +} + +export class MonitorStoppedError extends Schema.TaggedErrorClass()( + "MonitorStoppedError", + {}, +) { + override get message() { + return "Monitoring is unavailable or was stopped. Start a new turn to resume."; + } +} + +export class MonitorProcessMissingError extends Schema.TaggedErrorClass()( + "MonitorProcessMissingError", + { processId: Schema.String }, +) { + override get message() { + return `No running background process with session ID ${this.processId}. Launch a watcher with exec_command first.`; + } +} + +export class MonitorStartError extends Schema.TaggedErrorClass()( + "MonitorStartError", + { cause: Schema.Defect() }, +) { + override get message() { + return "Could not schedule the background monitor."; + } +} + +export const MonitorError = Schema.Union([ + MonitorStartError, + MonitorUnavailableError, + MonitorCapabilityError, + MonitorStoppedError, + MonitorProcessMissingError, +]); + +export interface MonitorSession { + readonly start: ( + command: ReadonlyArray, + ) => Effect.Effect<{ monitorId: string; status: "scheduled" }, typeof MonitorError.Type>; + readonly subscribe: (processId: string) => Effect.Effect; + readonly unsubscribe: (processId: string) => Effect.Effect; +} + +// Like McpProviderSession, the bridge lives only for the provider session. +// Keying by credential session ID prevents a replaced runtime receiving calls +// authenticated for its predecessor. +export class MonitorSessions extends Context.Service< + MonitorSessions, + { + readonly register: ( + sessionId: string, + session: MonitorSession, + ) => Effect.Effect; + readonly invoke: ( + sessionId: string, + operation: "subscribe" | "unsubscribe", + processId: string, + ) => Effect.Effect<{ processId: string; subscribed: boolean }, typeof MonitorError.Type>; + readonly start: ( + sessionId: string, + command: ReadonlyArray, + ) => Effect.Effect<{ monitorId: string; status: "scheduled" }, typeof MonitorError.Type>; + } +>()("t3/mcp/MonitorSession/MonitorSessions") {} + +export const make = Effect.sync(() => { + const sessions = new Map(); + + const register = (sessionId: string, session: MonitorSession) => + Effect.acquireRelease( + Effect.sync(() => { + sessions.set(sessionId, session); + }), + () => + Effect.sync(() => { + if (sessions.get(sessionId) === session) sessions.delete(sessionId); + }), + ); + + const invoke = Effect.fn("MonitorSession.invoke")(function* ( + sessionId: string, + operation: "subscribe" | "unsubscribe", + processId: string, + ) { + const session = sessions.get(sessionId); + if (!session) return yield* new MonitorUnavailableError({ sessionId }); + yield* session[operation](processId); + return { processId, subscribed: operation === "subscribe" }; + }); + const start = Effect.fn("MonitorSession.start")(function* ( + sessionId: string, + command: ReadonlyArray, + ) { + const session = sessions.get(sessionId); + if (!session) return yield* new MonitorUnavailableError({ sessionId }); + return yield* session.start(command); + }); + return { register, invoke, start }; +}); + +export const layer = Layer.effect(MonitorSessions, make); diff --git a/apps/server/src/mcp/toolkits/monitor/handlers.ts b/apps/server/src/mcp/toolkits/monitor/handlers.ts new file mode 100644 index 000000000000..8be578c6fd39 --- /dev/null +++ b/apps/server/src/mcp/toolkits/monitor/handlers.ts @@ -0,0 +1,26 @@ +import * as Effect from "effect/Effect"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as MonitorSession from "../../MonitorSession.ts"; +import { MonitorToolkit } from "./tools.ts"; + +const invoke = Effect.fn("MonitorToolkit.invoke")(function* ( + operation: "subscribe" | "unsubscribe", + processId: string, +) { + const scope = yield* McpInvocationContext.McpInvocationContext; + const sessions = yield* MonitorSession.MonitorSessions; + if (!scope.capabilities.has("monitor")) + return yield* new MonitorSession.MonitorCapabilityError({}); + return yield* sessions.invoke(scope.providerSessionId, operation, processId); +}); + +export const MonitorToolkitHandlersLive = MonitorToolkit.toLayer({ + monitor_start: ({ command }) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + if (!scope.capabilities.has("monitor")) + return yield* new MonitorSession.MonitorCapabilityError({}); + return yield* (yield* MonitorSession.MonitorSessions).start(scope.providerSessionId, command); + }), + monitor_unsubscribe: ({ processId }) => invoke("unsubscribe", processId), +}); diff --git a/apps/server/src/mcp/toolkits/monitor/tools.ts b/apps/server/src/mcp/toolkits/monitor/tools.ts new file mode 100644 index 000000000000..3163ab9196f3 --- /dev/null +++ b/apps/server/src/mcp/toolkits/monitor/tools.ts @@ -0,0 +1,61 @@ +import * as Context from "effect/Context"; +import * as Fiber from "effect/Fiber"; +import * as Schema from "effect/Schema"; +import { McpSchema, Tool, Toolkit } from "effect/unstable/ai"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as MonitorSession from "../../MonitorSession.ts"; + +// MCP's discovery predicate is synchronous; read the authenticated request +// context from its current fiber rather than trusting client-supplied metadata. +const monitoringEnabled = () => { + const fiber = Fiber.getCurrent(); + return ( + fiber !== undefined && + (Context.getOrUndefined( + fiber.context, + McpInvocationContext.McpInvocationContext, + )?.capabilities.has("monitor") ?? + false) + ); +}; + +const parameters = Schema.Struct({ + processId: Schema.String.annotate({ + description: "The monitorId returned by monitor_start.", + }), +}); +const success = Schema.Struct({ processId: Schema.String, subscribed: Schema.Boolean }); + +export const MonitorStartTool = Tool.make("monitor_start", { + description: + "Start a background command that wakes a new agent turn when it emits a complete output line, exits, or fails to launch. Use this when asked to wait, watch, monitor, or notify later, including timers, instead of sleeping or polling in the current turn. Returns immediately; finish your turn after scheduling.", + parameters: Schema.Struct({ + command: Schema.NonEmptyArray(Schema.String).annotate({ + description: + "Executable and arguments. For Bash commands, use ['bash', '-c', 'sleep 30; echo done'].", + }), + }), + success: Schema.Struct({ monitorId: Schema.String, status: Schema.Literal("scheduled") }), + failure: MonitorSession.MonitorError, + dependencies: [McpInvocationContext.McpInvocationContext, MonitorSession.MonitorSessions], +}) + .annotate(Tool.Title, "Monitor background process") + .annotate(Tool.Destructive, true) + .annotate(Tool.Idempotent, false) + .annotate(Tool.OpenWorld, true) + .annotate(McpSchema.EnabledWhen, monitoringEnabled); + +export const MonitorUnsubscribeTool = Tool.make("monitor_unsubscribe", { + description: + "Unsubscribe from the background process selected by processId and discard its pending wakes. The process continues running.", + parameters, + success, + failure: MonitorSession.MonitorError, + dependencies: [McpInvocationContext.McpInvocationContext, MonitorSession.MonitorSessions], +}) + .annotate(Tool.Title, "Unsubscribe from background process") + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, true) + .annotate(McpSchema.EnabledWhen, monitoringEnabled); + +export const MonitorToolkit = Toolkit.make(MonitorStartTool, MonitorUnsubscribeTool); diff --git a/apps/server/src/provider/Drivers/CodexDriver.test.ts b/apps/server/src/provider/Drivers/CodexDriver.test.ts index bac34db452fd..13823cc9c8ec 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.test.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.test.ts @@ -8,6 +8,7 @@ import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as MonitorSession from "../../mcp/MonitorSession.ts"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; import { HttpClient } from "effect/unstable/http"; @@ -31,6 +32,7 @@ const testLayer = ServerConfig.layerTest(process.cwd(), { prefix: "t3-codex-driver-maintenance-", }).pipe( Layer.provideMerge(NodeServices.layer), + Layer.provideMerge(MonitorSession.layer), Layer.provideMerge(ServerSettingsService.layerTest()), Layer.provideMerge(ModelManifest.layerTest), Layer.provideMerge(codexResetCreditLayerTest), diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index 071fb20674a8..b73762aebb94 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -32,6 +32,7 @@ import { ChildProcessSpawner } from "effect/unstable/process"; import { makeCodexTextGeneration } from "../../textGeneration/CodexTextGeneration.ts"; import * as BackgroundPolicy from "../../background/BackgroundPolicy.ts"; +import * as MonitorSession from "../../mcp/MonitorSession.ts"; import { ServerConfig } from "../../config.ts"; import { expandHomePath } from "../../pathExpansion.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; @@ -104,6 +105,7 @@ function makeCodexMaintenanceResolver(sharedHomePath: string) { * registered driver and the runtime satisfies them once. */ export type CodexDriverEnv = + | MonitorSession.MonitorSessions | BackgroundPolicy.BackgroundPolicy | ChildProcessSpawner.ChildProcessSpawner | CodexResetCreditCoordinator diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 4676d780a530..0223285672df 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -27,6 +27,7 @@ import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as MonitorSession from "../../mcp/MonitorSession.ts"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; @@ -245,6 +246,7 @@ const validationLayer = it.layer( Layer.provideMerge(ServerSettingsService.layerTest()), Layer.provideMerge(providerSessionDirectoryTestLayer), Layer.provideMerge(NodeServices.layer), + Layer.provideMerge(MonitorSession.layer), ), ); @@ -315,6 +317,7 @@ const sessionErrorLayer = it.layer( Layer.provideMerge(ServerSettingsService.layerTest()), Layer.provideMerge(providerSessionDirectoryTestLayer), Layer.provideMerge(NodeServices.layer), + Layer.provideMerge(MonitorSession.layer), ), ); @@ -462,6 +465,7 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => { Layer.provideMerge(ServerSettingsService.layerTest()), Layer.provideMerge(providerSessionDirectoryTestLayer), Layer.provideMerge(NodeServices.layer), + Layer.provideMerge(MonitorSession.layer), ); return Effect.gen(function* () { @@ -494,6 +498,7 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => { Layer.provideMerge(ServerSettingsService.layerTest()), Layer.provideMerge(providerSessionDirectoryTestLayer), Layer.provideMerge(NodeServices.layer), + Layer.provideMerge(MonitorSession.layer), ); return Effect.gen(function* () { @@ -527,6 +532,7 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => { Layer.provideMerge(ServerSettingsService.layerTest()), Layer.provideMerge(providerSessionDirectoryTestLayer), Layer.provideMerge(NodeServices.layer), + Layer.provideMerge(MonitorSession.layer), ); return Effect.gen(function* () { @@ -581,6 +587,7 @@ const lifecycleLayer = it.layer( Layer.provideMerge(ServerSettingsService.layerTest()), Layer.provideMerge(providerSessionDirectoryTestLayer), Layer.provideMerge(NodeServices.layer), + Layer.provideMerge(MonitorSession.layer), ), ); @@ -669,6 +676,68 @@ function codexTurnEvent(method: "turn/started" | "turn/completed", turnId: strin } lifecycleLayer("CodexAdapterLive lifecycle", (it) => { + it.effect( + "maps native background work and delivered monitor events into the shared timeline", + () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const mapped = yield* adapter.streamEvents.pipe( + Stream.filter( + (e) => + e.type === "task.started" || + e.type === "task.completed" || + (e.type === "item.completed" && e.payload.title === "Monitor event"), + ), + Stream.take(3), + Stream.runCollect, + Effect.forkChild, + ); + const base = { + kind: "notification" as const, + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + threadId: asThreadId("thread-1"), + }; + yield* runtime.emit({ + ...base, + id: asEventId("bg-start"), + method: "backgroundTask/changed", + payload: { taskId: "shell", description: "watch-ci", status: "running" }, + }); + yield* runtime.emit({ + ...base, + id: asEventId("bg-output"), + method: "backgroundMonitor/delivered", + turnId: asTurnId("wake-turn"), + itemId: asItemId("wake-event"), + payload: { name: "background_monitor", output: "CI passed" }, + }); + yield* runtime.emit({ + ...base, + id: asEventId("bg-stop"), + method: "backgroundTask/changed", + payload: { taskId: "shell", description: "watch-ci", status: "stopped" }, + }); + const events = Array.from(yield* Fiber.join(mapped)); + NodeAssert.deepStrictEqual( + events.map((e) => e.type), + ["task.started", "item.completed", "task.completed"], + ); + NodeAssert.deepStrictEqual(events[0]?.payload, { + taskId: "shell", + description: "watch-ci", + taskType: "shell", + }); + NodeAssert.equal(events[1]?.turnId, "wake-turn"); + NodeAssert.equal(events[1]?.itemId, "wake-event"); + NodeAssert.deepStrictEqual(events[2]?.payload, { + taskId: "shell", + status: "stopped", + taskType: "shell", + }); + }), + ); + it.effect("calculates one Codex turn total from cumulative counters", () => Effect.gen(function* () { const { adapter, runtime } = yield* startLifecycleRuntime(); @@ -2557,6 +2626,7 @@ const scopedLifecycleLayer = it.layer( Layer.provideMerge(ServerSettingsService.layerTest()), Layer.provideMerge(providerSessionDirectoryTestLayer), Layer.provideMerge(NodeServices.layer), + Layer.provideMerge(MonitorSession.layer), ), ); @@ -2601,6 +2671,7 @@ const scopedFailureLayer = it.layer( Layer.provideMerge(ServerSettingsService.layerTest()), Layer.provideMerge(providerSessionDirectoryTestLayer), Layer.provideMerge(NodeServices.layer), + Layer.provideMerge(MonitorSession.layer), ), ); @@ -2653,6 +2724,7 @@ it.effect("flushes managed native logs when the adapter layer shuts down", () => Layer.provideMerge(ServerSettingsService.layerTest()), Layer.provideMerge(providerSessionDirectoryTestLayer), Layer.provideMerge(NodeServices.layer), + Layer.provideMerge(MonitorSession.layer), ); const context = yield* Layer.buildWithScope(layer, scope); const adapter = yield* Effect.service(CodexAdapter).pipe(Effect.provide(context)); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index d1981b33d47d..d3984d68b706 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -70,8 +70,10 @@ import { type CodexSessionRuntimeShape, } from "./CodexSessionRuntime.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; +import { CodexBackgroundTaskEvent, CodexMonitorOutput } from "./CodexBackgroundTasks.ts"; import { resolveCodexLaunchArgs } from "./codexLaunchArgs.ts"; import { codexRateLimitsToUpdate } from "./codexUsageLimits.ts"; +import * as MonitorSession from "../../mcp/MonitorSession.ts"; const isCodexAppServerProcessExitedError = Schema.is(CodexErrors.CodexAppServerProcessExitedError); const isCodexAppServerTransportError = Schema.is(CodexErrors.CodexAppServerTransportError); const isCodexSessionRuntimeThreadIdMissingError = Schema.is( @@ -1296,6 +1298,43 @@ function mapToRuntimeEvents( event: ProviderEvent, canonicalThreadId: ThreadId, ): ReadonlyArray { + if (event.kind === "notification" && event.method === "backgroundMonitor/delivered") { + const output = readPayload(CodexMonitorOutput, event.payload); + if (!output) return []; + return [ + { + ...runtimeEventBase(event, canonicalThreadId), + type: "item.completed", + payload: { + itemType: "dynamic_tool_call", + status: "completed", + title: "Monitor event", + detail: output.output, + }, + }, + ]; + } + if (event.kind === "notification" && event.method === "backgroundTask/changed") { + const task = readPayload(CodexBackgroundTaskEvent, event.payload); + if (!task) return []; + const base = runtimeEventBase(event, canonicalThreadId); + const taskId = RuntimeTaskId.make(task.taskId); + return task.status === "running" + ? [ + { + ...base, + type: "task.started", + payload: { taskId, description: task.description, taskType: "shell" }, + }, + ] + : [ + { + ...base, + type: "task.completed", + payload: { taskId, status: task.status, taskType: "shell" }, + }, + ]; + } if (event.kind === "notification" && event.method.startsWith("collabAgent/")) { return mapCollabAgentEvent(event, canonicalThreadId); } @@ -2213,6 +2252,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( options?: CodexAdapterLiveOptions, ) { const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("codex"); + const monitorSessions = yield* MonitorSession.MonitorSessions; const fileSystem = yield* FileSystem.FileSystem; const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const crypto = yield* Crypto.Crypto; @@ -2268,6 +2308,8 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ...(serviceTier ? { serviceTier } : {}), ...(mcpSession ? { + mcpProviderSessionId: mcpSession.providerSessionId, + browserToolsAvailable: mcpSession.capabilities?.includes("preview") ?? true, environment: { ...(options?.environment ?? process.env), T3_MCP_BEARER_TOKEN: mcpSession.authorizationHeader.replace(/^Bearer\s+/, ""), @@ -2289,6 +2331,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ); const createRuntime = options?.makeRuntime ?? makeCodexSessionRuntime; const runtime = yield* createRuntime(runtimeInput).pipe( + Effect.provideService(MonitorSession.MonitorSessions, monitorSessions), Effect.provideService(Scope.Scope, sessionScope), Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner), Effect.provideService(Crypto.Crypto, crypto), diff --git a/apps/server/src/provider/Layers/CodexBackgroundTasks.test.ts b/apps/server/src/provider/Layers/CodexBackgroundTasks.test.ts new file mode 100644 index 000000000000..a48de96e3f9d --- /dev/null +++ b/apps/server/src/provider/Layers/CodexBackgroundTasks.test.ts @@ -0,0 +1,185 @@ +import { describe, expect, it } from "vite-plus/test"; +import { CodexBackgroundTasks, supportsCodexMonitoring } from "./CodexBackgroundTasks.ts"; +import { make as makeLiveness } from "../../orchestration/ThreadBackgroundLiveness.ts"; + +const command = (id = "watch", monitor = true) => ({ + id, + processId: `process-${id}`, + source: "unifiedExecStartup", + command: monitor ? "watch-ci" : "dev-server", + commandActions: [{ command: monitor ? "watch-ci" : "dev-server" }], + exitCode: 0, +}); + +describe("Codex background tasks", () => { + it("keeps Monitoring until the last background shell exits", () => { + const tasks = new CodexBackgroundTasks(); + const liveness = makeLiveness(); + for (const id of ["one", "two"]) { + const task = tasks.started(command(id, false))!; + liveness.recordTaskLiveness({ + threadId: "thread", + ...task, + taskType: "shell", + kind: "started", + }); + } + expect(liveness.getThreadBackgroundLiveness("thread")).toBe("monitoring"); + const first = tasks.completed(command("one"))!; + liveness.recordTaskLiveness({ + threadId: "thread", + ...first, + taskType: "shell", + kind: "completed", + }); + expect(liveness.getThreadBackgroundLiveness("thread")).toBe("monitoring"); + const last = tasks.completed(command("two"))!; + liveness.recordTaskLiveness({ + threadId: "thread", + ...last, + taskType: "shell", + kind: "completed", + }); + expect(liveness.getThreadBackgroundLiveness("thread")).toBeNull(); + expect(tasks.takeWake()).toBeUndefined(); + }); + + it("subscribes by native process ID and discards queued output on unsubscribe", () => { + const tasks = new CodexBackgroundTasks(); + tasks.started(command()); + tasks.output("watch", "before subscription\n"); + expect(tasks.takeWake()).toBeUndefined(); + expect(tasks.subscribe("unknown")).toBe(false); + expect(tasks.subscribe("process-watch")).toBe(true); + tasks.output("watch", "event\n"); + tasks.completed(command()); + tasks.unsubscribe("process-watch"); + expect(tasks.takeWake()).toBeUndefined(); + }); + + it("frames split lines and coalesces pending events without replaying drained output", () => { + const tasks = new CodexBackgroundTasks(); + tasks.started(command()); + tasks.subscribe("process-watch"); + tasks.output("watch", "CI pass"); + expect(tasks.takeWake()).toBeUndefined(); + tasks.output("watch", "ed\r\nDeploy ready\npartial"); + expect(tasks.takeWake()!).toEqual({ + taskId: "watch", + processId: "process-watch", + output: "CI passed\nDeploy ready", + }); + expect(tasks.takeWake()).toBeUndefined(); + tasks.completed(command()); + expect(tasks.takeWake()!.output).toBe("partial\nWatcher exited with code 0."); + expect(tasks.completed(command())).toBeUndefined(); + expect(tasks.takeWake()).toBeUndefined(); + }); + + it("requires an explicit subscription and ignores interaction items", () => { + const tasks = new CodexBackgroundTasks(); + expect(tasks.started({ ...command(), source: "unifiedExecInteraction" })).toBeUndefined(); + expect(tasks.started({ ...command(), processId: null })).toBeUndefined(); + tasks.started(command()); + expect(tasks.started(command())).toBeUndefined(); + tasks.subscribe("process-watch"); + expect( + tasks.completed({ ...command("interaction"), source: "unifiedExecInteraction" }), + ).toBeUndefined(); + tasks.output("watch", "event\n"); + expect(tasks.takeWake()).toBeDefined(); + tasks.started({ + ...command("echo"), + command: "echo watch-ci", + }); + tasks.output("echo", "no wake\n"); + expect(tasks.takeWake()).toBeUndefined(); + }); + + it("suppresses queued and shutdown output after Stop, including late starts", () => { + const tasks = new CodexBackgroundTasks(); + tasks.started(command()); + tasks.subscribe("process-watch"); + tasks.output("watch", "pending\n"); + tasks.cancelWakes(); + tasks.started(command("late")); + tasks.output("watch", "shutdown\n"); + tasks.output("late", "shutdown\n"); + tasks.completed({ ...command(), exitCode: -1 }); + expect(tasks.takeWake()).toBeUndefined(); + expect(tasks.stop()).toEqual([ + { taskId: "late", description: command("late").command, status: "stopped" }, + ]); + expect(tasks.stop()).toEqual([]); + tasks.output("late", "delayed\n"); + expect(tasks.takeWake()).toBeUndefined(); + }); + + it("bounds long lines and noisy output while waiting for an idle turn", () => { + const tasks = new CodexBackgroundTasks(); + tasks.started(command()); + tasks.subscribe("process-watch"); + for (let i = 0; i < 100; i++) tasks.output("watch", "x".repeat(1000)); + tasks.output("watch", "\nnext event\n"); + const output = tasks.takeWake()!.output; + expect(output.length).toBeLessThan(8300); + expect(output).toContain("omitted"); + }); + + it("marks a single oversized line as truncated", () => { + const tasks = new CodexBackgroundTasks(); + tasks.started(command()); + tasks.subscribe("process-watch"); + tasks.output("watch", "x".repeat(9000) + "\n"); + expect(tasks.takeWake()?.output).toBe("x".repeat(8192) + "\n[Further watcher output omitted]"); + }); + + it("restores a rejected event ahead of later output and still supports unsubscribe", () => { + const tasks = new CodexBackgroundTasks(); + tasks.started(command()); + tasks.subscribe("process-watch"); + tasks.output("watch", "first\n"); + const rejected = tasks.takeWake()!; + tasks.output("watch", "second\n"); + tasks.restoreWake(rejected); + expect(tasks.takeWake()?.output).toBe("first\nsecond"); + tasks.restoreWake(rejected); + tasks.unsubscribe("process-watch"); + expect(tasks.takeWake()).toBeUndefined(); + }); + + it.each([false, true])("unsubscribe invalidates an in-flight wake (exited: %s)", (exited) => { + const tasks = new CodexBackgroundTasks(); + tasks.started(command()); + tasks.subscribe("process-watch"); + tasks.output("watch", "old event\n"); + if (exited) tasks.completed(command()); + const wake = tasks.takeWake()!; + tasks.unsubscribe("process-watch"); + tasks.subscribe("process-watch"); + tasks.restoreWake(wake); + expect(tasks.takeWake()).toBeUndefined(); + }); + + it("retains a rejected process-exit event when it has not been unsubscribed", () => { + const tasks = new CodexBackgroundTasks(); + tasks.started(command()); + tasks.subscribe("process-watch"); + tasks.completed(command()); + const wake = tasks.takeWake()!; + tasks.restoreWake(wake); + expect(tasks.takeWake()?.output).toBe("Watcher exited with code 0."); + }); + + it.each([ + ["t3/0.146.0 (linux)", false], + ["t3/0.153.1 (linux)", false], + ["t3/0.153.2 (linux)", true], + ["t3/0.154.0 (linux)", true], + ["t3/1.0.0 (linux)", true], + ["t3/0.153.2-alpha (linux)", false], + ["unknown", false], + ])("gates experimental wakes for %s", (version, supported) => { + expect(supportsCodexMonitoring(version)).toBe(supported); + }); +}); diff --git a/apps/server/src/provider/Layers/CodexBackgroundTasks.ts b/apps/server/src/provider/Layers/CodexBackgroundTasks.ts new file mode 100644 index 000000000000..b99f7d7fc498 --- /dev/null +++ b/apps/server/src/provider/Layers/CodexBackgroundTasks.ts @@ -0,0 +1,196 @@ +import * as Schema from "effect/Schema"; + +// These experimental fields are absent from the pinned generated bindings. +// Keep the extension at the adapter boundary until the bindings are refreshed. +export const CodexMonitorOutput = Schema.Struct({ name: Schema.String, output: Schema.String }); +export const CodexMonitorTurnInput = Schema.Struct({ + threadId: Schema.String, + input: Schema.Array(Schema.Never), + toolOutput: CodexMonitorOutput, +}); +export const CodexBackgroundCleanResponse = Schema.Struct({}); +export const CodexBackgroundTaskEvent = Schema.Struct({ + taskId: Schema.String, + description: Schema.String, + status: Schema.Literals(["running", "completed", "failed", "stopped"]), +}); + +export function supportsCodexMonitoring(userAgent: string): boolean { + const version = /\/(\d+)\.(\d+)\.(\d+)(?:\s|$)/.exec(userAgent); + if (!version) return false; + const [, major, minor, patch] = version; + return Number(major) > 0 || Number(minor) > 153 || (Number(minor) === 153 && Number(patch) >= 2); +} + +interface Command { + readonly id: string; + readonly processId?: string | null; + readonly source?: string; + readonly command: string; + readonly exitCode?: number | null; +} + +interface BackgroundTask { + readonly taskId: string; + readonly description: string; + readonly processId: string; + monitor: boolean; + remainder: string; +} + +interface PendingWake { + readonly taskId: string; + readonly processId: string; + readonly output: string; +} + +const MAX_EVENT_LENGTH = 8_192; +const MAX_PENDING_EVENTS = 32; +const TRUNCATED = "\n[Further watcher output omitted]"; + +/** Tracks native command lifetimes separately from turns. Only explicitly + * subscribed commands can enqueue wakes; output chunks are not event boundaries. */ +export class CodexBackgroundTasks { + private readonly tasks = new Map(); + private readonly pending = new Map(); + // The runtime delivers one wake at a time. Unsubscribe can invalidate it + // while turn/start is pending, including after the process has exited. + private inFlightWake: PendingWake | undefined; + + started(command: Command): typeof CodexBackgroundTaskEvent.Type | undefined { + if ( + command.source !== "unifiedExecStartup" || + !command.processId || + this.tasks.has(command.id) + ) { + return; + } + return this.register(command.id, command.processId, command.command, false); + } + + register( + taskId: string, + processId: string, + description: string, + monitor = true, + ): typeof CodexBackgroundTaskEvent.Type { + const task = { + taskId, + processId, + description, + monitor, + remainder: "", + }; + this.tasks.set(taskId, task); + return { taskId: task.taskId, description: task.description, status: "running" }; + } + + subscribe(processId: string): boolean { + const task = Array.from(this.tasks.values()).find((task) => task.processId === processId); + if (!task) return false; + task.monitor = true; + return true; + } + + unsubscribe(processId: string): void { + if (this.inFlightWake?.processId === processId) this.inFlightWake = undefined; + for (const [id, event] of this.pending) + if (event.processId === processId) this.pending.delete(id); + for (const task of this.tasks.values()) { + if (task.processId !== processId) continue; + task.monitor = false; + task.remainder = ""; + this.pending.delete(task.taskId); + } + } + + output(itemId: string, delta: string): void { + const task = this.tasks.get(itemId); + if (!task?.monitor) return; + // Bound both an unterminated line and a burst while the foreground is busy. + let offset = 0; + for (;;) { + const newline = delta.indexOf("\n", offset); + const end = newline === -1 ? delta.length : newline; + task.remainder = (task.remainder + delta.slice(offset, end)).slice(0, MAX_EVENT_LENGTH + 1); + if (newline === -1) break; + this.enqueue(task, task.remainder.replace(/\r$/, "")); + task.remainder = ""; + offset = end + 1; + } + } + + completed(command: Command): typeof CodexBackgroundTaskEvent.Type | undefined { + // Interaction items (write_stdin) must not finish the owning startup item. + const task = this.tasks.get(command.id); + if (!task) return; + this.tasks.delete(command.id); + if (command.exitCode === -1) { + this.pending.delete(command.id); + if (this.inFlightWake?.taskId === command.id) this.inFlightWake = undefined; + } + if (task.monitor && command.exitCode !== -1) { + this.enqueue(task, task.remainder); + this.enqueue(task, `Watcher exited with code ${command.exitCode ?? "unknown"}.`); + } + return { + taskId: task.taskId, + description: task.description, + status: command.exitCode === -1 ? "stopped" : command.exitCode === 0 ? "completed" : "failed", + }; + } + + private enqueue(task: Pick, line: string): void { + if (!line.trim()) return; + const previous = this.pending.get(task.taskId)?.output; + if (previous === undefined && this.pending.size >= MAX_PENDING_EVENTS) return; + if (previous?.endsWith(TRUNCATED)) return; + const next = previous ? `${previous}\n${line}` : line; + this.pending.set(task.taskId, { + processId: task.processId, + output: next.length > MAX_EVENT_LENGTH ? next.slice(0, MAX_EVENT_LENGTH) + TRUNCATED : next, + }); + } + + takeWake(): PendingWake | undefined { + const entry = this.pending.entries().next().value; + if (!entry) return; + const [taskId, event] = entry; + this.pending.delete(taskId); + return (this.inFlightWake = { taskId, ...event }); + } + + restoreWake(wake: PendingWake): void { + if (this.inFlightWake !== wake) return; + this.inFlightWake = undefined; + const later = this.pending.get(wake.taskId)?.output; + // Reserve the failed event's place within the same bounded queue. + if (!this.pending.has(wake.taskId) && this.pending.size >= MAX_PENDING_EVENTS) { + const last = Array.from(this.pending.keys()).at(-1); + if (last !== undefined) this.pending.delete(last); + } + this.pending.delete(wake.taskId); + this.enqueue(wake, wake.output); + if (later) this.enqueue(wake, later); + } + + /** Disable wakes before interrupting the provider: termination output is + * not a new event to act on. Keep liveness until actual completion arrives. */ + cancelWakes(): void { + this.inFlightWake = undefined; + this.pending.clear(); + for (const [id, task] of this.tasks) + this.tasks.set(id, { ...task, monitor: false, remainder: "" }); + } + + stop(): ReadonlyArray { + this.cancelWakes(); + const stopped = Array.from(this.tasks.values(), ({ taskId, description }) => ({ + taskId, + description, + status: "stopped" as const, + })); + this.tasks.clear(); + return stopped; + } +} diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 02b7a45f33ad..4e8c461e8fef 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -17,6 +17,8 @@ import { it } from "@effect/vitest"; import { type ProviderApprovalDecision, type ProviderEvent, ThreadId } from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as MonitorSession from "../../mcp/MonitorSession.ts"; import * as Fiber from "effect/Fiber"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; @@ -233,7 +235,10 @@ describe("CodexSessionRuntime collab integration", () => { ]); yield* runtime.close; - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer)), + ), ); it.effect("keeps child settings and reroutes newer than the resume snapshot", () => @@ -341,7 +346,10 @@ describe("CodexSessionRuntime collab integration", () => { assert.equal(readRecordedRequests().length, 1); yield* runtime.close; - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer)), + ), ); it.effect("does not delay the parent turn when the child lookup fails", () => @@ -397,7 +405,10 @@ describe("CodexSessionRuntime collab integration", () => { NodeFS.rmSync(`${scriptPath}.requests`, { force: true }); }).pipe(Effect.scoped); } - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer)), + ), ); it.effect("replays the captured fan-out into synthetic agent events without child leaks", () => @@ -470,7 +481,10 @@ describe("CodexSessionRuntime collab integration", () => { ); yield* runtime.close; - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer)), + ), ); // it.live: the runtime talks to a real child process; under it.effect's @@ -602,7 +616,10 @@ describe("CodexSessionRuntime collab integration", () => { assert.isTrue(interruptedThreads.has(ROOT), "parent turn must be interrupted last"); yield* runtime.close; - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer)), + ), ); it.live("Stop targets the active turn when Codex has accepted a queued follow-up", () => @@ -651,7 +668,10 @@ describe("CodexSessionRuntime collab integration", () => { }); yield* runtime.close; - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer)), + ), ); const elicitationCases = [ @@ -762,7 +782,10 @@ describe("CodexSessionRuntime collab integration", () => { assert.deepEqual(recordedResponse.result, response); yield* runtime.close; - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer)), + ), ); } }); diff --git a/apps/server/src/provider/Layers/CodexMonitoringRuntime.test.ts b/apps/server/src/provider/Layers/CodexMonitoringRuntime.test.ts new file mode 100644 index 000000000000..2bf8b89b3cf1 --- /dev/null +++ b/apps/server/src/provider/Layers/CodexMonitoringRuntime.test.ts @@ -0,0 +1,363 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Schema from "effect/Schema"; +import * as Fiber from "effect/Fiber"; +import * as TestClock from "effect/testing/TestClock"; +import * as Stream from "effect/Stream"; +import { ThreadId, type ProviderEvent, type RuntimeMode } from "@t3tools/contracts"; +import * as MonitorSession from "../../mcp/MonitorSession.ts"; +import { makeCodexSessionRuntime } from "./CodexSessionRuntime.ts"; + +const decodeInspection = Schema.decodeUnknownSync( + Schema.fromJsonString( + Schema.Struct({ + wakes: Schema.Array(Schema.Struct({ name: Schema.String, output: Schema.String })), + cleanCount: Schema.Number, + interrupted: Schema.Number, + terminatedMonitors: Schema.Number, + monitorExecutions: Schema.Array( + Schema.Struct({ + command: Schema.Array(Schema.String), + processId: Schema.String, + sandboxPolicy: Schema.Struct({ type: Schema.String }), + streamStdoutStderr: Schema.Boolean, + }), + ), + }), + ), +); + +const setup = Effect.fn("setup")(function* ( + version = "0.153.2", + mcp = false, + runtimeMode: RuntimeMode = "full-access", +) { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const cwd = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-monitor-runtime-test-" }); + // The runtime invokes app-server; Node executes this local fixture. + yield* fileSystem.copyFile( + yield* path.fromFileUrl(new URL("../testFixtures/codexMonitorAppServer.cjs", import.meta.url)), + path.join(cwd, "app-server"), + ); + const runtime = yield* makeCodexSessionRuntime({ + threadId: ThreadId.make("monitor-test"), + binaryPath: process.execPath, + mcpProviderSessionId: cwd, + ...(mcp ? { appServerArgs: ["-c", "mcp_servers.t3-code.url=http://localhost/mcp"] } : {}), + cwd, + runtimeMode, + environment: { ...process.env, T3_MONITOR_TEST_VERSION: version }, + }); + const events = yield* Queue.unbounded(); + yield* runtime.events.pipe( + Stream.runForEach((event) => Queue.offer(events, event)), + Effect.forkChild, + ); + yield* runtime.start(); + const until = Effect.fn("until")(function* (method: string) { + for (;;) { + const event = yield* Queue.take(events); + if (event.method === method) return event; + } + }); + const inspect = runtime.readThread.pipe( + Effect.map((snapshot) => { + const item = snapshot.turns[0]?.items[0]; + assert.isDefined(item); + assert.equal(item.type, "agentMessage"); + if (item.type !== "agentMessage") throw new Error("Expected inspection message"); + return decodeInspection(item.text); + }), + ); + const subscribe = (yield* MonitorSession.MonitorSessions).invoke(cwd, "subscribe", "42"); + const sessions = yield* MonitorSession.MonitorSessions; + const startMonitor = (command: ReadonlyArray) => sessions.start(cwd, command); + return { runtime, until, inspect, subscribe, startMonitor }; +}); + +it.effect( + "starts and subscribes atomically, captures immediate output, and terminates on Stop", + () => + Effect.gen(function* () { + const { runtime, until, inspect, startMonitor } = yield* setup(); + yield* runtime.sendTurn({ input: "watch" }); + yield* until("turn/completed"); + const monitor = yield* startMonitor(["watch-ci"]); + assert.equal(monitor.status, "scheduled"); + yield* until("turn/completed"); + const snapshot = yield* inspect; + assert.deepStrictEqual(snapshot.monitorExecutions, [ + { + command: ["watch-ci"], + processId: monitor.monitorId, + sandboxPolicy: { type: "dangerFullAccess" }, + streamStdoutStderr: true, + }, + ]); + assert.include(snapshot.wakes[0]!.output, "Immediate monitor event"); + yield* runtime.interruptTurn(); + assert.equal((yield* inspect).terminatedMonitors, 1); + assert.equal((yield* inspect).wakes.length, 1); + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer)), + ), +); + +it.effect("stops a quiet monitor before its first output and preserves the thread sandbox", () => + Effect.gen(function* () { + const { runtime, until, inspect, startMonitor } = yield* setup( + "0.153.2", + false, + "approval-required", + ); + yield* runtime.sendTurn({ input: "watch" }); + yield* until("turn/completed"); + yield* startMonitor(["quiet"]); + yield* runtime.interruptTurn(); + const snapshot = yield* inspect; + assert.equal(snapshot.terminatedMonitors, 1); + assert.equal(snapshot.wakes.length, 0); + assert.deepStrictEqual(snapshot.monitorExecutions[0]?.sandboxPolicy, { type: "readOnly" }); + assert.equal((yield* startMonitor(["quiet"]).pipe(Effect.result))._tag, "Failure"); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer))), +); + +it.effect("reports a launch failure through the monitor wake", () => + Effect.gen(function* () { + const { runtime, until, inspect, startMonitor } = yield* setup(); + yield* runtime.sendTurn({ input: "watch" }); + yield* until("turn/completed"); + yield* startMonitor(["fail"]); + yield* until("turn/completed"); + assert.include((yield* inspect).wakes[0]!.output, "Watcher failed to start or execute"); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer))), +); + +it.effect("wakes an idle thread from tool output and stops without a shutdown wake", () => + Effect.gen(function* () { + const { runtime, until, inspect, subscribe } = yield* setup(); + yield* runtime.sendTurn({ input: "watch" }); + const task = yield* until("backgroundTask/changed"); + assert.deepStrictEqual(task.payload, { + taskId: "watch-command", + description: "watch-ci", + status: "running", + }); + yield* until("turn/completed"); + yield* subscribe; + yield* runtime.compactThread; + yield* until("turn/completed"); + assert.deepStrictEqual((yield* inspect).wakes, [ + { + name: "background_monitor", + output: '{"taskId":"watch-command","output":"CI passed"}', + }, + ]); + yield* runtime.interruptTurn(); + yield* until("item/completed"); + const final = yield* inspect; + assert.equal(final.cleanCount, 1); + assert.equal(final.wakes.length, 1); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer))), +); + +it.effect("delivers wakes when a user turn completes before its start response", () => + Effect.gen(function* () { + const { runtime, until, inspect, subscribe } = yield* setup(); + const sending = yield* runtime.sendTurn({ input: "early-completion" }).pipe(Effect.forkChild); + yield* until("turn/completed"); + yield* runtime.compactThread; + yield* Fiber.join(sending); + assert.equal((yield* runtime.getSession).activeTurnId, undefined); + yield* subscribe; + yield* runtime.compactThread; + yield* until("turn/completed"); + assert.equal((yield* inspect).wakes.length, 1); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer))), +); + +it.effect("queues watcher events until the foreground turn completes", () => + Effect.gen(function* () { + const { runtime, until, inspect, subscribe } = yield* setup(); + yield* runtime.sendTurn({ input: "busy" }); + yield* until("item/started"); + yield* subscribe; + yield* runtime.compactThread; + yield* until("thread/name/updated"); + assert.equal((yield* inspect).wakes.length, 0); + yield* runtime.compactThread; + yield* until("turn/completed"); + yield* until("turn/completed"); + assert.equal((yield* inspect).wakes.length, 1); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer))), +); + +it.effect("Stop drops queued events and still interrupts if terminal cleanup fails", () => + Effect.gen(function* () { + const { runtime, until, inspect, subscribe } = yield* setup(); + yield* runtime.sendTurn({ input: "cleanup-failure" }); + yield* until("item/started"); + yield* subscribe; + yield* runtime.compactThread; + yield* until("thread/name/updated"); + const result = yield* runtime.interruptTurn().pipe(Effect.result); + assert.equal(result._tag, "Failure"); + const stopped = yield* until("backgroundTask/changed"); + assert.deepStrictEqual(stopped.payload, { + taskId: "watch-command", + description: "watch-ci", + status: "stopped", + }); + yield* until("turn/completed"); + const final = yield* inspect; + assert.equal(final.interrupted, 1); + assert.equal(final.wakes.length, 0); + yield* runtime.sendTurn({ input: "resume" }); + yield* until("turn/completed"); + assert.equal((yield* subscribe.pipe(Effect.result))._tag, "Failure"); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer))), +); + +it.effect("flushes a final partial event when the process exits", () => + Effect.gen(function* () { + const { runtime, until, inspect, subscribe } = yield* setup(); + yield* runtime.sendTurn({ input: "exit" }); + yield* until("turn/completed"); + yield* subscribe; + yield* runtime.compactThread; + yield* until("turn/completed"); + assert.include((yield* inspect).wakes[0]!.output, "partial"); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer))), +); + +it.effect("does not register monitoring on unsupported Codex versions", () => + Effect.gen(function* () { + const { runtime, until, inspect, subscribe } = yield* setup("0.146.0"); + yield* runtime.sendTurn({ input: "watch" }); + yield* until("turn/completed"); + assert.equal((yield* subscribe.pipe(Effect.result))._tag, "Failure"); + yield* runtime.compactThread; + yield* until("thread/name/updated"); + yield* runtime.interruptTurn(); + const final = yield* inspect; + assert.equal(final.cleanCount, 0); + assert.equal(final.wakes.length, 0); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer))), +); + +it.effect("never delivers a child command's output to the parent monitor", () => + Effect.gen(function* () { + const { runtime, until, inspect, subscribe } = yield* setup(); + yield* runtime.sendTurn({ input: "child" }); + yield* until("turn/completed"); + yield* subscribe; + yield* runtime.compactThread; + yield* until("thread/name/updated"); + assert.equal((yield* inspect).wakes.length, 0); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer))), +); + +for (const scenario of ["stall-turn", "stall-reload"]) { + it.effect(`Stop reaches the provider when ${scenario} never responds`, () => + Effect.gen(function* () { + const { runtime, until, inspect } = yield* setup("0.153.2", scenario === "stall-reload"); + yield* runtime.sendTurn({ input: scenario }); + yield* until("item/started"); + const sending = yield* runtime + .sendTurn({ input: "follow up" }) + .pipe(Effect.result, Effect.forkChild); + yield* until("thread/name/updated"); + const stopping = yield* runtime.interruptTurn().pipe(Effect.forkChild); + yield* TestClock.adjust("10 seconds"); + assert.equal((yield* Fiber.join(sending))._tag, "Failure"); + yield* Fiber.join(stopping); + const final = yield* inspect; + assert.equal(final.cleanCount, 1); + assert.equal(final.interrupted, 1); + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer)), + ), + ); +} + +it.effect("retains an explicitly rejected wake until a user turn resumes delivery", () => + Effect.gen(function* () { + const { runtime, until, inspect, subscribe } = yield* setup(); + yield* runtime.sendTurn({ input: "reject-wake" }); + yield* until("turn/completed"); + yield* subscribe; + yield* runtime.compactThread; + const failed = yield* until("backgroundMonitor/wakeFailed"); + assert.deepStrictEqual(failed.payload, { + monitorEvent: '{"taskId":"watch-command","output":"CI passed"}', + }); + assert.equal((yield* inspect).wakes.length, 0); + yield* runtime.sendTurn({ input: "resume" }); + yield* until("turn/completed"); + yield* until("turn/completed"); + assert.deepStrictEqual((yield* inspect).wakes, [ + { name: "background_monitor", output: '{"taskId":"watch-command","output":"CI passed"}' }, + ]); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer))), +); + +for (const scenario of ["reject-resume", "timeout-resume"]) { + it.effect(`keeps wakes suppressed after ${scenario} until a successful user turn`, () => + Effect.gen(function* () { + const { runtime, until, inspect, subscribe } = yield* setup(); + yield* runtime.sendTurn({ input: "reject-wake" }); + yield* until("turn/completed"); + yield* subscribe; + yield* runtime.compactThread; + yield* until("backgroundMonitor/wakeFailed"); + const sending = yield* runtime + .sendTurn({ input: scenario }) + .pipe(Effect.result, Effect.forkChild); + if (scenario === "timeout-resume") { + yield* until("thread/name/updated"); + yield* TestClock.adjust("10 seconds"); + } + assert.equal((yield* Fiber.join(sending))._tag, "Failure"); + assert.equal((yield* subscribe.pipe(Effect.result))._tag, "Failure"); + assert.equal((yield* inspect).wakes.length, 0); + yield* runtime.sendTurn({ input: "resume" }); + yield* until("turn/completed"); + yield* until("turn/completed"); + assert.deepStrictEqual((yield* inspect).wakes, [ + { name: "background_monitor", output: '{"taskId":"watch-command","output":"CI passed"}' }, + ]); + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer)), + ), + ); +} + +it.effect("preserves timed-out wake evidence without retrying an ambiguous delivery", () => + Effect.gen(function* () { + const { runtime, until, inspect, subscribe } = yield* setup(); + yield* runtime.sendTurn({ input: "timeout-wake" }); + yield* until("turn/completed"); + yield* subscribe; + yield* runtime.compactThread; + // The fixture emits one barrier for compact and one for the stalled wake. + yield* until("thread/name/updated"); + yield* until("thread/name/updated"); + yield* TestClock.adjust("10 seconds"); + const failed = yield* until("backgroundMonitor/wakeFailed"); + assert.deepStrictEqual(failed.payload, { + monitorEvent: '{"taskId":"watch-command","output":"CI passed"}', + }); + yield* runtime.sendTurn({ input: "resume" }); + yield* until("turn/completed"); + assert.equal((yield* inspect).wakes.length, 0); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, MonitorSession.layer))), +); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 4b88b7ce01c0..d47dc11e9c76 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -28,6 +28,7 @@ import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; @@ -36,10 +37,24 @@ import * as CodexErrors from "effect-codex-app-server/errors"; import * as CodexRpc from "effect-codex-app-server/rpc"; import * as EffectCodexSchema from "effect-codex-app-server/schema"; +import { + CodexBackgroundTasks, + CodexBackgroundTaskEvent, + CodexBackgroundCleanResponse, + CodexMonitorTurnInput, + supportsCodexMonitoring, +} from "./CodexBackgroundTasks.ts"; import { buildCodexInitializeParams } from "./CodexProvider.ts"; import { codexSessionAppServerArgs } from "./codexLaunchArgs.ts"; import { expandHomePath } from "../../pathExpansion.ts"; import { buildCodexDeveloperInstructions } from "../CodexDeveloperInstructions.ts"; +import * as MonitorSession from "../../mcp/MonitorSession.ts"; +const isCodexRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); +const isMonitorStoppedError = Schema.is(MonitorSession.MonitorStoppedError); +const encodeMonitorWake = Schema.encodeSync( + Schema.fromJsonString(Schema.Struct({ taskId: Schema.String, output: Schema.String })), +); +const decodeBackgroundCleanResponse = Schema.decodeUnknownEffect(CodexBackgroundCleanResponse); const decodeV2TurnStartResponse = Schema.decodeUnknownEffect(EffectCodexSchema.V2TurnStartResponse); const PROVIDER = ProviderDriverKind.make("codex"); @@ -166,6 +181,8 @@ export interface CodexSessionRuntimeOptions { readonly serviceTier?: CodexServiceTier | undefined; readonly resumeCursor?: CodexResumeCursor; readonly appServerArgs?: ReadonlyArray; + readonly mcpProviderSessionId?: string; + readonly browserToolsAvailable?: boolean; } export interface CodexSessionRuntimeSendTurnInput { @@ -1155,10 +1172,14 @@ export const makeCodexSessionRuntime = ( ): Effect.Effect< CodexSessionRuntimeShape, CodexErrors.CodexAppServerError, - ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto | Scope.Scope + | ChildProcessSpawner.ChildProcessSpawner + | Crypto.Crypto + | Scope.Scope + | MonitorSession.MonitorSessions > => Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const monitorSessions = yield* MonitorSession.MonitorSessions; const runtimeScope = yield* Scope.Scope; const crypto = yield* Crypto.Crypto; const events = yield* Queue.unbounded(); @@ -1172,6 +1193,18 @@ export const makeCodexSessionRuntime = ( const collabChildLiveTurnsRef = yield* Ref.make(new Map()); const suppressMemoryConsolidationNotification = makeMemoryConsolidationNotificationFilter(); const closedRef = yield* Ref.make(false); + const backgroundTasks = new CodexBackgroundTasks(); + const monitorCommands = new Map< + string, + { stdout: TextDecoder; stderr: TextDecoder; stopped: boolean } + >(); + const turnLock = yield* Semaphore.make(1); + const wakeSignals = yield* Queue.sliding(1); + let monitoringAvailable = false; + let suppressMonitorWakes = false; + let pendingUserSends = 0; + const queuedUserTurns = new Set(); + let lastCompletedTurnId: string | undefined; // `~` is not shell-expanded when env vars are set via // `child_process.spawn`; `expandHomePath` lets a configured @@ -1725,6 +1758,7 @@ export const makeCodexSessionRuntime = ( const handleRawNotification = (notification: CodexServerNotification) => Effect.gen(function* () { + if (notification.method === "command/exec/outputDelta") return; const isMemoryConsolidationNotification = suppressMemoryConsolidationNotification(notification); @@ -1809,6 +1843,29 @@ export const makeCodexSessionRuntime = ( return; } + if (monitoringAvailable && !foreignConversation && childParentTurnId === undefined) { + let taskEvent: typeof CodexBackgroundTaskEvent.Type | undefined; + if ( + (notification.method === "item/started" || notification.method === "item/completed") && + notification.params.item.type === "commandExecution" + ) { + taskEvent = + notification.method === "item/started" + ? backgroundTasks.started(notification.params.item) + : backgroundTasks.completed(notification.params.item); + } else if (notification.method === "item/commandExecution/outputDelta") { + backgroundTasks.output(notification.params.itemId, notification.params.delta); + } + if (taskEvent) { + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "backgroundTask/changed", + payload: taskEvent, + }); + } + } + let requestId: ApprovalRequestId | undefined; let requestKind: ProviderRequestKind | undefined; let turnId = childParentTurnId ?? route.turnId; @@ -1849,6 +1906,15 @@ export const makeCodexSessionRuntime = ( : {}), ...(payload !== undefined ? { payload } : {}), }); + if ( + monitoringAvailable && + !foreignConversation && + (notification.method === "turn/completed" || + notification.method === "item/commandExecution/outputDelta" || + (notification.method === "item/completed" && + notification.params.item.type === "commandExecution")) + ) + yield* Queue.offer(wakeSignals, undefined); }); const currentSessionProviderThreadId = Effect.map(Ref.get(sessionRef), currentProviderThreadId); @@ -1890,6 +1956,8 @@ export const makeCodexSessionRuntime = ( payload.turn.status === "failed" && "error" in payload.turn && payload.turn.error ? payload.turn.error.message : undefined; + lastCompletedTurnId = payload.turn.id; + queuedUserTurns.delete(payload.turn.id); return updateSession(sessionRef, { status: payload.turn.status === "failed" ? "error" : "ready", activeTurnId: undefined, @@ -2149,6 +2217,20 @@ export const makeCodexSessionRuntime = ( Effect.fail(CodexErrors.CodexAppServerRequestError.methodNotFound(method)), ); + yield* client.handleServerNotification("command/exec/outputDelta", (payload) => + Effect.gen(function* () { + const decoders = monitorCommands.get(payload.processId); + if (!decoders) return; + backgroundTasks.output( + payload.processId, + decoders[payload.stream].decode(Buffer.from(payload.deltaBase64, "base64"), { + stream: true, + }), + ); + yield* Queue.offer(wakeSignals, undefined); + }), + ); + const registerServerNotification = (method: M) => client.handleServerNotification(method, (params) => Queue.offer(serverNotifications, makeCodexServerNotification(method, params)).pipe( @@ -2231,7 +2313,8 @@ export const makeCodexSessionRuntime = ( const start = Effect.fn("CodexSessionRuntime.start")(function* () { yield* emitSessionEvent("session/connecting", "Starting Codex App Server session."); - yield* client.request("initialize", buildCodexInitializeParams()); + const initialized = yield* client.request("initialize", buildCodexInitializeParams()); + monitoringAvailable = supportsCodexMonitoring(initialized.userAgent); yield* client.notify("initialized", undefined); const requestedModel = normalizeCodexModelSlug(options.model); @@ -2270,11 +2353,194 @@ export const makeCodexSessionRuntime = ( return providerThreadId; }); + const wakeMonitor = turnLock.withPermit( + Effect.gen(function* () { + const session = yield* Ref.get(sessionRef); + if ( + !monitoringAvailable || + suppressMonitorWakes || + pendingUserSends > 0 || + queuedUserTurns.size > 0 || + session.status !== "ready" || + session.activeTurnId || + (yield* Ref.get(closedRef)) || + (yield* Ref.get(pendingApprovalsRef)).size > 0 || + (yield* Ref.get(pendingUserInputsRef)).size > 0 + ) + return; + const wake = backgroundTasks.takeWake(); + if (wake === undefined) return; + const output = encodeMonitorWake({ taskId: wake.taskId, output: wake.output }); + yield* Effect.gen(function* () { + const params = CodexMonitorTurnInput.make({ + threadId: yield* readProviderThreadId, + input: [], + toolOutput: { name: "background_monitor", output }, + }); + const raw = yield* client.raw.request("turn/start", params).pipe( + Effect.timeout("10 seconds"), + Effect.tapError((error) => + Effect.sync(() => { + // An explicit rejection is safe to retry. A timeout may have + // been accepted, so preserve its evidence without replaying it. + if (!suppressMonitorWakes && isCodexRequestError(error)) + backgroundTasks.restoreWake(wake); + }), + ), + ); + const response = yield* decodeV2TurnStartResponse(raw).pipe( + Effect.mapError((error) => + CodexErrors.CodexAppServerProtocolParseError.fromSchemaError( + "decode-response-payload", + error, + { method: "turn/start" }, + ), + ), + ); + yield* updateSession(sessionRef, (current) => ({ + status: "running", + activeTurnId: current.activeTurnId ?? TurnId.make(response.turn.id), + })); + // The pinned item union predates functionCallOutput. Publish the + // delivered event in the same timeline as the automated response. + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + turnId: TurnId.make(response.turn.id), + itemId: ProviderItemId.make(`monitor-event:${response.turn.id}`), + method: "backgroundMonitor/delivered", + payload: params.toolOutput, + }); + }).pipe( + Effect.catch((cause) => + Effect.gen(function* () { + // A rejected wake must not retry indefinitely or hide behind Monitoring. + suppressMonitorWakes = true; + yield* emitEvent({ + kind: "error", + threadId: options.threadId, + method: "backgroundMonitor/wakeFailed", + payload: { monitorEvent: output }, + message: "Could not wake Codex for a background monitor. Send a message to resume.", + }); + yield* Effect.logWarning("Codex monitor wake failed", { cause }); + }), + ), + ); + }), + ); + yield* Stream.fromQueue(wakeSignals).pipe( + Stream.runForEach(() => wakeMonitor), + Effect.forkIn(runtimeScope), + ); + + if (options.mcpProviderSessionId) { + yield* monitorSessions.register(options.mcpProviderSessionId, { + start: (command) => + turnLock + .withPermit( + Effect.gen(function* () { + if (!monitoringAvailable || suppressMonitorWakes || (yield* Ref.get(closedRef))) + return yield* new MonitorSession.MonitorStoppedError({}); + const monitorId = yield* randomUUIDv4("provider-event"); + const description = command.join(" "); + const task = backgroundTasks.register(monitorId, monitorId, description); + monitorCommands.set(monitorId, { + stdout: new TextDecoder(), + stderr: new TextDecoder(), + stopped: false, + }); + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "backgroundTask/changed", + payload: task, + }); + // The native RPC replies at exit, not startup. Capture is registered + // first and the tool explicitly reports scheduled rather than running. + yield* client + .request("command/exec", { + command, + processId: monitorId, + cwd: options.cwd, + sandboxPolicy: runtimeModeToTurnSandboxPolicy(options.runtimeMode), + streamStdoutStderr: true, + disableTimeout: true, + disableOutputCap: true, + }) + .pipe( + Effect.matchEffect({ + onFailure: (cause) => + Effect.logWarning("Codex monitor command failed", { cause }).pipe( + Effect.andThen( + Effect.sync(() => { + backgroundTasks.output( + monitorId, + "Watcher failed to start or execute.\n", + ); + return 1; + }), + ), + ), + onSuccess: ({ exitCode }) => Effect.succeed(exitCode), + }), + Effect.flatMap((exitCode) => + Effect.gen(function* () { + const decoders = monitorCommands.get(monitorId); + if (decoders) { + backgroundTasks.output( + monitorId, + decoders.stdout.decode() + decoders.stderr.decode(), + ); + } + const completed = backgroundTasks.completed({ + id: monitorId, + command: description, + exitCode: decoders?.stopped ? -1 : exitCode, + }); + if (completed) + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "backgroundTask/changed", + payload: completed, + }); + yield* Queue.offer(wakeSignals, undefined); + }), + ), + Effect.ensuring( + Effect.sync(() => { + monitorCommands.delete(monitorId); + }), + ), + Effect.forkIn(runtimeScope, { startImmediately: true }), + ); + return { monitorId, status: "scheduled" as const }; + }), + ) + .pipe( + Effect.mapError((cause) => + isMonitorStoppedError(cause) + ? cause + : new MonitorSession.MonitorStartError({ cause }), + ), + ), + subscribe: Effect.fn("CodexSessionRuntime.subscribeMonitor")(function* (processId) { + if (!monitoringAvailable || suppressMonitorWakes || (yield* Ref.get(closedRef))) + return yield* new MonitorSession.MonitorStoppedError({}); + if (!backgroundTasks.subscribe(processId)) + return yield* new MonitorSession.MonitorProcessMissingError({ processId }); + }), + unsubscribe: (processId) => Effect.sync(() => backgroundTasks.unsubscribe(processId)), + }); + } + const close = Effect.gen(function* () { const alreadyClosed = yield* Ref.getAndSet(closedRef, true); if (alreadyClosed) { return; } + backgroundTasks.stop(); yield* settlePendingApprovals("cancel"); yield* settlePendingUserInputs({}); yield* updateSession(sessionRef, { @@ -2299,95 +2565,190 @@ export const makeCodexSessionRuntime = ( yield* client.request("thread/compact/start", { threadId: providerThreadId }); }), sendTurn: (input) => - Effect.gen(function* () { - const providerThreadId = yield* readProviderThreadId; - if (hasConfiguredMcpServer(options.appServerArgs)) { - yield* client.request("config/mcpServer/reload", undefined).pipe( - Effect.catch((cause) => - Effect.logWarning("Failed to refresh Codex MCP tool catalog before turn.", { - cause, + Effect.acquireUseRelease( + Effect.sync(() => { + pendingUserSends += 1; + }), + () => + turnLock.withPermit( + Effect.gen(function* () { + const providerThreadId = yield* readProviderThreadId; + if (hasConfiguredMcpServer(options.appServerArgs)) { + yield* client.request("config/mcpServer/reload", undefined).pipe( + Effect.catch((cause) => + Effect.logWarning("Failed to refresh Codex MCP tool catalog before turn.", { + cause, + }), + ), + ); + } + const normalizedModel = normalizeCodexModelSlug( + input.model ?? (yield* Ref.get(sessionRef)).model, + ); + const params = yield* buildTurnStartParams({ + threadId: providerThreadId, + runtimeMode: options.runtimeMode, + ...(input.input ? { prompt: input.input } : {}), + ...(input.attachments ? { attachments: input.attachments } : {}), + ...(normalizedModel ? { model: normalizedModel } : {}), + ...(input.serviceTier ? { serviceTier: input.serviceTier } : {}), + ...(input.effort ? { effort: input.effort } : {}), + ...(input.interactionMode ? { interactionMode: input.interactionMode } : {}), + // Derived from the session's own MCP configuration rather than the + // setting, so the prompt describes the tools this turn actually + // has even if the setting changed after the session started. + browserToolsAvailable: + options.browserToolsAvailable ?? hasConfiguredMcpServer(options.appServerArgs), + }); + const rawResponse = yield* client.raw.request("turn/start", params); + const response = yield* decodeV2TurnStartResponse(rawResponse).pipe( + Effect.mapError((error) => + CodexErrors.CodexAppServerProtocolParseError.fromSchemaError( + "decode-response-payload", + error, + { method: "turn/start" }, + ), + ), + ); + const turnId = TurnId.make(response.turn.id); + suppressMonitorWakes = false; + // A fast turn can complete before its start response reaches us. + if (lastCompletedTurnId !== turnId) { + queuedUserTurns.add(turnId); + yield* updateSession(sessionRef, (session) => ({ + status: "running", + // Codex accepts follow-ups while the current turn is still + // running. The response contains the queued turn id, but + // turn/interrupt only accepts the id that is active now. + activeTurnId: session.activeTurnId ?? turnId, + ...(normalizedModel ? { model: normalizedModel } : {}), + })); + } + const resumedProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef)); + return { + threadId: options.threadId, + turnId, + ...(resumedProviderThreadId + ? { resumeCursor: { threadId: resumedProviderThreadId } } + : {}), + } satisfies ProviderTurnStartResult; + }).pipe( + Effect.timeoutOrElse({ + duration: "10 seconds", + orElse: () => + Effect.fail( + CodexErrors.CodexAppServerRequestError.internalError( + "Timed out starting Codex turn.", + ), + ), }), ), - ); - } - const normalizedModel = normalizeCodexModelSlug( - input.model ?? (yield* Ref.get(sessionRef)).model, - ); - const params = yield* buildTurnStartParams({ - threadId: providerThreadId, - runtimeMode: options.runtimeMode, - ...(input.input ? { prompt: input.input } : {}), - ...(input.attachments ? { attachments: input.attachments } : {}), - ...(normalizedModel ? { model: normalizedModel } : {}), - ...(input.serviceTier ? { serviceTier: input.serviceTier } : {}), - ...(input.effort ? { effort: input.effort } : {}), - ...(input.interactionMode ? { interactionMode: input.interactionMode } : {}), - // Derived from the session's own MCP configuration rather than the - // setting, so the prompt describes the tools this turn actually - // has even if the setting changed after the session started. - browserToolsAvailable: hasConfiguredMcpServer(options.appServerArgs), - }); - const rawResponse = yield* client.raw.request("turn/start", params); - const response = yield* decodeV2TurnStartResponse(rawResponse).pipe( - Effect.mapError((error) => - CodexErrors.CodexAppServerProtocolParseError.fromSchemaError( - "decode-response-payload", - error, - { method: "turn/start" }, - ), ), - ); - const turnId = TurnId.make(response.turn.id); - yield* updateSession(sessionRef, (session) => ({ - status: "running", - // Codex accepts follow-ups while the current turn is still - // running. The response contains the queued turn id, but - // turn/interrupt only accepts the id that is active now. - activeTurnId: session.activeTurnId ?? turnId, - ...(normalizedModel ? { model: normalizedModel } : {}), - })); - const resumedProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef)); - return { - threadId: options.threadId, - turnId, - ...(resumedProviderThreadId - ? { resumeCursor: { threadId: resumedProviderThreadId } } - : {}), - } satisfies ProviderTurnStartResult; - }), + () => + Effect.sync(() => { + pendingUserSends -= 1; + }).pipe(Effect.andThen(Queue.offer(wakeSignals, undefined))), + ), interruptTurn: (turnId) => - Effect.gen(function* () { - const providerThreadId = yield* readProviderThreadId; - const session = yield* Ref.get(sessionRef); - // Stop-everything: children are full threads with their own turns; - // interrupting only the parent leaves the fleet running. Interrupt - // each live child turn first, best-effort per child, BOUNDED: the - // transport awaits an unbounded Deferred per request, so a wedged - // child would otherwise block the parent interrupt forever — - // exactly during the runaway fleet where Stop matters most - // (review finding). Per-child and overall deadlines guarantee the - // parent interrupt below always runs. - const liveChildTurns = yield* Ref.get(collabChildLiveTurnsRef); - yield* Effect.forEach( - Array.from(liveChildTurns.entries()), - ([childThreadId, childTurnId]) => - client - .request("turn/interrupt", { - threadId: childThreadId, - turnId: childTurnId, - }) - .pipe(Effect.timeoutOption("3 seconds"), Effect.ignore), - { concurrency: 8, discard: true }, - ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore); - const effectiveTurnId = turnId ?? session.activeTurnId; - if (!effectiveTurnId) { - return; - } - yield* client.request("turn/interrupt", { - threadId: providerThreadId, - turnId: effectiveTurnId, - }); - }), + Effect.sync(() => { + suppressMonitorWakes = true; + backgroundTasks.cancelWakes(); + }).pipe( + Effect.andThen( + turnLock.withPermit( + Effect.gen(function* () { + const providerThreadId = yield* readProviderThreadId; + suppressMonitorWakes = true; + backgroundTasks.cancelWakes(); + queuedUserTurns.clear(); + for (const monitor of monitorCommands.values()) monitor.stopped = true; + const monitorCleanup = yield* Effect.forEach( + Array.from(monitorCommands.keys()), + (processId) => + client.request("command/exec/terminate", { processId }).pipe( + Effect.timeoutOrElse({ + duration: "10 seconds", + orElse: () => + Effect.fail( + CodexErrors.CodexAppServerRequestError.internalError( + "Timed out stopping Codex monitor.", + ), + ), + }), + Effect.exit, + ), + { concurrency: "unbounded" }, + ); + // Stop-everything: children are full threads with their own turns; + // interrupting only the parent leaves the fleet running. Interrupt + // each live child turn first, best-effort per child, BOUNDED: the + // transport awaits an unbounded Deferred per request, so a wedged + // child would otherwise block the parent interrupt forever — + // exactly during the runaway fleet where Stop matters most + // (review finding). Per-child and overall deadlines guarantee the + // parent interrupt below always runs. + const liveChildTurns = yield* Ref.get(collabChildLiveTurnsRef); + yield* Effect.forEach( + Array.from(liveChildTurns.entries()), + ([childThreadId, childTurnId]) => + client + .request("turn/interrupt", { + threadId: childThreadId, + turnId: childTurnId, + }) + .pipe(Effect.timeoutOption("3 seconds"), Effect.ignore), + { concurrency: 8, discard: true }, + ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore); + const cleaned = yield* Effect.exit( + Effect.gen(function* () { + if (!monitoringAvailable) return; + const raw = yield* client.raw + .request("thread/backgroundTerminals/clean", { + threadId: providerThreadId, + }) + .pipe( + Effect.timeoutOrElse({ + duration: "10 seconds", + orElse: () => + Effect.fail( + CodexErrors.CodexAppServerRequestError.internalError( + "Timed out stopping Codex background terminals.", + ), + ), + }), + ); + yield* decodeBackgroundCleanResponse(raw).pipe( + Effect.mapError((error) => + CodexErrors.CodexAppServerProtocolParseError.fromSchemaError( + "decode-response-payload", + error, + { method: "thread/backgroundTerminals/clean" }, + ), + ), + ); + }), + ); + for (const task of backgroundTasks.stop()) { + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "backgroundTask/changed", + payload: task, + }); + } + const effectiveTurnId = turnId ?? (yield* Ref.get(sessionRef)).activeTurnId; + if (effectiveTurnId) { + yield* client.request("turn/interrupt", { + threadId: providerThreadId, + turnId: effectiveTurnId, + }); + } + yield* cleaned; + for (const result of monitorCleanup) yield* result; + }), + ), + ), + ), readThread: Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; const response = yield* client.request("thread/read", { diff --git a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts index 25dafa5ba040..5f2043448196 100644 --- a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts +++ b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts @@ -39,6 +39,7 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as MonitorSession from "../../mcp/MonitorSession.ts"; import * as Path from "effect/Path"; import * as Stream from "effect/Stream"; import { HttpClient, HttpClientResponse } from "effect/unstable/http"; @@ -232,6 +233,7 @@ describe("ProviderInstanceRegistryLive — multi-instance codex slice", () => { Layer.provideMerge(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), Layer.provideMerge(ModelManifest.layerTest), Layer.provideMerge(CodexResetCredit.layerTest), + Layer.provideMerge(MonitorSession.layer), ); it.live("boots two independent codex instances from a ProviderInstanceConfigMap", () => @@ -460,6 +462,7 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { Layer.provideMerge(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), Layer.provideMerge(ModelManifest.layerTest), Layer.provideMerge(CodexResetCredit.layerTest), + Layer.provideMerge(MonitorSession.layer), ); it.live("boots one instance of every shipped driver from a single config map", () => diff --git a/apps/server/src/provider/Layers/ProviderRegistry.test.ts b/apps/server/src/provider/Layers/ProviderRegistry.test.ts index 988c89e1e679..29d05b287419 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.test.ts @@ -7,6 +7,7 @@ import * as Path from "effect/Path"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as MonitorSession from "../../mcp/MonitorSession.ts"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; @@ -361,7 +362,9 @@ const awaitPersistedProvider = ( Effect.forkScoped, ); -it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), TestHttpClientLive))( +const TestServices = Layer.mergeAll(NodeServices.layer, MonitorSession.layer); + +it.layer(Layer.mergeAll(TestServices, ServerSettingsModule.layerTest(), TestHttpClientLive))( "ProviderRegistry", (it) => { describe("checkCodexProviderStatus", () => { diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index fecd7fca9096..cb092ffc8ed1 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -4317,7 +4317,7 @@ describe("agent browser access", () => { projectOverride?: boolean, ) => Effect.gen(function* () { - const issued: Array = []; + const issued: Array<{ threadId: ThreadId; capabilities?: ReadonlyArray }> = []; const codex = makeFakeCodexAdapter(); const providerAdapterLayer = Layer.succeed( ProviderAdapterRegistry.ProviderAdapterRegistry, @@ -4375,7 +4375,7 @@ describe("agent browser access", () => { const providerLayer = makeProviderServiceLive({ issueMcpCredential: (request) => Effect.sync(() => { - issued.push(request.threadId); + issued.push(request); return undefined; }), revokeMcpCredential: (revoked) => Effect.sync(() => void revokedThreads.push(revoked)), @@ -4413,14 +4413,14 @@ describe("agent browser access", () => { return issued; }); - // Credential issuance is the observable that matters: it is the only place a - // credential is minted, and `/mcp` accepts nothing else, so withholding it is - // what actually denies every provider and external MCP client. - it.effect("requests no MCP credential when agent browser access is off", () => + it.effect("requests only monitoring capability for Codex when browser access is off", () => Effect.gen(function* () { const issued = yield* startSessionWith(false, asThreadId("thread-browser-off")); - assert.deepEqual(issued, []); + assert.deepEqual( + issued.map(({ capabilities }) => capabilities), + [["monitor"]], + ); }).pipe(Effect.provide(NodeServices.layer)), ); @@ -4444,25 +4444,36 @@ describe("agent browser access", () => { const issued = yield* startSessionWith(true, threadId); - assert.deepEqual(issued, [threadId]); + assert.deepEqual( + issued.map(({ capabilities }) => capabilities), + [["preview", "monitor"]], + ); }).pipe(Effect.provide(NodeServices.layer)), ); - it.effect("withholds and revokes MCP credentials when the project disables browser access", () => - Effect.gen(function* () { - const threadId = asThreadId("thread-project-browser-off"); - revokedThreads.length = 0; - const issued = yield* startSessionWith(true, threadId, false); - assert.deepEqual(issued, []); - assert.deepEqual(revokedThreads, [threadId]); - }).pipe(Effect.provide(NodeServices.layer)), + it.effect( + "retains only monitoring and revokes old credentials when the project disables browser access", + () => + Effect.gen(function* () { + const threadId = asThreadId("thread-project-browser-off"); + revokedThreads.length = 0; + const issued = yield* startSessionWith(true, threadId, false); + assert.deepEqual( + issued.map(({ capabilities }) => capabilities), + [["monitor"]], + ); + assert.deepEqual(revokedThreads, [threadId]); + }).pipe(Effect.provide(NodeServices.layer)), ); it.effect("requests an MCP credential when the project overrides browser access to on", () => Effect.gen(function* () { const threadId = asThreadId("thread-project-browser-on"); const issued = yield* startSessionWith(false, threadId, true); - assert.deepEqual(issued, [threadId]); + assert.deepEqual( + issued.map(({ capabilities }) => capabilities), + [["preview", "monitor"]], + ); }).pipe(Effect.provide(NodeServices.layer)), ); }); diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index d9cac46ec4d9..dfbb657dc926 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -46,6 +46,7 @@ import * as SchemaIssue from "effect/SchemaIssue"; import * as Stream from "effect/Stream"; import { resolveAttachmentPath } from "../../attachmentStore.ts"; +import type { McpCapability } from "../../mcp/McpInvocationContext.ts"; import * as ServerConfig from "../../config.ts"; import { increment, @@ -703,14 +704,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }); yield* recordCompletedTurnProperties(properties); }); - /** - * Attach the `t3-code` MCP server to the session that is about to start. - * - * This is the only place a credential is minted, so withholding one here is - * what disables agent browser access everywhere: every adapter already - * treats a missing session as "no MCP server", and the `/mcp` endpoint - * accepts nothing but tokens issued from this path. - */ + // Preview permission is independent of other session-scoped MCP toolkits. /** * Deny on an unreadable settings file rather than letting the read failure * escape: adding `ServerSettingsError` to `ProviderServiceError` would widen @@ -740,20 +734,19 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ), ); - const prepareMcpSession = (threadId: ThreadId, providerInstanceId: ProviderInstanceId) => + const prepareMcpSession = ( + threadId: ThreadId, + providerInstanceId: ProviderInstanceId, + provider: ProviderDriverKind, + ) => Effect.gen(function* () { - if (!(yield* agentBrowserAccessEnabled(threadId))) { - // Revoke as well as clear. Every other prepare path reaches - // `issueActiveMcpCredential`, which revokes the thread first, so - // skipping it here would leave a previously issued bearer token valid - // against `/mcp` for the rest of its liveness window — and later turns - // would keep refreshing it. A session restart (runtime mode, cwd, - // model) re-prepares without stopping, so it relies on this. - yield* revokeMcpCredential(threadId); - yield* Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId)); - return undefined; - } - const credential = yield* issueMcpCredential({ threadId, providerInstanceId }); + const capabilities: Array = []; + if (yield* agentBrowserAccessEnabled(threadId)) capabilities.push("preview"); + if (provider === "codex") capabilities.push("monitor"); + yield* revokeMcpCredential(threadId); + yield* Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId)); + if (capabilities.length === 0) return undefined; + const credential = yield* issueMcpCredential({ threadId, providerInstanceId, capabilities }); if (credential) { yield* Effect.sync(() => McpProviderSession.setMcpProviderSession(credential.config)); } @@ -1042,7 +1035,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const persistedCwd = readPersistedCwd(input.binding.runtimePayload); const persistedModelSelection = readPersistedModelSelection(input.binding.runtimePayload); - yield* prepareMcpSession(input.binding.threadId, bindingInstanceId); + yield* prepareMcpSession(input.binding.threadId, bindingInstanceId, input.binding.provider); const resumed = yield* adapter .startSession({ threadId: input.binding.threadId, @@ -1273,7 +1266,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( } const adapter = yield* registry.getByInstance(resolvedInstanceId); yield* clearTurnAnalyticsSession(resolvedInstanceId, threadId); - yield* prepareMcpSession(threadId, resolvedInstanceId); + yield* prepareMcpSession(threadId, resolvedInstanceId, input.provider); const session = yield* adapter .startSession({ ...input, diff --git a/apps/server/src/provider/testFixtures/codexMonitorAppServer.cjs b/apps/server/src/provider/testFixtures/codexMonitorAppServer.cjs new file mode 100644 index 000000000000..797258fe9621 --- /dev/null +++ b/apps/server/src/provider/testFixtures/codexMonitorAppServer.cjs @@ -0,0 +1,242 @@ +// A deterministic stdio peer for the real Codex session runtime. Each compact +// request advances one notification step; no clocks or model calls are used. +const readline = require("node:readline"); +const threadId = "provider-monitor-thread"; +const cwd = process.cwd(); +const thread = { + id: threadId, + sessionId: threadId, + forkedFromId: null, + preview: "", + ephemeral: true, + modelProvider: "openai", + createdAt: 1, + updatedAt: 1, + status: { type: "idle" }, + path: null, + cwd, + cliVersion: "0.153.2", + source: "vscode", + agentNickname: null, + agentRole: null, + gitInfo: null, + name: null, + turns: [], +}; +const command = { + id: "watch-command", + type: "commandExecution", + processId: "42", + source: "unifiedExecStartup", + command: "watch-ci", + commandActions: [{ type: "unknown", command: "watch-ci" }], + cwd, + status: "inProgress", + aggregatedOutput: null, + exitCode: null, + durationMs: null, +}; +let active; +let original; +let scenario; +let serial = 0; +let step = 0; +let cleanCount = 0; +let interrupted = 0; +let wakeRejected = false; +let earlyTurnReply; +const wakes = []; +const monitorExecutions = []; +const activeMonitors = new Map(); +let terminatedMonitors = 0; +const write = (message) => process.stdout.write(`${JSON.stringify(message)}\n`); +const notify = (method, params) => + write({ + method, + params: { + ...(method === "item/started" ? { startedAtMs: 1000 } : {}), + ...(method === "item/completed" ? { completedAtMs: 2000 } : {}), + ...params, + }, + }); +const reply = (id, result) => write({ id, result }); +const turn = (id, status) => ({ id, status, items: [], error: null }); +const finish = (id) => { + active = undefined; + notify("turn/completed", { threadId, turn: turn(id, "completed") }); +}; +const output = (delta) => + notify("item/commandExecution/outputDelta", { + threadId, + turnId: original, + itemId: command.id, + delta, + }); +const barrier = () => notify("thread/name/updated", { threadId, threadName: `barrier-${step}` }); +readline.createInterface({ input: process.stdin }).on("line", (line) => { + const { id, method, params = {} } = JSON.parse(line); + if (id === undefined) return; + switch (method) { + case "initialize": + reply(id, { + codexHome: cwd, + platformFamily: "unix", + platformOs: "linux", + userAgent: `t3/${process.env.T3_MONITOR_TEST_VERSION ?? "0.153.2"} (test)`, + }); + break; + case "thread/start": + reply(id, { + thread, + model: "gpt-test", + modelProvider: "openai", + cwd, + approvalPolicy: "never", + approvalsReviewer: "user", + sandbox: { type: "dangerFullAccess" }, + reasoningEffort: null, + }); + break; + case "config/mcpServer/reload": + if (scenario === "stall-reload") barrier(); + else reply(id, {}); + break; + case "command/exec": { + monitorExecutions.push(params); + if (params.command[0] === "fail") { + write({ id, error: { code: -32603, message: "Launch failed" } }); + } else { + activeMonitors.set(params.processId, id); + if (params.command[0] !== "quiet") + notify("command/exec/outputDelta", { + processId: params.processId, + stream: "stdout", + capReached: false, + deltaBase64: Buffer.from("Immediate monitor event\n").toString("base64"), + }); + } + barrier(); + break; + } + case "command/exec/terminate": { + terminatedMonitors++; + reply(id, {}); + const requestId = activeMonitors.get(params.processId); + activeMonitors.delete(params.processId); + if (requestId !== undefined) reply(requestId, { exitCode: 0, stdout: "", stderr: "" }); + break; + } + case "turn/start": { + if (params.input?.[0]?.text === "reject-resume") { + write({ id, error: { code: -32603, message: "Resume rejected" } }); + break; + } + if (params.input?.[0]?.text === "timeout-resume") { + barrier(); + break; + } + if (scenario === "stall-turn") { + barrier(); + break; + } + if (params.toolOutput && scenario === "timeout-wake") { + barrier(); + break; + } + if (params.toolOutput && scenario === "reject-wake" && !wakeRejected) { + wakeRejected = true; + write({ id, error: { code: -32603, message: "Wake rejected" } }); + break; + } + active = `turn-${++serial}`; + if (params.input?.[0]?.text === "early-completion") earlyTurnReply = { id, turnId: active }; + else reply(id, { turn: turn(active, "inProgress") }); + notify("turn/started", { threadId, turn: turn(active, "inProgress") }); + if (params.toolOutput) { + wakes.push(params.toolOutput); + finish(active); + } else { + original = active; + scenario = params.input[0]?.text; + if (serial === 1) notify("item/started", { threadId, turnId: active, item: command }); + if (scenario !== "busy" && scenario !== "cleanup-failure" && !scenario.startsWith("stall-")) + finish(active); + } + break; + } + case "thread/compact/start": + if (earlyTurnReply) { + reply(earlyTurnReply.id, { turn: turn(earlyTurnReply.turnId, "completed") }); + earlyTurnReply = undefined; + reply(id, {}); + break; + } + step++; + reply(id, {}); + if (scenario === "busy" && step === 2) finish(original); + else if (scenario === "exit") { + output("partial"); + notify("item/completed", { + threadId, + turnId: original, + item: { ...command, status: "completed", exitCode: 0 }, + }); + } else if (scenario === "child") { + notify("item/commandExecution/outputDelta", { + threadId: "foreign-child", + turnId: original, + itemId: command.id, + delta: "child event\n", + }); + } else output("CI passed\n"); + barrier(); + break; + case "thread/backgroundTerminals/clean": + cleanCount++; + if (scenario === "cleanup-failure") { + write({ id, error: { code: -32603, message: "Cleanup failed" } }); + } else { + reply(id, {}); + output("shutdown output\n"); + notify("item/completed", { + threadId, + turnId: original, + item: { ...command, status: "failed", exitCode: -1 }, + }); + } + break; + case "turn/interrupt": + interrupted++; + reply(id, {}); + finish(params.turnId); + break; + case "thread/read": + reply(id, { + thread: { + ...thread, + turns: [ + { + ...turn("inspection", "completed"), + items: [ + { + type: "agentMessage", + id: "inspection-message", + memoryCitation: null, + text: JSON.stringify({ + wakes, + cleanCount, + interrupted, + monitorExecutions, + terminatedMonitors, + }), + }, + ], + }, + ], + }, + }); + break; + default: + write({ id, error: { code: -32601, message: method } }); + } +}); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index ce39ee64f51a..789698a48788 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -56,6 +56,7 @@ import { ProviderInstanceRegistryHydrationLive } from "./provider/Layers/Provide import * as TerminalManager from "./terminal/Manager.ts"; import * as McpHttpServer from "./mcp/McpHttpServer.ts"; import * as McpSessionRegistry from "./mcp/McpSessionRegistry.ts"; +import * as MonitorSession from "./mcp/MonitorSession.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; import * as PreviewManager from "./preview/Manager.ts"; import * as PortScanner from "./preview/PortScanner.ts"; @@ -478,6 +479,7 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( Layer.provideMerge(ProviderInstanceRegistryHydrationLive), ).pipe( Layer.provideMerge(AntigravityInstallation.layer), + Layer.provideMerge(MonitorSession.layer), // Shared native/canonical NDJSON writers used by both the per-instance // drivers (native stream, written from inside each `Adapter`) and // `ProviderService` (canonical stream, written after event normalization). diff --git a/docs/user/providers-codex.md b/docs/user/providers-codex.md index 417287cc0032..a48fdf439e91 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -69,3 +69,16 @@ In an existing Codex thread, send `/feedback` with an optional description, for example `/feedback The agent stopped before finishing the tests`. This uploads the conversation and Codex logs to OpenAI. The returned thread ID can be shared with OpenAI support. + +## Monitor background changes + +With Codex 0.153.2 or later, ask Codex to watch a CI job, log, or other changing +source and react when something happens. It can leave a watcher running after +its response finishes using the built-in monitoring +tools. No special prompt syntax is needed. On web and desktop, the thread shows +**Monitoring** between turns and wakes when the watcher reports an event. +Ordinary background commands do not wake the agent just because they print output. + +Use **Stop** in the thread to cancel background work and pending monitor events. +Watchers belong to the current provider session; after stopping that session or +restarting the environment, ask Codex to start the watch again.