From 32355a1438065dcaaf29d355e945d621f4f4bdca Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 19:52:51 -0700 Subject: [PATCH 1/4] fix(server): isolate Codex identities and expose event diagnostics --- .../src/mcp/BridgeStatusService.test.ts | 3 + apps/server/src/mcp/BridgeStatusService.ts | 4 + apps/server/src/mcp/McpEventRelay.ts | 84 +++++++++++--- apps/server/src/mcp/McpEvents.test.ts | 109 +++++++++++++++++- apps/server/src/mcp/McpEvents.ts | 19 ++- apps/server/src/mcp/OrchestratorMcpService.ts | 3 + apps/server/src/mcp/ThreadLaunchMcpService.ts | 4 + .../Adapters/CodexAdapterV2.test.ts | 34 +++++- .../Adapters/CodexAdapterV2.ts | 22 +++- .../src/persistence/ForkMigrations.test.ts | 35 ++++++ apps/server/src/persistence/ForkMigrations.ts | 2 + .../ForkMigrations/014_McpRelayDiagnostics.ts | 20 ++++ .../src/provider/CodexMcpIsolation.test.ts | 55 +++++++++ apps/server/src/provider/CodexMcpIsolation.ts | 56 +++++++++ .../src/provider/codexLaunchArgs.test.ts | 22 ++++ apps/server/src/provider/codexLaunchArgs.ts | 10 +- packages/contracts/src/mcpEvents.ts | 10 ++ packages/contracts/src/orchestratorMcp.ts | 1 + 18 files changed, 463 insertions(+), 30 deletions(-) create mode 100644 apps/server/src/persistence/ForkMigrations/014_McpRelayDiagnostics.ts create mode 100644 apps/server/src/provider/CodexMcpIsolation.test.ts create mode 100644 apps/server/src/provider/CodexMcpIsolation.ts diff --git a/apps/server/src/mcp/BridgeStatusService.test.ts b/apps/server/src/mcp/BridgeStatusService.test.ts index 46e96a7229b7a..753b3d4671659 100644 --- a/apps/server/src/mcp/BridgeStatusService.test.ts +++ b/apps/server/src/mcp/BridgeStatusService.test.ts @@ -36,6 +36,9 @@ it.effect( relayFailures: 0, relayPending: 0, relayLastError: null, + relayAlerts: [], + relayBindings: [], + subscriptionDuplicates: [], lastSuccessfulEventDeliveryAt: null, deliveriesExhaustedRetries: 0, }), diff --git a/apps/server/src/mcp/BridgeStatusService.ts b/apps/server/src/mcp/BridgeStatusService.ts index 8a2498f9fe152..0e2a2c5989e05 100644 --- a/apps/server/src/mcp/BridgeStatusService.ts +++ b/apps/server/src/mcp/BridgeStatusService.ts @@ -3,6 +3,7 @@ import { EnvironmentConnectReachability, McpEventSubscriptionsResult, McpEventWatch, + McpEventSubscriptionDuplicates, OrchestratorMcpFailure, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; @@ -14,11 +15,14 @@ import * as PeerForwarding from "../peer/PeerForwarding.ts"; import * as Invocation from "./McpInvocationContext.ts"; import * as McpEvents from "./McpEvents.ts"; import * as Metrics from "./McpBridgeMetrics.ts"; +import { RelayDiagnostics } from "./McpEventRelay.ts"; export const BridgeStatusResult = Schema.Struct({ + ...RelayDiagnostics.fields, serviceStartedAt: Schema.String, tunnel: Schema.NullOr(EnvironmentConnectReachability), daysUntilSessionExpires: Schema.Number, subscriptions: McpEventSubscriptionsResult.fields.subscriptions, + subscriptionDuplicates: McpEventSubscriptionDuplicates, inboxThreadId: McpEventSubscriptionsResult.fields.inboxThreadId, watches: Schema.Array(McpEventWatch), relayFailures: Schema.Number, diff --git a/apps/server/src/mcp/McpEventRelay.ts b/apps/server/src/mcp/McpEventRelay.ts index e93fd3f53d198..4129436887faa 100644 --- a/apps/server/src/mcp/McpEventRelay.ts +++ b/apps/server/src/mcp/McpEventRelay.ts @@ -74,6 +74,18 @@ export interface PreparedRelay { readonly events: ReadonlyArray; } +export const RelayDiagnostics = Schema.Struct({ + relayAlerts: Schema.Array(Schema.Struct({ reason: Schema.String, day: Schema.Number })), + relayBindings: Schema.Array( + Schema.Struct({ + threadId: ThreadId, + runId: Schema.String, + expiresAt: Schema.NullOr(Schema.String), + closedAt: Schema.NullOr(Schema.String), + }), + ), +}); + export class McpEventRelay extends Context.Service< McpEventRelay, { @@ -82,6 +94,7 @@ export class McpEventRelay extends Context.Service< threadId: ThreadId | null, ) => Effect.Effect; readonly inbox: (principal: string) => Effect.Effect; + readonly missingInbox: (principal: string) => Effect.Effect; readonly prepare: ( scope: Invocation.McpInvocationScope, input: McpThreadNotify, @@ -104,10 +117,12 @@ export class McpEventRelay extends Context.Service< threadId: ThreadId, reason: string, ) => Effect.Effect; - readonly status: ( - principal: string, - ) => Effect.Effect< - { relayFailures: number; relayPending: number; relayLastError: string | null }, + readonly status: (principal: string) => Effect.Effect< + { + relayFailures: number; + relayPending: number; + relayLastError: string | null; + } & typeof RelayDiagnostics.Type, McpEventsError >; } @@ -203,8 +218,9 @@ const make = Effect.gen(function* () { yield* sql .withTransaction( Effect.gen(function* () { - const added = yield* sql`INSERT INTO mcp_event_relay_alerts (principal, source_key, day) - VALUES (${principal}, ${sourceKey}, ${day}) ON CONFLICT DO NOTHING RETURNING source_key`; + const added = + yield* sql`INSERT INTO mcp_event_relay_alerts (principal, source_key, day, reason) + VALUES (${principal}, ${sourceKey}, ${day}, ${reason}) ON CONFLICT DO NOTHING RETURNING source_key`; if (added.length === 0) return; yield* enqueue({ ...pin, @@ -225,7 +241,8 @@ const make = Effect.gen(function* () { yield* sql`SELECT id, principal, thread_id AS "threadId", inbox_thread_id AS "inboxThreadId", run_id AS "runId", events, ledger_key AS "ledgerKey", scan_cursor AS "scanCursor", runtime_mode AS "runtimeMode", interaction_mode AS "interactionMode" - FROM mcp_event_relay_bindings WHERE expires_at > ${yield* Clock.currentTimeMillis}`.pipe( + FROM mcp_event_relay_bindings WHERE closed_at IS NULL + AND (expires_at IS NULL OR expires_at > ${yield* Clock.currentTimeMillis})`.pipe( Effect.flatMap(Schema.decodeUnknownEffect(Schema.Array(Binding))), Effect.mapError(storage), ); @@ -238,7 +255,7 @@ const make = Effect.gen(function* () { } const allowed = yield* authorize(binding.principal, binding.threadId).pipe(Effect.result); if (allowed._tag === "Failure") { - yield* sql`UPDATE mcp_event_relay_bindings SET expires_at = 0, last_error = ${allowed.failure.reason} WHERE id = ${binding.id}`.pipe( + yield* sql`UPDATE mcp_event_relay_bindings SET closed_at = ${yield* Clock.currentTimeMillis}, last_error = ${allowed.failure.reason} WHERE id = ${binding.id}`.pipe( Effect.mapError(storage), ); yield* alert(binding.principal, binding.id, binding.threadId, allowed.failure.reason); @@ -323,7 +340,7 @@ const make = Effect.gen(function* () { break; } } - yield* sql`UPDATE mcp_event_relay_bindings SET scan_cursor = ${through}, expires_at = CASE WHEN ${finished ? 1 : 0} = 1 THEN 0 ELSE expires_at END WHERE id = ${binding.id}`.pipe( + yield* sql`UPDATE mcp_event_relay_bindings SET scan_cursor = ${through}, closed_at = CASE WHEN ${finished ? 1 : 0} = 1 THEN ${yield* Clock.currentTimeMillis} ELSE closed_at END WHERE id = ${binding.id}`.pipe( Effect.mapError(storage), ); behind ||= !finished && through < latest; @@ -389,6 +406,14 @@ const make = Effect.gen(function* () { }, (effect) => pinGate.withPermit(effect), ), + missingInbox: Effect.fn("McpEventRelay.missingInbox")(function* (principal) { + const pins = yield* sql<{ signIn: string }>`SELECT substr(principal, 1, 8) AS "signIn" + FROM mcp_event_inboxes ORDER BY principal`.pipe(Effect.mapError(storage)); + yield* sql`INSERT INTO mcp_event_relay_alerts (principal, source_key, day, reason) + VALUES (${principal}, 'no_pinned_inbox', ${Math.floor((yield* Clock.currentTimeMillis) / 86400_000)}, 'no_pinned_inbox') + ON CONFLICT DO NOTHING`.pipe(Effect.mapError(storage)); + return `WARNING: Notifications are NOT bound. Caller sign-in ${principal.slice(0, 8)} has no pinned inbox. Sign-ins with pins: ${pins.map((pin) => pin.signIn).join(", ") || "none"}. Pin an inbox using the caller's own sign-in; pins are never shared across sign-ins.`; + }), prepare: Effect.fn("McpEventRelay.prepare")(function* (scope, input) { if ( scope.client === undefined || @@ -491,6 +516,11 @@ const make = Effect.gen(function* () { return { replayed }; }), drain: Effect.fn("McpEventRelay.drain")(function* () { + yield* sql`DELETE FROM mcp_event_relay_bindings + WHERE closed_at <= ${(yield* Clock.currentTimeMillis) - DELIVERED_RETENTION_MS} + OR expires_at <= ${(yield* Clock.currentTimeMillis) - DELIVERED_RETENTION_MS}`.pipe( + Effect.mapError(storage), + ); yield* sql`DELETE FROM mcp_event_relay_outbox WHERE status = 'delivered' AND next_attempt_at < ${(yield* Clock.currentTimeMillis) - DELIVERED_RETENTION_MS}`.pipe( Effect.mapError(storage), @@ -564,16 +594,38 @@ const make = Effect.gen(function* () { Effect.mapError(storage), ), alert, - status: Effect.fn("McpEventRelay.status")((principal) => - sql<{ relayFailures: number; relayPending: number; relayLastError: string | null }>` + status: Effect.fn("McpEventRelay.status")(function* (principal) { + const rows = yield* sql<{ + relayFailures: number; + relayPending: number; + relayLastError: string | null; + }>` SELECT (SELECT COALESCE(SUM(attempts), 0) FROM mcp_event_relay_outbox WHERE principal = ${principal}) + (SELECT COUNT(*) FROM mcp_event_relay_bindings WHERE principal = ${principal} AND last_error IS NOT NULL) AS "relayFailures", (SELECT COUNT(*) FROM mcp_event_relay_outbox WHERE principal = ${principal} AND status = 'pending') AS "relayPending", COALESCE((SELECT last_error FROM mcp_event_relay_outbox WHERE principal = ${principal} AND last_error IS NOT NULL ORDER BY rowid DESC LIMIT 1), - (SELECT last_error FROM mcp_event_relay_bindings WHERE principal = ${principal} AND last_error IS NOT NULL ORDER BY rowid DESC LIMIT 1)) AS "relayLastError"`.pipe( - Effect.map((rows) => rows[0]!), - Effect.mapError(storage), - ), - ), + (SELECT last_error FROM mcp_event_relay_bindings WHERE principal = ${principal} AND last_error IS NOT NULL ORDER BY rowid DESC LIMIT 1)) AS "relayLastError"`; + const relayAlerts = yield* sql<{ reason: string; day: number }>`SELECT reason, day + FROM mcp_event_relay_alerts WHERE principal = ${principal} ORDER BY day DESC, source_key`; + const bindings = yield* sql<{ + threadId: ThreadId; + runId: string; + expiresAt: number | null; + closedAt: number | null; + }>` + SELECT thread_id AS "threadId", run_id AS "runId", expires_at AS "expiresAt", closed_at AS "closedAt" + FROM mcp_event_relay_bindings WHERE principal = ${principal} ORDER BY rowid`; + const timestamp = (value: number | null) => + value === null ? null : DateTime.formatIso(DateTime.makeUnsafe(value)); + return { + ...rows[0]!, + relayAlerts, + relayBindings: bindings.map((binding) => ({ + ...binding, + expiresAt: timestamp(binding.expiresAt), + closedAt: timestamp(binding.closedAt), + })), + }; + }, Effect.mapError(storage)), }); }); export const layer = Layer.effect(McpEventRelay, make); diff --git a/apps/server/src/mcp/McpEvents.test.ts b/apps/server/src/mcp/McpEvents.test.ts index ff05ff9ab1784..885769ca69aaf 100644 --- a/apps/server/src/mcp/McpEvents.test.ts +++ b/apps/server/src/mcp/McpEvents.test.ts @@ -154,8 +154,8 @@ const fixture = Effect.gen(function* () { Effect.map((built) => Context.get(built, McpEvents.McpEvents)), ); const service = yield* build(); - const principal = AuthSessionId.make("principal-1"); - const other = AuthSessionId.make("principal-2"); + const principal = AuthSessionId.make("caller01-session"); + const other = AuthSessionId.make("caller02-session"); for (const sessionId of [principal, other]) yield* sessions.create({ sessionId, @@ -289,6 +289,111 @@ const fixture = Effect.gen(function* () { }; }); +it.effect("warns about the caller's missing pin without sharing another sign-in's inbox", () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture; + yield* f.grantWrite(f.principal); + yield* f.grantWrite(f.other); + yield* f.call(f.service.setInbox({ threadId: ThreadId.make("other-inbox") }), f.other); + const result = yield* f.call(f.service.prepareNotify({})); + expect(result).toMatchObject({ + status: "unavailable", + reason: "no_pinned_inbox", + warning: expect.stringContaining(f.principal.slice(0, 8)), + }); + expect("warning" in result && result.warning).toContain(f.other.slice(0, 8)); + const status = yield* f.call(f.service.bridgeStatus()); + expect(status.inboxThreadId).toBeNull(); + expect(status.relayAlerts).toContainEqual( + expect.objectContaining({ reason: "no_pinned_inbox" }), + ); + const sql = Context.get(f.context, SqlClient.SqlClient); + expect(yield* sql`SELECT id FROM mcp_event_relay_outbox`).toEqual([]); + }), + ).pipe(Effect.provide(NodeCrypto.layer)), +); + +it.effect( + "reuses canonical subscription identities and reports overlaps without deleting callbacks", + () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture; + const input = inputFor(ThreadId.make("duplicate-watch")); + const [first, second] = yield* Effect.all( + [ + f.call(f.service.subscribe(input)), + f.call( + f.service.subscribe({ + ...input, + delivery: { ...input.delivery, url: "https://CALLBACKS.example.com:443/events" }, + }), + ), + ], + { concurrency: "unbounded" }, + ); + expect(second.id).toBe(first.id); + expect((yield* f.call(f.service.listSubscriptions())).subscriptions).toHaveLength(1); + yield* f.call( + f.service.subscribe({ + ...input, + delivery: { ...input.delivery, url: "https://callbacks.example.com/another" }, + }), + ); + const status = yield* f.call(f.service.bridgeStatus()); + expect(status.subscriptionDuplicates).toEqual([ + expect.objectContaining({ + threadId: input.arguments.threadId, + name: input.name, + subscriptionIds: expect.arrayContaining([first.id]), + callbackGroups: expect.any(Array), + }), + ]); + expect(status.subscriptions).toHaveLength(2); + expect(encode(status.subscriptionDuplicates)).not.toContain(input.delivery.url); + }), + ).pipe(Effect.provide(NodeCrypto.layer)), +); + +it.effect( + "treats null relay expiry as none and collects only stale closed or expired bindings", + () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture; + yield* f.grantWrite(f.principal); + yield* f.call(f.service.setInbox({ threadId: ThreadId.make("expiry-inbox") })); + const prepared = yield* f.call( + f.service.prepareNotify({ events: ["thread.resource-released"] }), + ); + if ("status" in prepared) return yield* Effect.die("expected relay"); + for (const run of ["unbounded", "expired", "closed", "recent"]) { + yield* f.call( + f.service.notifyThread(prepared, ThreadId.make("expiry-source"), RunId.make(run)), + ); + } + const sql = Context.get(f.context, SqlClient.SqlClient); + yield* sql`UPDATE mcp_event_relay_bindings SET expires_at = NULL WHERE run_id = 'unbounded'`; + yield* sql`UPDATE mcp_event_relay_bindings SET expires_at = 0 WHERE run_id = 'expired'`; + yield* sql`UPDATE mcp_event_relay_bindings SET closed_at = 0 WHERE run_id = 'closed'`; + yield* TestClock.adjust("2 days"); + yield* sql`UPDATE mcp_event_relay_bindings SET closed_at = ${yield* Clock.currentTimeMillis} WHERE run_id = 'recent'`; + yield* f.append([ThreadId.make("expiry-source")]); + yield* f.service.drain(); + expect(f.inboxSends).toHaveLength(1); + const status = yield* f.call(f.service.bridgeStatus()); + expect(status.relayBindings).toEqual( + expect.arrayContaining([ + expect.objectContaining({ runId: "unbounded", expiresAt: null, closedAt: null }), + expect.objectContaining({ runId: "recent", closedAt: expect.any(String) }), + ]), + ); + expect(status.relayBindings).toHaveLength(2); + }), + ).pipe(Effect.provide(NodeCrypto.layer)), +); + it.effect( "binds notify to a pinned inbox without creating subscriptions and survives rebuild", () => diff --git a/apps/server/src/mcp/McpEvents.ts b/apps/server/src/mcp/McpEvents.ts index 94955f867ac86..004d68798c2e1 100644 --- a/apps/server/src/mcp/McpEvents.ts +++ b/apps/server/src/mcp/McpEvents.ts @@ -14,6 +14,7 @@ import { McpEventArguments, McpEventData, McpEventWatch, + McpEventSubscriptionDuplicates, McpEventsError, type McpEventSubscribe, type McpEventInboxInput, @@ -163,11 +164,12 @@ export class McpEvents extends Context.Service< clientRequestId?: string, ) => Effect.Effect; readonly bridgeStatus: () => Effect.Effect< - { + typeof Relay.RelayDiagnostics.Type & { readonly serviceStartedAt: string; readonly tunnel: EnvironmentConnectReachability | null; readonly daysUntilSessionExpires: number; readonly subscriptions: McpEventSubscriptionsResult["subscriptions"]; + readonly subscriptionDuplicates: typeof McpEventSubscriptionDuplicates.Type; readonly inboxThreadId: ThreadId | null; readonly relayFailures: number; readonly relayPending: number; @@ -640,6 +642,7 @@ const make = Effect.gen(function* () { status: "unavailable" as const, subscriptionIds: [], reason: "no_pinned_inbox", + warning: yield* relay.missingInbox(scope.client.sessionId), } ); }), @@ -666,7 +669,18 @@ const make = Effect.gen(function* () { const listing = yield* listSubscriptions(); const listed = listing.subscriptions; const watches = new Map(); + const overlaps = new Map(); for (const sub of listed.filter((sub) => sub.status === "active")) { + const key = yield* encodeJson([sub.arguments.threadId, sub.name]).pipe( + Effect.mapError(storage), + ); + const overlap = overlaps.get(key); + overlaps.set(key, { + threadId: sub.arguments.threadId, + name: sub.name, + subscriptionIds: [...(overlap?.subscriptionIds ?? []), sub.id], + callbackGroups: [...new Set([...(overlap?.callbackGroups ?? []), sub.callbackGroup])], + }); const watch = watches.get(sub.callbackGroup); watches.set(sub.callbackGroup, { callbackGroup: sub.callbackGroup, @@ -687,6 +701,9 @@ const make = Effect.gen(function* () { ), inboxThreadId: listing.inboxThreadId, subscriptions: listed, + subscriptionDuplicates: [...overlaps.values()].filter( + (group) => group.subscriptionIds.length > 1, + ), watches: [...watches.values()], ...(yield* relay.status(principal.sessionId)), lastSuccessfulEventDeliveryAt: diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 59442dc012c60..16fa3dbbce787 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -3475,6 +3475,9 @@ const make = Effect.gen(function* () { ).pipe(Effect.provideService(Invocation.McpInvocationContext, scope)); return { ...(notify === undefined ? {} : { notify }), + ...(notify !== undefined && "warning" in notify && notify.warning !== undefined + ? { warning: notify.warning } + : {}), threadId: input.threadId, link: threadLinkFor(scope, input.threadId, target.thread.title), messageId, diff --git a/apps/server/src/mcp/ThreadLaunchMcpService.ts b/apps/server/src/mcp/ThreadLaunchMcpService.ts index 146d4879409f8..a12a1fa7a2fb5 100644 --- a/apps/server/src/mcp/ThreadLaunchMcpService.ts +++ b/apps/server/src/mcp/ThreadLaunchMcpService.ts @@ -84,6 +84,7 @@ export const launchInput = Schema.Struct({ }); export const launchResult = Schema.Struct({ notify: Schema.optional(McpThreadNotifyResult), + warning: Schema.optional(Schema.String), threadId: ThreadId, link: Schema.optional(Schema.String), projectId: ProjectId, @@ -378,6 +379,9 @@ const launch = Effect.fn("ThreadLaunchMcpService.launch")( ); return { ...(notify === undefined ? {} : { notify }), + ...(notify !== undefined && "warning" in notify && notify.warning !== undefined + ? { warning: notify.warning } + : {}), threadId: thread.id, link: McpInvocationContext.threadLinkFor(scope, thread.id, thread.title), projectId: thread.projectId, diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index ba98605722e5a..b86bdab8aa44b 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -954,6 +954,7 @@ describe("CodexAdapterV2 process exit", () => { it.effect("counts the app-server as a running agent until its session closes", () => { let runningAgents = 0; + const commands: Array> = []; const factoryLayer = CodexAdapterV2.layerAppServerClientFactory.pipe( Layer.provide( Layer.mock(PortScanner.PortDiscovery)({ @@ -974,23 +975,32 @@ describe("CodexAdapterV2 process exit", () => { Layer.provide( Layer.succeed( ChildProcessSpawner.ChildProcessSpawner, - ChildProcessSpawner.make(() => - Effect.succeed( + ChildProcessSpawner.make((command) => { + const args = ChildProcess.isStandardCommand(command) ? command.args : []; + commands.push(args); + const listing = args.includes("mcp"); + return Effect.succeed( ChildProcessSpawner.makeHandle({ pid: ChildProcessSpawner.ProcessId(999_999_999), - exitCode: Effect.never, + exitCode: listing ? Effect.succeed(ChildProcessSpawner.ExitCode(0)) : Effect.never, isRunning: Effect.succeed(true), kill: () => Effect.void, unref: Effect.succeed(Effect.void), stdin: Sink.drain, - stdout: Stream.never, + stdout: listing + ? Stream.succeed( + new TextEncoder().encode( + '[{"name":"t3_omarchy"},{"name":"t3_macbook"},{"name":"other"}]', + ), + ) + : Stream.never, stderr: Stream.empty, all: Stream.empty, getInputFd: () => Sink.drain, getOutputFd: () => Stream.empty, }), - ), - ), + ); + }), ), ), ); @@ -1009,6 +1019,18 @@ describe("CodexAdapterV2 process exit", () => { .pipe(Effect.tap(() => Effect.sync(() => assert.equal(runningAgents, 1)))), ); assert.equal(runningAgents, 0); + assert.deepEqual(commands, [ + ["mcp", "list", "--json"], + [ + "app-server", + "-c", + "thread_unload_delay_secs=0", + "-c", + "mcp_servers.t3_omarchy.enabled=false", + "-c", + "mcp_servers.t3_macbook.enabled=false", + ], + ]); }).pipe(Effect.provide(factoryLayer)); }); }); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index b163e015335ce..4d90cfc316177 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -124,7 +124,8 @@ import { shouldPersistProviderEvent, } from "../../provider/EventNdjsonLogger.ts"; import * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers"; -import { codexAppServerArgs, resolveCodexLaunchArgs } from "../../provider/codexLaunchArgs.ts"; +import { resolveCodexLaunchArgs } from "../../provider/codexLaunchArgs.ts"; +import { isolatedCodexAppServerArgs } from "../../provider/CodexMcpIsolation.ts"; import { mergeProviderInstanceEnvironment } from "@t3tools/provider-core/server/instanceEnvironment"; import * as McpProviderSession from "@t3tools/provider-core/server/mcpSession"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; @@ -1643,11 +1644,24 @@ export const layerAppServerClientFactory: Layer.Layer< ...codexAppServerEnvironment(input.environment), ...(input.settings.homePath ? { CODEX_HOME: input.settings.homePath } : {}), }; + const args = yield* isolatedCodexAppServerArgs({ + binary: input.settings.binaryPath || "codex", + env: environment, + launchArgs: resolveCodexLaunchArgs(input.settings.launchArgs, input.environment), + }).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterOpenSessionError({ + driver: CODEX_PROVIDER, + providerSessionId: input.providerSessionId, + cause, + }), + ), + ); const command = yield* makeCodexAppServerSpawnCommand({ command: input.settings.binaryPath || "codex", - args: codexAppServerArgs( - resolveCodexLaunchArgs(input.settings.launchArgs, input.environment), - ), + args, env: environment, threadId: input.threadId, }).pipe(Effect.provideService(AgentScope, agentScope)); diff --git a/apps/server/src/persistence/ForkMigrations.test.ts b/apps/server/src/persistence/ForkMigrations.test.ts index c8cc3b39900b6..722c171c446f4 100644 --- a/apps/server/src/persistence/ForkMigrations.test.ts +++ b/apps/server/src/persistence/ForkMigrations.test.ts @@ -11,6 +11,8 @@ import { FORK_MIGRATIONS_TABLE, forkMigrationManifest, legacyForkRows } from "./ import McpEvents from "./ForkMigrations/001_McpEvents.ts"; import McpBridgeStatus from "./ForkMigrations/004_McpBridgeStatus.ts"; import McpSubscriptionRenewal from "./ForkMigrations/006_McpSubscriptionRenewal.ts"; +import McpInboxRelay from "./ForkMigrations/009_McpInboxRelay.ts"; +import McpRelayDiagnostics from "./ForkMigrations/014_McpRelayDiagnostics.ts"; import { hasPendingMigrations, migrationManifest, runMigrations } from "./Migrations.ts"; // The last upstream id before fork builds started taking upstream's ids for fork migrations. @@ -49,6 +51,39 @@ const seedLegacyForkDatabase = Effect.gen(function* () { const pendingUpstream = migrationManifest.filter(([id]) => id > LAST_SHARED_ID); describe("fork migrations", () => { + it.effect("makes relay expiry nullable while preserving terminal bindings and cursors", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: migrationManifest.at(-1)![0] }); + yield* McpEvents; + yield* McpInboxRelay; + yield* sql`INSERT INTO auth_sessions (session_id, subject, scopes, method, issued_at, expires_at) + VALUES ('relay', 'mcp-client', '[]', 'bearer-access-token', '2026-11-12', '2026-12-12')`; + yield* sql`INSERT INTO mcp_event_relay_bindings + (id, principal, thread_id, inbox_thread_id, run_id, events, runtime_mode, interaction_mode, scan_cursor, expires_at, last_error) + VALUES ('closed', 'relay', 'source', 'inbox', 'run', '[]', 'full-access', 'default', 5, 0, 'denied'), + ('finite', 'relay', 'source', 'inbox', 'run2', '[]', 'full-access', 'default', 7, 123456, NULL)`; + yield* McpRelayDiagnostics; + const rows = yield* sql`SELECT id, scan_cursor, expires_at, closed_at, last_error + FROM mcp_event_relay_bindings ORDER BY id`; + assert.deepEqual(rows, [ + { + id: "closed", + scan_cursor: 5, + expires_at: null, + closed_at: rows[0]!.closed_at, + last_error: "denied", + }, + { id: "finite", scan_cursor: 7, expires_at: 123456, closed_at: null, last_error: null }, + ]); + assert.isNumber(rows[0]!.closed_at); + yield* sql`UPDATE mcp_event_relay_bindings SET expires_at = NULL WHERE id = 'finite'`; + assert.deepEqual( + yield* sql`SELECT id FROM mcp_event_relay_bindings WHERE expires_at IS NULL AND closed_at IS NULL`, + [{ id: "finite" }], + ); + }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), + ); it("registers unique fork ids in strictly increasing order", () => { const ids = forkMigrationManifest.map(([id]) => id); assert.deepEqual( diff --git a/apps/server/src/persistence/ForkMigrations.ts b/apps/server/src/persistence/ForkMigrations.ts index 271f0fb4b53b8..752300b11997e 100644 --- a/apps/server/src/persistence/ForkMigrations.ts +++ b/apps/server/src/persistence/ForkMigrations.ts @@ -15,6 +15,7 @@ import ForkMigration0010 from "./ForkMigrations/010_WorkLedgerPortfolio.ts"; import ForkMigration0011 from "./ForkMigrations/011_ForkRunExclusions.ts"; import ForkMigration0012 from "./ForkMigrations/012_WorkLedgerApproval.ts"; import ForkMigration0013 from "./ForkMigrations/013_WorkLedgerHygiene.ts"; +import ForkMigration0014 from "./ForkMigrations/014_McpRelayDiagnostics.ts"; // Fork-only migrations, recorded in their own table so upstream keeps every id in // effect_sql_migrations. They run after all upstream migrations. Add new ones here, never in // Migrations.ts, and never rename one: the repair below finds old rows by name. See FORK.md. @@ -32,6 +33,7 @@ const forkMigrationEntries = [ [11, "ForkRunExclusions", ForkMigration0011], [12, "WorkLedgerApproval", ForkMigration0012], [13, "WorkLedgerHygiene", ForkMigration0013], + [14, "McpRelayDiagnostics", ForkMigration0014], ] as const; export const FORK_MIGRATIONS_TABLE = "t3_fork_migrations"; diff --git a/apps/server/src/persistence/ForkMigrations/014_McpRelayDiagnostics.ts b/apps/server/src/persistence/ForkMigrations/014_McpRelayDiagnostics.ts new file mode 100644 index 0000000000000..a6e4b83211564 --- /dev/null +++ b/apps/server/src/persistence/ForkMigrations/014_McpRelayDiagnostics.ts @@ -0,0 +1,20 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`ALTER TABLE mcp_event_relay_bindings RENAME TO mcp_event_relay_bindings_legacy`; + yield* sql`CREATE TABLE mcp_event_relay_bindings ( + id TEXT PRIMARY KEY, principal TEXT NOT NULL REFERENCES auth_sessions(session_id) ON DELETE CASCADE, + thread_id TEXT NOT NULL, inbox_thread_id TEXT NOT NULL, run_id TEXT NOT NULL, + events TEXT NOT NULL, ledger_key TEXT, runtime_mode TEXT NOT NULL, interaction_mode TEXT NOT NULL, + scan_cursor INTEGER NOT NULL, expires_at INTEGER, last_error TEXT, closed_at INTEGER + )`; + yield* sql`INSERT INTO mcp_event_relay_bindings + SELECT id, principal, thread_id, inbox_thread_id, run_id, events, ledger_key, runtime_mode, + interaction_mode, scan_cursor, NULLIF(expires_at, 0), last_error, + CASE WHEN expires_at = 0 THEN CAST(strftime('%s', 'now') AS INTEGER) * 1000 ELSE NULL END + FROM mcp_event_relay_bindings_legacy`; + yield* sql`DROP TABLE mcp_event_relay_bindings_legacy`; + yield* sql`ALTER TABLE mcp_event_relay_alerts ADD COLUMN reason TEXT NOT NULL DEFAULT 'delivery_failed'`; +}); diff --git a/apps/server/src/provider/CodexMcpIsolation.test.ts b/apps/server/src/provider/CodexMcpIsolation.test.ts new file mode 100644 index 0000000000000..5da17c73e1438 --- /dev/null +++ b/apps/server/src/provider/CodexMcpIsolation.test.ts @@ -0,0 +1,55 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { writeFakeCli } from "@t3tools/provider-testing/fakeCli"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as FileSystem from "effect/FileSystem"; + +import { isolatedCodexAppServerArgs } from "./CodexMcpIsolation.ts"; + +it.live("isolates configured T3 sign-ins without introducing absent transports", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const directory = yield* fs.makeTempDirectoryScoped(); + const binary = writeFakeCli({ + directory, + name: "codex", + source: ` + import assert from "node:assert/strict"; + assert.deepEqual(process.argv.slice(2), ["-c", "mcp_servers.t3_macbook.enabled=true", "mcp", "list", "--json"]); + console.log(process.env.SERVERS); + `, + }); + for (const servers of [[], [{ name: "other" }], [{ name: "t3_macbook" }, { name: "other" }]]) { + const args = yield* isolatedCodexAppServerArgs({ + binary, + env: { ...process.env, SERVERS: JSON.stringify(servers) }, + launchArgs: "-c mcp_servers.t3_macbook.enabled=true", + }); + assert.deepEqual(args, [ + "app-server", + "-c", + "thread_unload_delay_secs=0", + "-c", + "mcp_servers.t3_macbook.enabled=true", + ...(servers.some((server) => server.name === "t3_macbook") + ? ["-c", "mcp_servers.t3_macbook.enabled=false"] + : []), + ]); + } + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.live("refuses to launch when inherited MCP configuration cannot be inspected", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const directory = yield* fs.makeTempDirectoryScoped(); + for (const source of ["process.exit(1)", 'console.log("not JSON")']) { + const binary = writeFakeCli({ directory, name: "codex", source }); + const result = yield* Effect.exit( + isolatedCodexAppServerArgs({ binary, env: process.env, launchArgs: "" }), + ); + assert.isTrue(Exit.isFailure(result)); + } + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/provider/CodexMcpIsolation.ts b/apps/server/src/provider/CodexMcpIsolation.ts new file mode 100644 index 0000000000000..b5c86a827fd3f --- /dev/null +++ b/apps/server/src/provider/CodexMcpIsolation.ts @@ -0,0 +1,56 @@ +import { resolveSpawnCommand } from "@t3tools/shared/shell"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import { ChildProcess, ChildProcessSpawner } from "effect/process"; + +import { codexAppServerArgs, codexMcpListLaunchArgs } from "./codexLaunchArgs.ts"; + +const decodeServers = Schema.decodeUnknownEffect( + Schema.fromJsonString(Schema.Array(Schema.Struct({ name: Schema.String }))), +); + +class CodexMcpIsolationError extends Schema.TaggedError()( + "CodexMcpIsolationError", + { exitCode: Schema.Number }, +) {} + +export const isolatedCodexAppServerArgs = Effect.fn("CodexMcpIsolation.appServerArgs")( + function* (input: { + readonly binary: string; + readonly env: NodeJS.ProcessEnv; + readonly launchArgs: string; + }) { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const command = yield* resolveSpawnCommand( + input.binary, + [...codexMcpListLaunchArgs(input.launchArgs), "mcp", "list", "--json"], + { env: input.env }, + ); + const child = yield* spawner.spawn( + ChildProcess.make(command.command, command.args, { + env: input.env, + shell: command.shell, + forceKillAfter: "1 second", + }), + ); + const [stdout, , exitCode] = yield* Effect.all( + [ + child.stdout.pipe(Stream.decodeText(), Stream.mkString), + child.stderr.pipe(Stream.runDrain), + child.exitCode, + ], + { concurrency: "unbounded" }, + ); + if (exitCode !== 0) { + return yield* new CodexMcpIsolationError({ exitCode: Number(exitCode) }); + } + const servers = yield* decodeServers(stdout); + return codexAppServerArgs( + input.launchArgs, + servers.map((server) => server.name), + ); + }, + Effect.scoped, + Effect.timeout("10 seconds"), +); diff --git a/apps/server/src/provider/codexLaunchArgs.test.ts b/apps/server/src/provider/codexLaunchArgs.test.ts index 8f4e1b7ca6d6a..cee477575221f 100644 --- a/apps/server/src/provider/codexLaunchArgs.test.ts +++ b/apps/server/src/provider/codexLaunchArgs.test.ts @@ -28,6 +28,28 @@ describe("codexAppServerArgs", () => { ...row.expected, ]); }); + it("disables inherited T3 sign-ins after user overrides without disabling other MCPs", () => { + const args = codexAppServerArgs( + "-c mcp_servers.t3_omarchy.enabled=true -c mcp_servers.other.enabled=true", + ["t3_omarchy", "t3_macbook", "other", "t3-code", "t3-code-direct"], + ); + NodeAssert.deepStrictEqual(args.slice(-4), [ + "-c", + "mcp_servers.t3_omarchy.enabled=false", + "-c", + "mcp_servers.t3_macbook.enabled=false", + ]); + NodeAssert.ok(args.includes("mcp_servers.other.enabled=true")); + }); + it("does not create a transport-less override for absent inherited entries", () => { + NodeAssert.deepStrictEqual(codexAppServerArgs(undefined, ["t3_macbook"]), [ + "app-server", + "-c", + "thread_unload_delay_secs=0", + "-c", + "mcp_servers.t3_macbook.enabled=false", + ]); + }); }); describe("codexExecLaunchArgs", () => { diff --git a/apps/server/src/provider/codexLaunchArgs.ts b/apps/server/src/provider/codexLaunchArgs.ts index 31ff37e31deb5..40cbc2547c267 100644 --- a/apps/server/src/provider/codexLaunchArgs.ts +++ b/apps/server/src/provider/codexLaunchArgs.ts @@ -10,11 +10,19 @@ export const resolveCodexLaunchArgs = ( const codexLaunchArgv = (launchArgs?: string): ReadonlyArray => tokenizeCliArgs(launchArgs); // T3 unsubscribes only when it unloads a thread; Codex's own unload delay would stack on that. -export const codexAppServerArgs = (launchArgs?: string) => [ +export const codexAppServerArgs = ( + launchArgs?: string, + inheritedMcpServers: ReadonlyArray = [], +) => [ "app-server", "-c", "thread_unload_delay_secs=0", ...codexLaunchArgv(launchArgs), + ...inheritedMcpServers.flatMap((name) => + name === "t3_omarchy" || name === "t3_macbook" + ? ["-c", `mcp_servers.${name}.enabled=false`] + : [], + ), ]; interface LaunchArgFilter { diff --git a/packages/contracts/src/mcpEvents.ts b/packages/contracts/src/mcpEvents.ts index 17a707122895a..4bae4dee70fba 100644 --- a/packages/contracts/src/mcpEvents.ts +++ b/packages/contracts/src/mcpEvents.ts @@ -70,6 +70,15 @@ export const McpEventWatch = Schema.Struct({ soonestExpiry: Schema.String, }); +export const McpEventSubscriptionDuplicates = Schema.Array( + Schema.Struct({ + threadId: ThreadId, + name: McpEventName, + subscriptionIds: Schema.Array(Schema.String), + callbackGroups: Schema.Array(Schema.String), + }), +); + export const McpEventSubscriptionResult = Schema.Struct({ id: Schema.String, refreshBefore: Schema.String, @@ -121,5 +130,6 @@ export const McpThreadNotifyResult = Schema.Struct({ relay: Schema.optional(Schema.String), subscriptionIds: Schema.Array(Schema.String), reason: Schema.optional(Schema.String), + warning: Schema.optional(Schema.String), }); export type McpThreadNotifyResult = typeof McpThreadNotifyResult.Type; diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 4a2706f09d4cf..816d2e8cfbe61 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -602,6 +602,7 @@ export type OrchestratorMcpThreadSendInput = typeof OrchestratorMcpThreadSendInp export const OrchestratorMcpThreadSendResult = Schema.Struct({ notify: Schema.optional(McpThreadNotifyResult), + warning: Schema.optional(Schema.String), threadId: ThreadId, link: Schema.optional(Schema.String), messageId: MessageId, From 81cab48bc1e0c3487c495e2c3431f12f2719d446 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 20:32:12 -0700 Subject: [PATCH 2/4] fix(server): inspect Codex config offline before thread startup --- apps/server/package.json | 1 + apps/server/src/mcp/McpEventRelay.ts | 28 ++- apps/server/src/mcp/McpEvents.test.ts | 175 +++++++++++++++++- .../Adapters/CodexAdapterV2.test.ts | 25 +-- .../Adapters/CodexAdapterV2.ts | 19 +- .../src/provider/CodexMcpIsolation.test.ts | 127 +++++++++---- apps/server/src/provider/CodexMcpIsolation.ts | 147 ++++++++++----- .../src/provider/codexLaunchArgs.test.ts | 4 +- apps/server/src/provider/codexLaunchArgs.ts | 4 +- pnpm-lock.yaml | 3 + 10 files changed, 411 insertions(+), 122 deletions(-) diff --git a/apps/server/package.json b/apps/server/package.json index d16b8d4e19f2f..8cb6d2aa73fb1 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -70,6 +70,7 @@ "playwright-core": "1.64.1", "proper-lockfile": "4.1.2", "saxes": "6.0.0", + "smol-toml": "^1.9.1", "stream-chain": "^4.2.5", "stream-json": "3.7.0", "transcribe-cpp": "0.4.1", diff --git a/apps/server/src/mcp/McpEventRelay.ts b/apps/server/src/mcp/McpEventRelay.ts index 4129436887faa..cf2b3dd8022b5 100644 --- a/apps/server/src/mcp/McpEventRelay.ts +++ b/apps/server/src/mcp/McpEventRelay.ts @@ -255,7 +255,8 @@ const make = Effect.gen(function* () { } const allowed = yield* authorize(binding.principal, binding.threadId).pipe(Effect.result); if (allowed._tag === "Failure") { - yield* sql`UPDATE mcp_event_relay_bindings SET closed_at = ${yield* Clock.currentTimeMillis}, last_error = ${allowed.failure.reason} WHERE id = ${binding.id}`.pipe( + const now = yield* Clock.currentTimeMillis; + yield* sql`UPDATE mcp_event_relay_bindings SET closed_at = ${now}, expires_at = ${now}, last_error = ${allowed.failure.reason} WHERE id = ${binding.id}`.pipe( Effect.mapError(storage), ); yield* alert(binding.principal, binding.id, binding.threadId, allowed.failure.reason); @@ -340,7 +341,8 @@ const make = Effect.gen(function* () { break; } } - yield* sql`UPDATE mcp_event_relay_bindings SET scan_cursor = ${through}, closed_at = CASE WHEN ${finished ? 1 : 0} = 1 THEN ${yield* Clock.currentTimeMillis} ELSE closed_at END WHERE id = ${binding.id}`.pipe( + const now = yield* Clock.currentTimeMillis; + yield* sql`UPDATE mcp_event_relay_bindings SET scan_cursor = ${through}, closed_at = CASE WHEN ${finished ? 1 : 0} = 1 THEN ${now} ELSE closed_at END, expires_at = CASE WHEN ${finished ? 1 : 0} = 1 THEN ${now} ELSE expires_at END WHERE id = ${binding.id}`.pipe( Effect.mapError(storage), ); behind ||= !finished && through < latest; @@ -407,12 +409,12 @@ const make = Effect.gen(function* () { (effect) => pinGate.withPermit(effect), ), missingInbox: Effect.fn("McpEventRelay.missingInbox")(function* (principal) { - const pins = yield* sql<{ signIn: string }>`SELECT substr(principal, 1, 8) AS "signIn" - FROM mcp_event_inboxes ORDER BY principal`.pipe(Effect.mapError(storage)); + const pins = yield* sql<{ count: number }>`SELECT count(*) AS count + FROM mcp_event_inboxes`.pipe(Effect.mapError(storage)); yield* sql`INSERT INTO mcp_event_relay_alerts (principal, source_key, day, reason) VALUES (${principal}, 'no_pinned_inbox', ${Math.floor((yield* Clock.currentTimeMillis) / 86400_000)}, 'no_pinned_inbox') ON CONFLICT DO NOTHING`.pipe(Effect.mapError(storage)); - return `WARNING: Notifications are NOT bound. Caller sign-in ${principal.slice(0, 8)} has no pinned inbox. Sign-ins with pins: ${pins.map((pin) => pin.signIn).join(", ") || "none"}. Pin an inbox using the caller's own sign-in; pins are never shared across sign-ins.`; + return `WARNING: Notifications are NOT bound. Caller sign-in ${principal.slice(0, 8)} has no pinned inbox. Pinned sign-ins: ${pins[0]?.count ?? 0}. Pin an inbox using the caller's own sign-in; pins are never shared across sign-ins.`; }), prepare: Effect.fn("McpEventRelay.prepare")(function* (scope, input) { if ( @@ -449,11 +451,21 @@ const make = Effect.gen(function* () { firstSequence: number | null; }>`SELECT MIN(sequence) AS "firstSequence" FROM orchestration_events WHERE application_event_version = 2 AND aggregate_kind = 'thread' AND stream_id = ${threadId} - AND event_type = 'run.updated' AND json_extract(payload_json, '$.id') = ${runId}`.pipe( + AND event_type IN ('run.created', 'run.updated') AND json_extract(payload_json, '$.id') = ${runId}`.pipe( Effect.mapError(storage), ); - if (run !== undefined && start?.firstSequence == null) - return yield* new McpEventsError({ code: -32602, reason: "source_history_pruned" }); + if (run !== undefined && start?.firstSequence == null) { + const [oldest] = yield* sql<{ occurredAt: string }>`SELECT occurred_at AS "occurredAt" + FROM orchestration_events + WHERE aggregate_kind = 'project' + OR (application_event_version = 2 AND aggregate_kind = 'thread') + ORDER BY sequence LIMIT 1`.pipe(Effect.mapError(storage)); + if ( + oldest !== undefined && + DateTime.isLessThan(run.requestedAt, DateTime.makeUnsafe(oldest.occurredAt)) + ) + return yield* new McpEventsError({ code: -32602, reason: "source_history_pruned" }); + } const cursor = Math.min( prepared.cursor, start?.firstSequence == null ? prepared.cursor : Math.max(0, start.firstSequence - 1), diff --git a/apps/server/src/mcp/McpEvents.test.ts b/apps/server/src/mcp/McpEvents.test.ts index 885769ca69aaf..ee507eeb5a796 100644 --- a/apps/server/src/mcp/McpEvents.test.ts +++ b/apps/server/src/mcp/McpEvents.test.ts @@ -26,6 +26,7 @@ import { ProviderInstanceId, ProviderDriverKind, type OrchestrationV2StoredEvent, + type OrchestrationV2Run, RuntimeRequestId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -302,7 +303,8 @@ it.effect("warns about the caller's missing pin without sharing another sign-in' reason: "no_pinned_inbox", warning: expect.stringContaining(f.principal.slice(0, 8)), }); - expect("warning" in result && result.warning).toContain(f.other.slice(0, 8)); + expect("warning" in result && result.warning).not.toContain(f.other.slice(0, 8)); + expect("warning" in result && result.warning).toContain("Pinned sign-ins: 1"); const status = yield* f.call(f.service.bridgeStatus()); expect(status.inboxThreadId).toBeNull(); expect(status.relayAlerts).toContainEqual( @@ -390,6 +392,13 @@ it.effect( ]), ); expect(status.relayBindings).toHaveLength(2); + yield* f.sessions.revoke({ sessionId: f.principal, revokedAt: DateTime.makeUnsafe(0) }); + yield* f.service.drain(); + expect( + yield* sql`SELECT expires_at, closed_at FROM mcp_event_relay_bindings WHERE run_id = 'unbounded'`, + ).toEqual([ + { expires_at: yield* Clock.currentTimeMillis, closed_at: yield* Clock.currentTimeMillis }, + ]); }), ).pipe(Effect.provide(NodeCrypto.layer)), ); @@ -487,6 +496,9 @@ it.effect( expect(f.inboxSends).toHaveLength(1); const sql = Context.get(f.context, SqlClient.SqlClient); expect(yield* sql`SELECT body FROM mcp_event_relay_outbox`).toEqual([{ body: "" }]); + expect(yield* sql`SELECT expires_at, closed_at FROM mcp_event_relay_bindings`).toEqual([ + { expires_at: yield* Clock.currentTimeMillis, closed_at: yield* Clock.currentTimeMillis }, + ]); expect(f.sent).toHaveLength(0); expect((yield* f.call(rebuilt.listSubscriptions())).subscriptions).toHaveLength(0); expect(f.inboxSends[0]).toMatchObject({ @@ -517,6 +529,167 @@ it.effect( ).pipe(Effect.provide(NodeCrypto.layer)), ); +it.effect.each(["idle", "busy"] as const)( + "binds notify for a send to a thread that is %s before its first run update", + (state) => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture; + yield* f.grantWrite(f.principal); + const threadId = ThreadId.make(`send-${state}`); + const requestedAt = DateTime.makeUnsafe(100); + const run: OrchestrationV2Run = { + id: RunId.make(`sent-${state}`), + threadId, + ordinal: state === "busy" ? 2 : 1, + providerInstanceId: ProviderInstanceId.make("replay"), + modelSelection: { instanceId: ProviderInstanceId.make("replay"), model: "fixture" }, + providerThreadId: null, + userMessageId: MessageId.make(`sent-message-${state}`), + rootNodeId: null, + activeAttemptId: null, + status: "queued", + requestedAt, + startedAt: null, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + const records = idleThreadProjection(liveThreadShell(threadId)); + const preceding = { + ...run, + id: RunId.make("preceding"), + ordinal: 1, + status: "running" as const, + }; + f.projections.set(threadId, { + ...records, + runs: state === "busy" ? [preceding, run] : [run], + }); + if (state === "busy") { + yield* f.store.appendAgentEvents({ + events: [ + { + id: EventId.make("preceding-started"), + threadId, + occurredAt: requestedAt, + type: "run.updated", + payload: preceding, + }, + ], + }); + } + yield* f.call(f.service.setInbox({ threadId: ThreadId.make("inbox") })); + const notification = yield* f.call( + ThreadNotifications.prepare({}).pipe( + Effect.provideService(McpEvents.McpEvents, f.service), + ), + ); + const [created] = yield* f.store.appendAgentEvents({ + events: [ + { + id: EventId.make("sent-created"), + threadId, + occurredAt: requestedAt, + type: "run.created", + payload: run, + }, + ], + }); + if (state === "busy") yield* f.outcome(threadId, preceding.id); + expect( + yield* f.call(ThreadNotifications.finish(notification, threadId, run.id)), + ).toMatchObject({ status: "bound" }); + const sql = Context.get(f.context, SqlClient.SqlClient); + const [binding] = yield* sql<{ + scan_cursor: number; + }>`SELECT scan_cursor FROM mcp_event_relay_bindings`; + expect(binding?.scan_cursor).toBe(created!.sequence - 1); + yield* f.service.drain(); + expect(f.inboxSends).toHaveLength(0); + yield* f.store.appendAgentEvents({ + events: [ + { + id: EventId.make("sent-started"), + threadId, + occurredAt: DateTime.makeUnsafe(200), + type: "run.updated", + payload: { ...run, status: "running", startedAt: DateTime.makeUnsafe(200) }, + }, + { + id: EventId.make("sent-completed"), + threadId, + occurredAt: DateTime.makeUnsafe(300), + type: "run.updated", + payload: { + ...run, + status: "completed", + startedAt: DateTime.makeUnsafe(200), + completedAt: DateTime.makeUnsafe(300), + }, + }, + ], + }); + const lifecycle = yield* sql<{ + event_type: string; + }>`SELECT event_type FROM orchestration_events + WHERE sequence > ${binding!.scan_cursor} AND json_extract(payload_json, '$.id') = ${run.id} ORDER BY sequence`; + expect(lifecycle.map((event) => event.event_type)).toEqual([ + "run.created", + "run.updated", + "run.updated", + ]); + yield* f.service.drain(); + expect(f.inboxSends).toHaveLength(1); + expect(f.inboxSends[0]?.text).toContain(`"runId":"${run.id}"`); + }), + ).pipe(Effect.provide(NodeCrypto.layer)), +); + +it.effect.each(["empty", "newer", "same", "pruned"] as const)( + "requires evidence older than retained history before refusing a missing run event (%s)", + (history) => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture; + yield* f.grantWrite(f.principal); + const threadId = ThreadId.make("missing-history"); + const runId = RunId.make("missing-run"); + const [stored] = yield* f.outcome(threadId, runId); + const records = idleThreadProjection(liveThreadShell(threadId)); + const event = stored!.event; + if (event.type !== "run.updated") return yield* Effect.die("expected run update"); + f.projections.set(threadId, { + ...records, + runs: [ + { + ...event.payload, + requestedAt: DateTime.makeUnsafe( + history === "newer" ? 2000 : history === "same" ? 1234 : 1, + ), + }, + ], + }); + const sql = Context.get(f.context, SqlClient.SqlClient); + yield* sql`DELETE FROM orchestration_events`; + if (history !== "empty") yield* f.append([ThreadId.make("retained")]); + yield* f.call(f.service.setInbox({ threadId: ThreadId.make("inbox") })); + const notification = yield* f.call( + ThreadNotifications.prepare({}).pipe( + Effect.provideService(McpEvents.McpEvents, f.service), + ), + ); + expect( + yield* f.call(ThreadNotifications.finish(notification, threadId, runId)), + ).toMatchObject( + history === "pruned" + ? { status: "failed", reason: "source_history_pruned" } + : { status: "bound" }, + ); + }), + ).pipe(Effect.provide(NodeCrypto.layer)), +); + it.effect("recovers the source result when an idempotent retry follows a lost binding", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index b86bdab8aa44b..16fce49baa71c 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -978,22 +978,15 @@ describe("CodexAdapterV2 process exit", () => { ChildProcessSpawner.make((command) => { const args = ChildProcess.isStandardCommand(command) ? command.args : []; commands.push(args); - const listing = args.includes("mcp"); return Effect.succeed( ChildProcessSpawner.makeHandle({ pid: ChildProcessSpawner.ProcessId(999_999_999), - exitCode: listing ? Effect.succeed(ChildProcessSpawner.ExitCode(0)) : Effect.never, + exitCode: Effect.never, isRunning: Effect.succeed(true), kill: () => Effect.void, unref: Effect.succeed(Effect.void), stdin: Sink.drain, - stdout: listing - ? Stream.succeed( - new TextEncoder().encode( - '[{"name":"t3_omarchy"},{"name":"t3_macbook"},{"name":"other"}]', - ), - ) - : Stream.never, + stdout: Stream.never, stderr: Stream.empty, all: Stream.empty, getInputFd: () => Sink.drain, @@ -1005,6 +998,12 @@ describe("CodexAdapterV2 process exit", () => { ), ); return Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const directory = yield* fs.makeTempDirectoryScoped(); + yield* fs.writeFileString( + `${directory}/config.toml`, + '[mcp_servers.t3_omarchy]\ncommand="unused"\n[mcp_servers.t3_macbook]\ncommand="unused"\n', + ); const factory = yield* CodexAdapterV2.CodexAppServerClientFactory; yield* Effect.scoped( factory @@ -1013,14 +1012,13 @@ describe("CodexAdapterV2 process exit", () => { threadId: ThreadId.make("thread-codex-agent"), providerSessionId: ProviderSessionId.make("session-codex-agent"), runtimePolicy: CODEX_TEST_RUNTIME_POLICY, - settings: DEFAULT_CODEX_SETTINGS, + settings: { ...DEFAULT_CODEX_SETTINGS, homePath: directory }, environment: {}, }) .pipe(Effect.tap(() => Effect.sync(() => assert.equal(runningAgents, 1)))), ); assert.equal(runningAgents, 0); assert.deepEqual(commands, [ - ["mcp", "list", "--json"], [ "app-server", "-c", @@ -1031,7 +1029,10 @@ describe("CodexAdapterV2 process exit", () => { "mcp_servers.t3_macbook.enabled=false", ], ]); - }).pipe(Effect.provide(factoryLayer)); + }).pipe( + Effect.scoped, + Effect.provide(factoryLayer.pipe(Layer.provideMerge(NodeServices.layer))), + ); }); }); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 4d90cfc316177..b91099f70b47f 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -125,7 +125,7 @@ import { } from "../../provider/EventNdjsonLogger.ts"; import * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers"; import { resolveCodexLaunchArgs } from "../../provider/codexLaunchArgs.ts"; -import { isolatedCodexAppServerArgs } from "../../provider/CodexMcpIsolation.ts"; +import { makeIsolatedCodexAppServerArgs } from "../../provider/CodexMcpIsolation.ts"; import { mergeProviderInstanceEnvironment } from "@t3tools/provider-core/server/instanceEnvironment"; import * as McpProviderSession from "@t3tools/provider-core/server/mcpSession"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; @@ -1629,6 +1629,8 @@ export const layerAppServerClientFactory: Layer.Layer< | ChildProcessSpawner.ChildProcessSpawner | PortScanner.PortDiscovery | ProviderEventLoggers.ProviderEventLoggers + | FileSystem.FileSystem + | Path.Path > = Layer.effect( CodexAppServerClientFactory, Effect.gen(function* () { @@ -1636,6 +1638,7 @@ export const layerAppServerClientFactory: Layer.Layer< const { native: nativeEventLogger } = yield* ProviderEventLoggers.ProviderEventLoggers; const agentScope = yield* AgentScope; const portDiscovery = yield* PortScanner.PortDiscovery; + const isolatedCodexAppServerArgs = yield* makeIsolatedCodexAppServerArgs; return CodexAppServerClientFactory.of({ open: Effect.fn("CodexAdapterV2.open")(function* (input) { @@ -1645,20 +1648,10 @@ export const layerAppServerClientFactory: Layer.Layer< ...(input.settings.homePath ? { CODEX_HOME: input.settings.homePath } : {}), }; const args = yield* isolatedCodexAppServerArgs({ - binary: input.settings.binaryPath || "codex", + instanceId: input.instanceId, env: environment, launchArgs: resolveCodexLaunchArgs(input.settings.launchArgs, input.environment), - }).pipe( - Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), - Effect.mapError( - (cause) => - new ProviderAdapter.ProviderAdapterOpenSessionError({ - driver: CODEX_PROVIDER, - providerSessionId: input.providerSessionId, - cause, - }), - ), - ); + }); const command = yield* makeCodexAppServerSpawnCommand({ command: input.settings.binaryPath || "codex", args, diff --git a/apps/server/src/provider/CodexMcpIsolation.test.ts b/apps/server/src/provider/CodexMcpIsolation.test.ts index 5da17c73e1438..0493aa6a2c108 100644 --- a/apps/server/src/provider/CodexMcpIsolation.test.ts +++ b/apps/server/src/provider/CodexMcpIsolation.test.ts @@ -1,55 +1,104 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; -import { writeFakeCli } from "@t3tools/provider-testing/fakeCli"; import { assert, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; -import * as Exit from "effect/Exit"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; +import * as TestClock from "effect/testing/TestClock"; -import { isolatedCodexAppServerArgs } from "./CodexMcpIsolation.ts"; +import { makeIsolatedCodexAppServerArgs } from "./CodexMcpIsolation.ts"; +import { codexAppServerArgs } from "./codexLaunchArgs.ts"; -it.live("isolates configured T3 sign-ins without introducing absent transports", () => +it.live("reads the selected home and profile offline, caching until configuration changes", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const directory = yield* fs.makeTempDirectoryScoped(); - const binary = writeFakeCli({ - directory, - name: "codex", - source: ` - import assert from "node:assert/strict"; - assert.deepEqual(process.argv.slice(2), ["-c", "mcp_servers.t3_macbook.enabled=true", "mcp", "list", "--json"]); - console.log(process.env.SERVERS); - `, - }); - for (const servers of [[], [{ name: "other" }], [{ name: "t3_macbook" }, { name: "other" }]]) { - const args = yield* isolatedCodexAppServerArgs({ - binary, - env: { ...process.env, SERVERS: JSON.stringify(servers) }, - launchArgs: "-c mcp_servers.t3_macbook.enabled=true", - }); - assert.deepEqual(args, [ - "app-server", - "-c", - "thread_unload_delay_secs=0", - "-c", - "mcp_servers.t3_macbook.enabled=true", - ...(servers.some((server) => server.name === "t3_macbook") - ? ["-c", "mcp_servers.t3_macbook.enabled=false"] - : []), - ]); - } + const configPath = `${directory}/config.toml`; + const source = `profile = "work" +[mcp_servers.other] +command = "unused" +[profiles.work.mcp_servers.t3_omarchy] +url = "http://127.0.0.1:1" +[profiles.travel.mcp_servers.t3_macbook] +command = "unused"`; + yield* fs.writeFileString(configPath, source); + let reads = 0; + const isolate = yield* makeIsolatedCodexAppServerArgs.pipe( + Effect.provideService(FileSystem.FileSystem, { + ...fs, + readFileString: (path, encoding) => { + reads++; + return fs.readFileString(path, encoding); + }, + }), + ); + const input = { instanceId: "one", env: { CODEX_HOME: directory }, launchArgs: "" }; + assert.deepEqual(yield* isolate(input), codexAppServerArgs("", ["t3_omarchy"])); + yield* isolate(input); + assert.equal(reads, 1); + assert.deepEqual( + yield* isolate({ ...input, launchArgs: "--profile travel" }), + codexAppServerArgs("--profile travel", ["t3_macbook"]), + ); + assert.deepEqual( + yield* isolate({ ...input, launchArgs: "-c profile=travel" }), + codexAppServerArgs("-c profile=travel", ["t3_macbook"]), + ); + yield* fs.writeFileString(configPath, '[mcp_servers.t3_macbook]\ncommand="unused"\n'); + assert.deepEqual(yield* isolate(input), codexAppServerArgs("", ["t3_macbook"])); + yield* fs.writeFileString(configPath, source); + assert.equal(yield* fs.readFileString(configPath), source); + yield* fs.makeDirectory(`${directory}/.codex`); + yield* fs.writeFileString( + `${directory}/.codex/config.toml`, + '[mcp_servers.t3_macbook]\ncommand="unused"', + ); + assert.deepEqual( + yield* isolate({ ...input, env: { HOME: directory } }), + codexAppServerArgs("", ["t3_macbook"]), + ); }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); -it.live("refuses to launch when inherited MCP configuration cannot be inspected", () => +it.effect("never holds startup behind stalled configuration IO", () => Effect.gen(function* () { + const reading = yield* Deferred.make(); const fs = yield* FileSystem.FileSystem; - const directory = yield* fs.makeTempDirectoryScoped(); - for (const source of ["process.exit(1)", 'console.log("not JSON")']) { - const binary = writeFakeCli({ directory, name: "codex", source }); - const result = yield* Effect.exit( - isolatedCodexAppServerArgs({ binary, env: process.env, launchArgs: "" }), - ); - assert.isTrue(Exit.isFailure(result)); - } + const isolate = yield* makeIsolatedCodexAppServerArgs.pipe( + Effect.provideService(FileSystem.FileSystem, { + ...fs, + exists: () => Deferred.succeed(reading, undefined).pipe(Effect.andThen(Effect.never)), + }), + ); + const startup = yield* isolate({ instanceId: "stalled", env: {}, launchArgs: "" }).pipe( + Effect.forkScoped, + ); + yield* Deferred.await(reading); + yield* TestClock.adjust("250 millis"); + assert.deepEqual(yield* Fiber.join(startup), codexAppServerArgs()); }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); + +it.live( + "adds no absent transports and skips isolation on unreadable or invalid configuration", + () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const directory = yield* fs.makeTempDirectoryScoped(); + const isolate = yield* makeIsolatedCodexAppServerArgs; + const input = { instanceId: "one", env: { CODEX_HOME: directory }, launchArgs: "" }; + for (const source of ["", '[mcp_servers.other]\ncommand="unused"', "invalid TOML !"]) { + yield* fs.writeFileString(`${directory}/config.toml`, source); + assert.deepEqual(yield* isolate(input), codexAppServerArgs()); + } + yield* fs.remove(`${directory}/config.toml`); + assert.deepEqual(yield* isolate(input), codexAppServerArgs()); + const launchArgs = String.raw`-c mcp_servers.t3_macbook.command=\"unused\"`; + assert.deepEqual( + yield* isolate({ ...input, launchArgs }), + codexAppServerArgs(launchArgs, ["t3_macbook"]), + ); + yield* fs.makeDirectory(`${directory}/config.toml`); + assert.deepEqual(yield* isolate(input), codexAppServerArgs()); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/provider/CodexMcpIsolation.ts b/apps/server/src/provider/CodexMcpIsolation.ts index b5c86a827fd3f..ab6e81aa2761c 100644 --- a/apps/server/src/provider/CodexMcpIsolation.ts +++ b/apps/server/src/provider/CodexMcpIsolation.ts @@ -1,56 +1,113 @@ -import { resolveSpawnCommand } from "@t3tools/shared/shell"; +import { deepMerge } from "@t3tools/shared/Struct"; +import * as HostProcess from "@t3tools/shared/HostProcess"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Predicate from "effect/Predicate"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; -import * as Stream from "effect/Stream"; -import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import { parse } from "smol-toml"; import { codexAppServerArgs, codexMcpListLaunchArgs } from "./codexLaunchArgs.ts"; -const decodeServers = Schema.decodeUnknownEffect( - Schema.fromJsonString(Schema.Array(Schema.Struct({ name: Schema.String }))), -); +class CodexMcpConfigError extends Schema.TaggedError()("CodexMcpConfigError", { + cause: Schema.Defect(), +}) {} -class CodexMcpIsolationError extends Schema.TaggedError()( - "CodexMcpIsolationError", - { exitCode: Schema.Number }, -) {} +const table = (value: unknown): Record => (Predicate.isObject(value) ? value : {}); -export const isolatedCodexAppServerArgs = Effect.fn("CodexMcpIsolation.appServerArgs")( - function* (input: { - readonly binary: string; +const configuredServers = (source: string, launchArgs: string) => { + const args = codexMcpListLaunchArgs(launchArgs); + let config: Record = parse(source); + let overrides: Record = {}; + let profile: string | undefined; + for (let index = 0; index < args.length; index++) { + const arg = args[index]!; + if (arg === "--profile" || arg === "-p") { + profile = args[++index]; + } else if (arg.startsWith("--profile=") || arg.startsWith("-p=")) { + profile = arg.slice(arg.indexOf("=") + 1); + } else if ( + arg === "--config" || + arg === "-c" || + arg.startsWith("--config=") || + arg.startsWith("-c=") + ) { + const override = + arg === "--config" || arg === "-c" ? args[++index]! : arg.slice(arg.indexOf("=") + 1); + const key = override.slice(0, override.indexOf("=")); + if ( + key === "profile" || + key === "mcp_servers" || + key.startsWith("mcp_servers.") || + key.startsWith("profiles.") + ) { + let parsed: Record; + try { + parsed = parse(override); + } catch { + parsed = parse(`${key}=${JSON.stringify(override.slice(override.indexOf("=") + 1))}`); + } + overrides = deepMerge(overrides, parsed); + } + } + } + config = deepMerge(config, overrides); + const selectedProfile = profile ?? config.profile; + if (typeof selectedProfile === "string") { + config = deepMerge(config, table(table(config.profiles)[selectedProfile])); + } + return Object.keys(table(deepMerge(config, overrides).mcp_servers)); +}; + +export const makeIsolatedCodexAppServerArgs = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const home = yield* HostProcess.HomeDirectory; + const cache = new Map }>(); + + return Effect.fn("CodexMcpIsolation.appServerArgs")(function* (input: { + readonly instanceId: string; readonly env: NodeJS.ProcessEnv; readonly launchArgs: string; }) { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const command = yield* resolveSpawnCommand( - input.binary, - [...codexMcpListLaunchArgs(input.launchArgs), "mcp", "list", "--json"], - { env: input.env }, - ); - const child = yield* spawner.spawn( - ChildProcess.make(command.command, command.args, { - env: input.env, - shell: command.shell, - forceKillAfter: "1 second", - }), - ); - const [stdout, , exitCode] = yield* Effect.all( - [ - child.stdout.pipe(Stream.decodeText(), Stream.mkString), - child.stderr.pipe(Stream.runDrain), - child.exitCode, - ], - { concurrency: "unbounded" }, - ); - if (exitCode !== 0) { - return yield* new CodexMcpIsolationError({ exitCode: Number(exitCode) }); - } - const servers = yield* decodeServers(stdout); - return codexAppServerArgs( - input.launchArgs, - servers.map((server) => server.name), + return yield* Effect.gen(function* () { + const configPath = path.resolve( + input.env.CODEX_HOME || + path.join(input.env.HOME || input.env.USERPROFILE || home, ".codex"), + "config.toml", + ); + const exists = yield* fs.exists(configPath); + if (!exists) { + const servers = yield* Effect.try({ + try: () => configuredServers("", input.launchArgs), + catch: (cause) => new CodexMcpConfigError({ cause }), + }); + return codexAppServerArgs(input.launchArgs, servers); + } + const info = yield* fs.stat(configPath); + const signature = `${configPath}\0${Option.getOrUndefined(info.mtime)?.getTime()}\0${info.size}\0${input.launchArgs}`; + let cached = cache.get(input.instanceId); + if (cached?.signature !== signature) { + const source = yield* fs.readFileString(configPath); + const servers = yield* Effect.try({ + try: () => configuredServers(source, input.launchArgs), + catch: (cause) => new CodexMcpConfigError({ cause }), + }); + cached = { signature, servers }; + cache.set(input.instanceId, cached); + } + return codexAppServerArgs(input.launchArgs, cached.servers); + }).pipe( + Effect.timeout("250 millis"), + Effect.catch(() => + Effect.logWarning( + "Could not inspect offline Codex MCP configuration; identity isolation skipped", + { + instanceId: input.instanceId, + }, + ).pipe(Effect.as(codexAppServerArgs(input.launchArgs))), + ), ); - }, - Effect.scoped, - Effect.timeout("10 seconds"), -); + }); +}); diff --git a/apps/server/src/provider/codexLaunchArgs.test.ts b/apps/server/src/provider/codexLaunchArgs.test.ts index cee477575221f..b1d2c031572ed 100644 --- a/apps/server/src/provider/codexLaunchArgs.test.ts +++ b/apps/server/src/provider/codexLaunchArgs.test.ts @@ -67,7 +67,7 @@ describe("codexExecLaunchArgs", () => { }); }); -describe("codex exec and mcp list launch args", () => { +describe("codex exec and config launch args", () => { const launchArgs = "--strict-config --oss --local-provider ollama --profile local -p=work --listen off --enable foo"; @@ -85,7 +85,7 @@ describe("codex exec and mcp list launch args", () => { ]); }); - it("keeps the selected profile and drops exec-only flags for codex mcp list", () => { + it("keeps the selected profile and drops exec-only flags for config discovery", () => { NodeAssert.deepStrictEqual(codexMcpListLaunchArgs(launchArgs), [ "--profile", "local", diff --git a/apps/server/src/provider/codexLaunchArgs.ts b/apps/server/src/provider/codexLaunchArgs.ts index 40cbc2547c267..a9d9fe0f7afd7 100644 --- a/apps/server/src/provider/codexLaunchArgs.ts +++ b/apps/server/src/provider/codexLaunchArgs.ts @@ -34,7 +34,7 @@ interface LaunchArgFilter { const SHARED_VALUE_FLAGS = ["--config", "-c", "--enable", "--disable"]; const SHARED_INLINE_PREFIXES = ["--config=", "-c=", "--enable=", "--disable="]; -const MCP_LIST_FILTER: LaunchArgFilter = { +const CONFIG_FILTER: LaunchArgFilter = { switches: new Set(), valueFlags: new Set([...SHARED_VALUE_FLAGS, "--profile", "-p"]), inlinePrefixes: [...SHARED_INLINE_PREFIXES, "--profile=", "-p="], @@ -74,4 +74,4 @@ export const codexExecLaunchArgs = (launchArgs?: string) => filterLaunchArgs(launchArgs, EXEC_FILTER); export const codexMcpListLaunchArgs = (launchArgs?: string) => - filterLaunchArgs(launchArgs, MCP_LIST_FILTER); + filterLaunchArgs(launchArgs, CONFIG_FILTER); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 0a9aa3d0f4c16..11e91d4a3a213 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -944,6 +944,9 @@ importers: saxes: specifier: 6.0.0 version: 6.0.0 + smol-toml: + specifier: ^1.9.1 + version: 1.9.1 stream-chain: specifier: ^4.2.5 version: 4.2.5 From c56bd6d0504459521f2b597387db70b8b981dcb1 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 21:43:05 -0700 Subject: [PATCH 3/4] fix(server): reject pruned finished runs and ignore profile MCP tables --- apps/server/src/mcp/McpEventRelay.ts | 2 + apps/server/src/mcp/McpEvents.test.ts | 47 +++++++++ .../src/provider/CodexMcpIsolation.test.ts | 97 ++++++++++--------- apps/server/src/provider/CodexMcpIsolation.ts | 21 +--- 4 files changed, 104 insertions(+), 63 deletions(-) diff --git a/apps/server/src/mcp/McpEventRelay.ts b/apps/server/src/mcp/McpEventRelay.ts index cf2b3dd8022b5..8f1e1f4443bdb 100644 --- a/apps/server/src/mcp/McpEventRelay.ts +++ b/apps/server/src/mcp/McpEventRelay.ts @@ -455,6 +455,8 @@ const make = Effect.gen(function* () { Effect.mapError(storage), ); if (run !== undefined && start?.firstSequence == null) { + if (Threads.isTerminalRunStatus(run.status)) + return yield* new McpEventsError({ code: -32602, reason: "source_history_pruned" }); const [oldest] = yield* sql<{ occurredAt: string }>`SELECT occurred_at AS "occurredAt" FROM orchestration_events WHERE aggregate_kind = 'project' diff --git a/apps/server/src/mcp/McpEvents.test.ts b/apps/server/src/mcp/McpEvents.test.ts index ee507eeb5a796..f551e9f28a230 100644 --- a/apps/server/src/mcp/McpEvents.test.ts +++ b/apps/server/src/mcp/McpEvents.test.ts @@ -664,6 +664,9 @@ it.effect.each(["empty", "newer", "same", "pruned"] as const)( runs: [ { ...event.payload, + status: "queued", + startedAt: null, + completedAt: null, requestedAt: DateTime.makeUnsafe( history === "newer" ? 2000 : history === "same" ? 1234 : 1, ), @@ -690,6 +693,50 @@ it.effect.each(["empty", "newer", "same", "pruned"] as const)( ).pipe(Effect.provide(NodeCrypto.layer)), ); +it.effect.each(["completed", "failed", "cancelled", "interrupted", "rolled_back"] as const)( + "refuses a finished run with pruned events despite an older retained sentinel (%s)", + (status) => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture; + yield* f.grantWrite(f.principal); + const threadId = ThreadId.make("finished-missing-history"); + const runId = RunId.make("finished-missing-run"); + const [stored] = yield* f.outcome(threadId, runId); + const event = stored!.event; + if (event.type !== "run.updated") return yield* Effect.die("expected run update"); + f.projections.set(threadId, { + ...idleThreadProjection(liveThreadShell(threadId)), + runs: [ + { + ...event.payload, + status, + requestedAt: DateTime.makeUnsafe(2000), + startedAt: DateTime.makeUnsafe(2000), + completedAt: DateTime.makeUnsafe(2001), + }, + ], + }); + const sql = Context.get(f.context, SqlClient.SqlClient); + yield* sql`DELETE FROM orchestration_events`; + yield* f.append([ThreadId.make("retained-sentinel")]); + yield* f.call(f.service.setInbox({ threadId: ThreadId.make("inbox") })); + const notification = yield* f.call( + ThreadNotifications.prepare({}).pipe( + Effect.provideService(McpEvents.McpEvents, f.service), + ), + ); + expect( + yield* f.call(ThreadNotifications.finish(notification, threadId, runId)), + ).toMatchObject({ + status: "failed", + reason: "source_history_pruned", + }); + expect(yield* sql`SELECT id FROM mcp_event_relay_bindings`).toEqual([]); + }), + ).pipe(Effect.provide(NodeCrypto.layer)), +); + it.effect("recovers the source result when an idempotent retry follows a lost binding", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/provider/CodexMcpIsolation.test.ts b/apps/server/src/provider/CodexMcpIsolation.test.ts index 0493aa6a2c108..8055feaddeba9 100644 --- a/apps/server/src/provider/CodexMcpIsolation.test.ts +++ b/apps/server/src/provider/CodexMcpIsolation.test.ts @@ -9,55 +9,62 @@ import * as TestClock from "effect/testing/TestClock"; import { makeIsolatedCodexAppServerArgs } from "./CodexMcpIsolation.ts"; import { codexAppServerArgs } from "./codexLaunchArgs.ts"; -it.live("reads the selected home and profile offline, caching until configuration changes", () => - Effect.gen(function* () { - const fs = yield* FileSystem.FileSystem; - const directory = yield* fs.makeTempDirectoryScoped(); - const configPath = `${directory}/config.toml`; - const source = `profile = "work" -[mcp_servers.other] +it.live( + "reads home MCP servers offline and ignores profile tables, caching until configuration changes", + () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const directory = yield* fs.makeTempDirectoryScoped(); + const configPath = `${directory}/config.toml`; + const source = `profile = "work" +[mcp_servers.t3_omarchy] command = "unused" -[profiles.work.mcp_servers.t3_omarchy] +[profiles.work.mcp_servers.t3_macbook] url = "http://127.0.0.1:1" [profiles.travel.mcp_servers.t3_macbook] command = "unused"`; - yield* fs.writeFileString(configPath, source); - let reads = 0; - const isolate = yield* makeIsolatedCodexAppServerArgs.pipe( - Effect.provideService(FileSystem.FileSystem, { - ...fs, - readFileString: (path, encoding) => { - reads++; - return fs.readFileString(path, encoding); - }, - }), - ); - const input = { instanceId: "one", env: { CODEX_HOME: directory }, launchArgs: "" }; - assert.deepEqual(yield* isolate(input), codexAppServerArgs("", ["t3_omarchy"])); - yield* isolate(input); - assert.equal(reads, 1); - assert.deepEqual( - yield* isolate({ ...input, launchArgs: "--profile travel" }), - codexAppServerArgs("--profile travel", ["t3_macbook"]), - ); - assert.deepEqual( - yield* isolate({ ...input, launchArgs: "-c profile=travel" }), - codexAppServerArgs("-c profile=travel", ["t3_macbook"]), - ); - yield* fs.writeFileString(configPath, '[mcp_servers.t3_macbook]\ncommand="unused"\n'); - assert.deepEqual(yield* isolate(input), codexAppServerArgs("", ["t3_macbook"])); - yield* fs.writeFileString(configPath, source); - assert.equal(yield* fs.readFileString(configPath), source); - yield* fs.makeDirectory(`${directory}/.codex`); - yield* fs.writeFileString( - `${directory}/.codex/config.toml`, - '[mcp_servers.t3_macbook]\ncommand="unused"', - ); - assert.deepEqual( - yield* isolate({ ...input, env: { HOME: directory } }), - codexAppServerArgs("", ["t3_macbook"]), - ); - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + yield* fs.writeFileString(configPath, source); + let reads = 0; + const isolate = yield* makeIsolatedCodexAppServerArgs.pipe( + Effect.provideService(FileSystem.FileSystem, { + ...fs, + readFileString: (path, encoding) => { + reads++; + return fs.readFileString(path, encoding); + }, + }), + ); + const input = { instanceId: "one", env: { CODEX_HOME: directory }, launchArgs: "" }; + assert.deepEqual(yield* isolate(input), codexAppServerArgs("", ["t3_omarchy"])); + yield* isolate(input); + assert.equal(reads, 1); + assert.deepEqual( + yield* isolate({ ...input, launchArgs: "--profile travel" }), + codexAppServerArgs("--profile travel", ["t3_omarchy"]), + ); + assert.deepEqual( + yield* isolate({ ...input, launchArgs: "-c profile=travel" }), + codexAppServerArgs("-c profile=travel", ["t3_omarchy"]), + ); + const profileOverride = String.raw`-c profiles.work.mcp_servers.t3_macbook.command=\"unused\"`; + assert.deepEqual( + yield* isolate({ ...input, launchArgs: profileOverride }), + codexAppServerArgs(profileOverride, ["t3_omarchy"]), + ); + yield* fs.writeFileString(configPath, '[mcp_servers.t3_macbook]\ncommand="unused"\n'); + assert.deepEqual(yield* isolate(input), codexAppServerArgs("", ["t3_macbook"])); + yield* fs.writeFileString(configPath, source); + assert.equal(yield* fs.readFileString(configPath), source); + yield* fs.makeDirectory(`${directory}/.codex`); + yield* fs.writeFileString( + `${directory}/.codex/config.toml`, + '[mcp_servers.t3_macbook]\ncommand="unused"', + ); + assert.deepEqual( + yield* isolate({ ...input, env: { HOME: directory } }), + codexAppServerArgs("", ["t3_macbook"]), + ); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); it.effect("never holds startup behind stalled configuration IO", () => diff --git a/apps/server/src/provider/CodexMcpIsolation.ts b/apps/server/src/provider/CodexMcpIsolation.ts index ab6e81aa2761c..f00749f21bbd7 100644 --- a/apps/server/src/provider/CodexMcpIsolation.ts +++ b/apps/server/src/provider/CodexMcpIsolation.ts @@ -18,16 +18,11 @@ const table = (value: unknown): Record => (Predicate.isObject(v const configuredServers = (source: string, launchArgs: string) => { const args = codexMcpListLaunchArgs(launchArgs); - let config: Record = parse(source); + const config: Record = parse(source); let overrides: Record = {}; - let profile: string | undefined; for (let index = 0; index < args.length; index++) { const arg = args[index]!; - if (arg === "--profile" || arg === "-p") { - profile = args[++index]; - } else if (arg.startsWith("--profile=") || arg.startsWith("-p=")) { - profile = arg.slice(arg.indexOf("=") + 1); - } else if ( + if ( arg === "--config" || arg === "-c" || arg.startsWith("--config=") || @@ -36,12 +31,7 @@ const configuredServers = (source: string, launchArgs: string) => { const override = arg === "--config" || arg === "-c" ? args[++index]! : arg.slice(arg.indexOf("=") + 1); const key = override.slice(0, override.indexOf("=")); - if ( - key === "profile" || - key === "mcp_servers" || - key.startsWith("mcp_servers.") || - key.startsWith("profiles.") - ) { + if (key === "mcp_servers" || key.startsWith("mcp_servers.")) { let parsed: Record; try { parsed = parse(override); @@ -52,11 +42,6 @@ const configuredServers = (source: string, launchArgs: string) => { } } } - config = deepMerge(config, overrides); - const selectedProfile = profile ?? config.profile; - if (typeof selectedProfile === "string") { - config = deepMerge(config, table(table(config.profiles)[selectedProfile])); - } return Object.keys(table(deepMerge(config, overrides).mcp_servers)); }; From 503c11584e832a3d2ab88d8c5b9230f018ad2b78 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 23:14:20 -0700 Subject: [PATCH 4/4] fix(server): remove duplicate launch warning schema field --- apps/server/src/mcp/ThreadLaunchMcpService.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/apps/server/src/mcp/ThreadLaunchMcpService.ts b/apps/server/src/mcp/ThreadLaunchMcpService.ts index cb4c1bb8e64a7..0c7e4920ee73a 100644 --- a/apps/server/src/mcp/ThreadLaunchMcpService.ts +++ b/apps/server/src/mcp/ThreadLaunchMcpService.ts @@ -85,7 +85,6 @@ export const launchInput = Schema.Struct({ }); export const launchResult = Schema.Struct({ notify: Schema.optional(McpThreadNotifyResult), - warning: Schema.optional(Schema.String), threadId: ThreadId, link: Schema.optional(Schema.String), projectId: ProjectId,