diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 06b7dd99bd4..8aeacd870cc 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -1,8 +1,9 @@ import * as NodeAssert from "node:assert/strict"; +import { it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; -import { describe, it } from "vite-plus/test"; +import { describe } from "vite-plus/test"; import { ThreadId } from "@t3tools/contracts"; import * as CodexErrors from "effect-codex-app-server/errors"; import * as CodexRpc from "effect-codex-app-server/rpc"; @@ -19,6 +20,23 @@ import { } from "./CodexSessionRuntime.ts"; const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); +describe("CodexSessionRuntimeIdentifierGenerationError", () => { + it("retains identifier purpose and the random source failure", () => { + const cause = new Error("random source unavailable"); + const error = new CodexErrors.CodexAppServerIdentifierGenerationError({ + purpose: "provider-event", + cause, + }); + + NodeAssert.equal(error.purpose, "provider-event"); + NodeAssert.strictEqual(error.cause, cause); + NodeAssert.equal( + error.message, + "Failed to generate Codex App Server identifier for provider-event.", + ); + }); +}); + function makeThreadOpenResponse( threadId: string, ): CodexRpc.ClientRequestResponsesByMethod["thread/start"] { @@ -43,6 +61,32 @@ function makeThreadOpenResponse( } describe("buildTurnStartParams", () => { + it("keeps invalid turn values only in the schema cause", () => { + const secret = "codex-turn-input-secret-sentinel"; + const error = Effect.runSync( + buildTurnStartParams({ + threadId: "provider-thread-1", + runtimeMode: "full-access", + attachments: [ + { + type: "image", + url: { secret } as unknown as string, + }, + ], + }).pipe(Effect.flip), + ); + const { cause, ...directDiagnostics } = error; + + NodeAssert.equal(error.operation, "decode-request-payload"); + NodeAssert.equal(error.method, "turn/start"); + NodeAssert.ok((error.issueCount ?? 0) > 0); + NodeAssert.ok(error.issueKinds?.includes("Pointer")); + NodeAssert.ok((error.maximumPathDepth ?? 0) > 0); + NodeAssert.ok(Schema.isSchemaError(cause)); + NodeAssert.doesNotMatch(error.message, new RegExp(secret)); + NodeAssert.doesNotMatch(JSON.stringify(directDiagnostics), new RegExp(secret)); + }); + it("includes plan collaboration mode when requested", () => { const params = Effect.runSync( buildTurnStartParams({ @@ -223,29 +267,29 @@ describe("isRecoverableThreadResumeError", () => { }); describe("openCodexThread", () => { - it("falls back to thread/start when resume fails recoverably", async () => { - const calls: Array<{ method: "thread/start" | "thread/resume"; payload: unknown }> = []; - const started = makeThreadOpenResponse("fresh-thread"); - const client = { - request: ( - method: M, - payload: CodexRpc.ClientRequestParamsByMethod[M], - ) => { - calls.push({ method, payload }); - if (method === "thread/resume") { - return Effect.fail( - new CodexErrors.CodexAppServerRequestError({ - code: -32603, - errorMessage: "thread not found", - }), - ); - } - return Effect.succeed(started as CodexRpc.ClientRequestResponsesByMethod[M]); - }, - }; + it.effect("falls back to thread/start when resume fails recoverably", () => + Effect.gen(function* () { + const calls: Array<{ method: "thread/start" | "thread/resume"; payload: unknown }> = []; + const started = makeThreadOpenResponse("fresh-thread"); + const client = { + request: ( + method: M, + payload: CodexRpc.ClientRequestParamsByMethod[M], + ) => { + calls.push({ method, payload }); + if (method === "thread/resume") { + return Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "thread not found", + }), + ); + } + return Effect.succeed(started as CodexRpc.ClientRequestResponsesByMethod[M]); + }, + }; - const opened = await Effect.runPromise( - openCodexThread({ + const opened = yield* openCodexThread({ client, threadId: ThreadId.make("thread-1"), runtimeMode: "full-access", @@ -253,51 +297,49 @@ describe("openCodexThread", () => { requestedModel: "gpt-5.3-codex", serviceTier: undefined, resumeThreadId: "stale-thread", - }), - ); + }); - NodeAssert.equal(opened.thread.id, "fresh-thread"); - NodeAssert.deepStrictEqual( - calls.map((call) => call.method), - ["thread/resume", "thread/start"], - ); - }); + NodeAssert.equal(opened.thread.id, "fresh-thread"); + NodeAssert.deepStrictEqual( + calls.map((call) => call.method), + ["thread/resume", "thread/start"], + ); + }), + ); - it("propagates non-recoverable resume failures", async () => { - const client = { - request: ( - method: M, - _payload: CodexRpc.ClientRequestParamsByMethod[M], - ) => { - if (method === "thread/resume") { - return Effect.fail( - new CodexErrors.CodexAppServerRequestError({ - code: -32603, - errorMessage: "timed out waiting for server", - }), + it.effect("propagates non-recoverable resume failures", () => + Effect.gen(function* () { + const client = { + request: ( + method: M, + _payload: CodexRpc.ClientRequestParamsByMethod[M], + ) => { + if (method === "thread/resume") { + return Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "timed out waiting for server", + }), + ); + } + return Effect.succeed( + makeThreadOpenResponse("fresh-thread") as CodexRpc.ClientRequestResponsesByMethod[M], ); - } - return Effect.succeed( - makeThreadOpenResponse("fresh-thread") as CodexRpc.ClientRequestResponsesByMethod[M], - ); - }, - }; + }, + }; - await NodeAssert.rejects( - Effect.runPromise( - openCodexThread({ - client, - threadId: ThreadId.make("thread-1"), - runtimeMode: "full-access", - cwd: "/tmp/project", - requestedModel: "gpt-5.3-codex", - serviceTier: undefined, - resumeThreadId: "stale-thread", - }), - ), - (error: unknown) => - isCodexAppServerRequestError(error) && - error.errorMessage === "timed out waiting for server", - ); - }); + const error = yield* openCodexThread({ + client, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: "stale-thread", + }).pipe(Effect.flip); + + NodeAssert.ok(isCodexAppServerRequestError(error)); + NodeAssert.equal(error.errorMessage, "timed out waiting for server"); + }), + ); }); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 03957081ded..99ac498f0c3 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -26,10 +26,9 @@ import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; -import * as Scope from "effect/Scope"; import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; -import * as SchemaIssue from "effect/SchemaIssue"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import * as CodexClient from "effect-codex-app-server/client"; import * as CodexErrors from "effect-codex-app-server/errors"; @@ -89,7 +88,6 @@ const decodeCodexTurnStartParamsWithCollaborationMode = Schema.decodeUnknownEffe export type CodexTurnStartParamsWithCollaborationMode = typeof CodexTurnStartParamsWithCollaborationMode.Type; -const formatSchemaIssue = SchemaIssue.makeFormatterDefault(); export type CodexResumeCursor = typeof CodexResumeCursorSchema.Type; type CodexServiceTier = NonNullable; @@ -390,7 +388,13 @@ export function buildTurnStartParams(input: { ...(input.effort ? { effort: input.effort } : {}), ...(collaborationMode ? { collaborationMode } : {}), }).pipe( - Effect.mapError((error) => toProtocolParseError("Invalid turn/start request payload", error)), + Effect.mapError((cause) => + CodexErrors.CodexAppServerProtocolParseError.fromSchemaError( + "decode-request-payload", + cause, + { method: "turn/start" }, + ), + ), ); } @@ -468,7 +472,7 @@ export const openCodexThread = (input: { requestedRuntimeMode: input.runtimeMode, resumeThreadId, recoverable: true, - cause: error.message, + cause: error, }).pipe(Effect.andThen(input.client.request("thread/start", startParams))), ), ); @@ -658,16 +662,6 @@ function toCodexUserInputAnswers( ).pipe(Effect.map((entries) => Object.fromEntries(entries))); } -function toProtocolParseError( - detail: string, - cause: Schema.SchemaError, -): CodexErrors.CodexAppServerProtocolParseError { - return new CodexErrors.CodexAppServerProtocolParseError({ - detail: `${detail}: ${formatSchemaIssue(cause.issue)}`, - cause, - }); -} - function currentProviderThreadId(session: ProviderSession): string | undefined { return readResumeCursorThreadId(session.resumeCursor); } @@ -760,15 +754,16 @@ export const makeCodexSessionRuntime = ( ); const serverNotifications = yield* Queue.unbounded(); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); - const randomUUIDv4 = crypto.randomUUIDv4.pipe( - Effect.mapError( - (cause) => - new CodexErrors.CodexAppServerTransportError({ - detail: "Failed to generate Codex runtime identifier.", - cause, - }), - ), - ); + const randomUUIDv4 = (purpose: CodexErrors.CodexAppServerIdentifierPurpose) => + crypto.randomUUIDv4.pipe( + Effect.mapError( + (cause) => + new CodexErrors.CodexAppServerIdentifierGenerationError({ + purpose, + cause, + }), + ), + ); const sessionCreatedAt = yield* nowIso; const initialSession = { @@ -788,7 +783,7 @@ export const makeCodexSessionRuntime = ( const emitEvent = (event: Omit) => Effect.gen(function* () { - const id = yield* randomUUIDv4; + const id = yield* randomUUIDv4("provider-event"); return yield* offerEvent({ id: EventId.make(id), provider: PROVIDER, @@ -956,7 +951,7 @@ export const makeCodexSessionRuntime = ( yield* client.handleServerRequest("item/commandExecution/requestApproval", (payload) => Effect.gen(function* () { - const requestId = ApprovalRequestId.make(yield* randomUUIDv4); + const requestId = ApprovalRequestId.make(yield* randomUUIDv4("command-approval-request")); const turnId = TurnId.make(payload.turnId); const itemId = ProviderItemId.make(payload.itemId); const decision = yield* Deferred.make(); @@ -1012,7 +1007,9 @@ export const makeCodexSessionRuntime = ( yield* client.handleServerRequest("item/fileChange/requestApproval", (payload) => Effect.gen(function* () { - const requestId = ApprovalRequestId.make(yield* randomUUIDv4); + const requestId = ApprovalRequestId.make( + yield* randomUUIDv4("file-change-approval-request"), + ); const turnId = TurnId.make(payload.turnId); const itemId = ProviderItemId.make(payload.itemId); const decision = yield* Deferred.make(); @@ -1068,7 +1065,7 @@ export const makeCodexSessionRuntime = ( yield* client.handleServerRequest("item/tool/requestUserInput", (payload) => Effect.gen(function* () { - const requestId = ApprovalRequestId.make(yield* randomUUIDv4); + const requestId = ApprovalRequestId.make(yield* randomUUIDv4("user-input-request")); const turnId = TurnId.make(payload.turnId); const itemId = ProviderItemId.make(payload.itemId); const answers = yield* Deferred.make(); @@ -1293,7 +1290,11 @@ export const makeCodexSessionRuntime = ( const rawResponse = yield* client.raw.request("turn/start", params); const response = yield* decodeV2TurnStartResponse(rawResponse).pipe( Effect.mapError((error) => - toProtocolParseError("Invalid turn/start response payload", error), + CodexErrors.CodexAppServerProtocolParseError.fromSchemaError( + "decode-response-payload", + error, + { method: "turn/start" }, + ), ), ); const turnId = TurnId.make(response.turn.id); diff --git a/packages/effect-codex-app-server/src/_internal/shared.test.ts b/packages/effect-codex-app-server/src/_internal/shared.test.ts new file mode 100644 index 00000000000..62b1373f12c --- /dev/null +++ b/packages/effect-codex-app-server/src/_internal/shared.test.ts @@ -0,0 +1,151 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +import * as CodexError from "../errors.ts"; +import * as Shared from "./shared.ts"; + +const decodeNestedNumberPayload = Schema.decodeUnknownEffect( + Schema.Struct({ profile: Schema.Struct({ token: Schema.Number }) }), +); +const encodeUnknownJson = Schema.encodeSync(Schema.UnknownFromJsonString); + +it.effect("preserves schema decode diagnostics without deriving the message from the cause", () => + Effect.gen(function* () { + const error = yield* Shared.decodeOptionalPayload("thread/start", Schema.String, 42).pipe( + Effect.flip, + ); + + assert.instanceOf(error, CodexError.CodexAppServerRequestError); + assert.equal(error.code, -32602); + assert.equal(error.method, "thread/start"); + assert.equal(error.operation, "decode-payload"); + assert.equal( + error.message, + "Invalid payload for method 'thread/start' during 'decode-payload'", + ); + assert.isTrue(Schema.isSchemaError(error.cause)); + + const protocolError = error.toProtocolError(); + assert.equal(protocolError.code, -32602); + assert.equal(protocolError.message, error.message); + assert.property(protocolError, "data"); + assert.notProperty(protocolError, "method"); + assert.notProperty(protocolError, "operation"); + assert.notProperty(protocolError, "cause"); + }), +); + +it.effect("preserves schema encode diagnostics", () => + Effect.gen(function* () { + const error = yield* Shared.encodeOptionalPayload( + "thread/start", + Schema.Number, + "not-a-number" as never, + ).pipe(Effect.flip); + + assert.equal(error.method, "thread/start"); + assert.equal(error.operation, "encode-payload"); + assert.equal( + error.message, + "Invalid payload for method 'thread/start' during 'encode-payload'", + ); + assert.isTrue(Schema.isSchemaError(error.cause)); + }), +); + +it.effect("does not invent a cause when a method has no payload schema", () => + Effect.gen(function* () { + const secret = "unexpected-payload-secret"; + const error = yield* Shared.decodeOptionalPayload("initialized", undefined, { + token: secret, + }).pipe(Effect.flip); + + assert.equal(error.method, "initialized"); + assert.equal(error.operation, "decode-payload"); + assert.equal(error.payloadKind, "object"); + assert.deepEqual(error.data, { payloadKind: "object" }); + assert.isUndefined(error.cause); + assert.notInclude(error.message, secret); + assert.notInclude(encodeUnknownJson(error.toProtocolError()), secret); + }), +); + +it.effect("keeps invalid payload values only in the exact schema cause", () => + Effect.gen(function* () { + const secret = "codex-schema-payload-secret"; + const cause = yield* decodeNestedNumberPayload({ profile: { token: secret } }).pipe( + Effect.flip, + ); + const error = CodexError.CodexAppServerRequestError.invalidPayload( + "thread/start", + "decode-payload", + cause, + ); + const { cause: directCause, ...directDiagnostics } = error; + + assert.strictEqual(directCause, cause); + assert.equal(error.method, "thread/start"); + assert.equal(error.operation, "decode-payload"); + assert.equal(error.maximumPathDepth, 2); + assert.isAbove(error.issueCount ?? 0, 0); + assert.include(error.issueKinds ?? [], "Pointer"); + assert.notInclude(error.message, secret); + assert.notInclude(encodeUnknownJson(directDiagnostics), secret); + assert.notInclude(encodeUnknownJson(error.toProtocolError()), secret); + }), +); + +it.effect("retains the request-handler error as the internal error cause", () => + Effect.gen(function* () { + const rootCause = new Error("socket closed"); + const source = new CodexError.CodexAppServerTransportError({ + operation: "read-input-stream", + cause: rootCause, + }); + const error = yield* Shared.runHandler( + (_payload: void) => Effect.fail(source), + undefined, + "thread/start", + ).pipe(Effect.flip); + + assert.equal(error.code, -32603); + assert.equal(error.method, "thread/start"); + assert.equal(error.operation, "handle-request"); + assert.equal( + error.message, + "Codex App Server request handler failed for method 'thread/start'", + ); + assert.strictEqual(error.cause, source); + assert.strictEqual(source.cause, rootCause); + assert.notInclude(error.message, source.message); + }), +); + +it.effect("passes request errors through without adding a wrapper", () => + Effect.gen(function* () { + const source = CodexError.CodexAppServerRequestError.invalidParams("Invalid thread id"); + const error = yield* Shared.runHandler( + (_payload: void) => Effect.fail(source), + undefined, + "thread/start", + ).pipe(Effect.flip); + + assert.strictEqual(error, source); + }), +); + +it.effect("retains the full notification payload decode cause chain", () => + Effect.gen(function* () { + const error = yield* Shared.decodeNotificationPayload( + "item/agentMessage/delta", + Schema.String, + 42, + ).pipe(Effect.flip); + + assert.equal(error.method, "item/agentMessage/delta"); + assert.equal(error.operation, "decode-notification-payload"); + assert.instanceOf(error.cause, CodexError.CodexAppServerRequestError); + assert.isTrue(Schema.isSchemaError(error.cause.cause)); + }), +); diff --git a/packages/effect-codex-app-server/src/_internal/shared.ts b/packages/effect-codex-app-server/src/_internal/shared.ts index 99fe8e5360b..34155348abf 100644 --- a/packages/effect-codex-app-server/src/_internal/shared.ts +++ b/packages/effect-codex-app-server/src/_internal/shared.ts @@ -1,11 +1,8 @@ import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; -import * as SchemaIssue from "effect/SchemaIssue"; import * as CodexError from "../errors.ts"; -const formatSchemaIssue = SchemaIssue.makeFormatterDefault(); - export const JsonRpcId = Schema.Union([Schema.Number, Schema.String]); export const JsonRpcError = Schema.Struct({ @@ -30,16 +27,13 @@ export const decodeOptionalPayload = ( return Effect.sync(() => undefined as A); } return Effect.fail( - CodexError.CodexAppServerRequestError.invalidParams(`${method} does not accept params`, raw), + CodexError.CodexAppServerRequestError.unexpectedPayload(method, "decode-payload", raw), ); } return Schema.decodeUnknownEffect(schema)(raw).pipe( Effect.mapError((error) => - CodexError.CodexAppServerRequestError.invalidParams( - `Invalid ${method} payload: ${formatSchemaIssue(error.issue)}`, - { issue: error.issue }, - ), + CodexError.CodexAppServerRequestError.invalidPayload(method, "decode-payload", error), ), ); }; @@ -54,19 +48,13 @@ export const encodeOptionalPayload = ( return Effect.sync(() => undefined); } return Effect.fail( - CodexError.CodexAppServerRequestError.invalidParams( - `${method} does not accept params`, - payload, - ), + CodexError.CodexAppServerRequestError.unexpectedPayload(method, "encode-payload", payload), ); } return Schema.encodeEffect(schema)(payload).pipe( Effect.mapError((error) => - CodexError.CodexAppServerRequestError.invalidParams( - `Invalid ${method} payload: ${formatSchemaIssue(error.issue)}`, - { issue: error.issue }, - ), + CodexError.CodexAppServerRequestError.invalidPayload(method, "encode-payload", error), ), ); }; @@ -77,12 +65,12 @@ export const decodeNotificationPayload = ( raw: unknown, ): Effect.Effect => decodeOptionalPayload(method, schema, raw).pipe( - Effect.mapError( - (error) => - new CodexError.CodexAppServerProtocolParseError({ - detail: error.message, - cause: error, - }), + Effect.mapError((error) => + CodexError.CodexAppServerProtocolParseError.fromRequestError( + "decode-notification-payload", + method, + error, + ), ), ); @@ -96,6 +84,8 @@ export const runHandler = Effect.fnUntraced(function* ( } return yield* handler(payload).pipe( - Effect.mapError((error) => CodexError.normalizeToRequestError(error)), + Effect.mapError((error) => + CodexError.CodexAppServerRequestError.fromAppServerError(error, method), + ), ); }); diff --git a/packages/effect-codex-app-server/src/_internal/stdio.ts b/packages/effect-codex-app-server/src/_internal/stdio.ts index ced07fb53e2..9167129db5c 100644 --- a/packages/effect-codex-app-server/src/_internal/stdio.ts +++ b/packages/effect-codex-app-server/src/_internal/stdio.ts @@ -50,7 +50,7 @@ export const makeTerminationError = ( Effect.match(handle.exitCode, { onFailure: (cause) => new CodexError.CodexAppServerTransportError({ - detail: "Failed to determine Codex App Server process exit status", + operation: "read-process-exit-status", cause, }), onSuccess: (code) => new CodexError.CodexAppServerProcessExitedError({ code }), diff --git a/packages/effect-codex-app-server/src/errors.ts b/packages/effect-codex-app-server/src/errors.ts index d9977c63597..2f769f47de2 100644 --- a/packages/effect-codex-app-server/src/errors.ts +++ b/packages/effect-codex-app-server/src/errors.ts @@ -1,4 +1,119 @@ import * as Schema from "effect/Schema"; +import type * as SchemaIssue from "effect/SchemaIssue"; + +export const CodexAppServerRequestOperation = Schema.Literals([ + "decode-payload", + "encode-payload", + "handle-request", +]); +export type CodexAppServerRequestOperation = typeof CodexAppServerRequestOperation.Type; + +export const CodexAppServerSchemaIssueKind = Schema.Literals([ + "Filter", + "Encoding", + "Pointer", + "Composite", + "AnyOf", + "InvalidType", + "InvalidValue", + "MissingKey", + "UnexpectedKey", + "Forbidden", + "OneOf", +]); +export type CodexAppServerSchemaIssueKind = typeof CodexAppServerSchemaIssueKind.Type; + +export interface CodexAppServerSchemaIssueDiagnostics { + readonly issueCount: number; + readonly issueKinds: ReadonlyArray; + readonly maximumPathDepth: number; +} + +const schemaIssueDiagnostics = (root: SchemaIssue.Issue): CodexAppServerSchemaIssueDiagnostics => { + let issueCount = 0; + let maximumPathDepth = 0; + const issueKinds = new Set(); + + const visit = (issue: SchemaIssue.Issue, pathDepth: number): void => { + issueCount += 1; + issueKinds.add(issue._tag); + maximumPathDepth = Math.max(maximumPathDepth, pathDepth); + switch (issue._tag) { + case "Filter": + case "Encoding": + visit(issue.issue, pathDepth); + break; + case "Pointer": + visit(issue.issue, pathDepth + issue.path.length); + break; + case "Composite": + case "AnyOf": + for (const child of issue.issues) visit(child, pathDepth); + break; + } + }; + + visit(root, 0); + return { + issueCount, + issueKinds: [...issueKinds], + maximumPathDepth, + }; +}; + +export const CodexAppServerPayloadKind = Schema.Literals([ + "null", + "array", + "string", + "number", + "boolean", + "bigint", + "object", + "symbol", + "function", + "undefined", +]); +export type CodexAppServerPayloadKind = typeof CodexAppServerPayloadKind.Type; + +const payloadKind = (payload: unknown): CodexAppServerPayloadKind => { + if (payload === null) return "null"; + if (Array.isArray(payload)) return "array"; + return typeof payload; +}; + +export interface CodexAppServerRequestDiagnostics { + readonly method?: string; + readonly operation?: CodexAppServerRequestOperation; + readonly cause?: unknown; + readonly issueCount?: number; + readonly issueKinds?: ReadonlyArray; + readonly maximumPathDepth?: number; + readonly payloadKind?: CodexAppServerPayloadKind; +} + +export const CodexAppServerProtocolParseOperation = Schema.Literals([ + "encode-wire-message", + "decode-wire-message", + "route-wire-message", + "decode-notification-payload", + "decode-request-payload", + "decode-response-payload", +]); +export type CodexAppServerProtocolParseOperation = typeof CodexAppServerProtocolParseOperation.Type; + +export const CodexAppServerTransportOperation = Schema.Literals([ + "read-input-stream", + "read-process-exit-status", +]); +export type CodexAppServerTransportOperation = typeof CodexAppServerTransportOperation.Type; + +export const CodexAppServerIdentifierPurpose = Schema.Literals([ + "provider-event", + "command-approval-request", + "file-change-approval-request", + "user-input-request", +]); +export type CodexAppServerIdentifierPurpose = typeof CodexAppServerIdentifierPurpose.Type; export interface CodexAppServerProtocolErrorShape { readonly code: number; @@ -37,24 +152,79 @@ export class CodexAppServerProcessExitedError extends Schema.TaggedErrorClass()( "CodexAppServerProtocolParseError", { - detail: Schema.String, + operation: CodexAppServerProtocolParseOperation, + method: Schema.optionalKey(Schema.String), + detail: Schema.optionalKey(Schema.String), + issueCount: Schema.optionalKey(Schema.Number), + issueKinds: Schema.optionalKey(Schema.Array(CodexAppServerSchemaIssueKind)), + maximumPathDepth: Schema.optionalKey(Schema.Number), cause: Schema.optional(Schema.Defect()), }, ) { override get message() { - return `Failed to parse Codex App Server protocol message: ${this.detail}`; + const method = this.method === undefined ? "" : ` for method '${this.method}'`; + return `Codex App Server protocol operation '${this.operation}' failed${method}.`; + } + + static fromSchemaError( + operation: CodexAppServerProtocolParseOperation, + cause: Schema.SchemaError, + context: { readonly method?: string } = {}, + ) { + return new CodexAppServerProtocolParseError({ + operation, + ...context, + ...schemaIssueDiagnostics(cause.issue), + cause, + }); + } + + static fromRequestError( + operation: CodexAppServerProtocolParseOperation, + method: string, + cause: CodexAppServerRequestError, + ) { + return new CodexAppServerProtocolParseError({ + operation, + method, + ...(cause.issueCount === undefined ? {} : { issueCount: cause.issueCount }), + ...(cause.issueKinds === undefined ? {} : { issueKinds: cause.issueKinds }), + ...(cause.maximumPathDepth === undefined ? {} : { maximumPathDepth: cause.maximumPathDepth }), + cause, + }); } } export class CodexAppServerTransportError extends Schema.TaggedErrorClass()( "CodexAppServerTransportError", { - detail: Schema.String, + operation: CodexAppServerTransportOperation, cause: Schema.Defect(), }, ) { override get message() { - return this.detail; + return `Codex App Server transport operation '${this.operation}' failed.`; + } +} + +export class CodexAppServerIdentifierGenerationError extends Schema.TaggedErrorClass()( + "CodexAppServerIdentifierGenerationError", + { + purpose: CodexAppServerIdentifierPurpose, + cause: Schema.Defect(), + }, +) { + override get message() { + return `Failed to generate Codex App Server identifier for ${this.purpose}.`; + } +} + +export class CodexAppServerInputStreamEndedError extends Schema.TaggedErrorClass()( + "CodexAppServerInputStreamEndedError", + {}, +) { + override get message() { + return "Codex App Server input stream ended."; } } @@ -64,6 +234,13 @@ export class CodexAppServerRequestError extends Schema.TaggedErrorClass { const bigintError = yield* transport.notify("x/test", 1n).pipe(Effect.flip); assert.instanceOf(bigintError, CodexError.CodexAppServerProtocolParseError); - assert.equal(bigintError.detail, "Failed to encode Codex App Server message"); + assert.equal(bigintError.operation, "encode-wire-message"); + assert.exists(bigintError.cause); + assert.equal( + bigintError.message, + "Codex App Server protocol operation 'encode-wire-message' failed.", + ); const circular: Record = {}; circular.self = circular; const circularError = yield* transport.notify("x/test", circular).pipe(Effect.flip); assert.instanceOf(circularError, CodexError.CodexAppServerProtocolParseError); - assert.equal(circularError.detail, "Failed to encode Codex App Server message"); + assert.equal(circularError.operation, "encode-wire-message"); + assert.exists(circularError.cause); + }), + ); + + it.effect("logs decode failures without copying the cause or wire payload", () => + Effect.gen(function* () { + const secret = "codex-wire-secret-sentinel"; + const { stdio, input } = yield* makeInMemoryStdio(); + const events: Array = []; + const termination = yield* Deferred.make(); + yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ + stdio, + logIncoming: true, + logger: (event) => + Effect.sync(() => { + events.push(event); + }), + onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid), + }); + + yield* Queue.offer(input, encoder.encode(`{"secret":"${secret}"\n`)); + yield* Deferred.await(termination); + + const event = events.find(({ stage }) => stage === "decode_failed"); + assert.exists(event); + assert.equal(event.direction, "incoming"); + const payload = event.payload as Record; + assert.equal(payload.operation, "decode-wire-message"); + assert.isNumber(payload.issueCount); + assert.isArray(payload.issueKinds); + assert.isNumber(payload.maximumPathDepth); + assert.equal("cause" in payload, false); + assert.equal("detail" in payload, false); + assert.notInclude(encodeUnknownJsonString(event), secret); + }), + ); + + it.effect("classifies an input stream ending without inventing a cause", () => + Effect.gen(function* () { + const { stdio, input } = yield* makeInMemoryStdio(); + const termination = yield* Deferred.make(); + yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ + stdio, + onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid), + }); + + yield* Queue.end(input); + + const error = yield* Deferred.await(termination); + assert.instanceOf(error, CodexError.CodexAppServerInputStreamEndedError); + assert.equal(error.message, "Codex App Server input stream ended."); + assert.equal("cause" in error, false); }), ); }); diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index 0fc2ce73c5c..c0f07f95a5a 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/packages/effect-codex-app-server/src/protocol.ts @@ -94,12 +94,8 @@ const encodeWireMessage = ( ): Effect.Effect => encodeJsonString(message).pipe( Effect.map((encoded) => `${encoded}\n`), - Effect.mapError( - (cause) => - new CodexError.CodexAppServerProtocolParseError({ - detail: "Failed to encode Codex App Server message", - cause, - }), + Effect.mapError((cause) => + CodexError.CodexAppServerProtocolParseError.fromSchemaError("encode-wire-message", cause), ), ); @@ -107,20 +103,19 @@ const decodeWireMessage = ( line: string, ): Effect.Effect => decodeJsonString(line).pipe( - Effect.mapError( - (cause) => - new CodexError.CodexAppServerProtocolParseError({ - detail: "Failed to decode Codex App Server wire message", - cause, - }), + Effect.mapError((cause) => + CodexError.CodexAppServerProtocolParseError.fromSchemaError("decode-wire-message", cause), ), ); -const normalizeIncomingError = (error: unknown, detail: string): CodexError.CodexAppServerError => +const normalizeIncomingError = ( + error: unknown, + operation: CodexError.CodexAppServerTransportOperation, +): CodexError.CodexAppServerError => isCodexAppServerError(error) ? error : new CodexError.CodexAppServerTransportError({ - detail, + operation, cause: error, }); @@ -262,7 +257,13 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa ? options.onRequest(request).pipe( Effect.matchEffect({ onFailure: (error) => - respondError(request.id, CodexError.normalizeToRequestError(error)), + respondError( + request.id, + CodexError.CodexAppServerRequestError.fromAppServerError( + error, + request.method, + ), + ), onSuccess: (result) => respond(request.id, result), }), ) @@ -292,6 +293,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa return Effect.fail( new CodexError.CodexAppServerProtocolParseError({ detail: "Received protocol message in an unknown shape", + operation: "route-wire-message", }), ); }; @@ -318,8 +320,13 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa direction: "incoming", stage: "decode_failed", payload: { - detail: error.detail, - cause: error.cause, + operation: error.operation, + ...(error.method === undefined ? {} : { method: error.method }), + ...(error.issueCount === undefined ? {} : { issueCount: error.issueCount }), + ...(error.issueKinds === undefined ? {} : { issueKinds: error.issueKinds }), + ...(error.maximumPathDepth === undefined + ? {} + : { maximumPathDepth: error.maximumPathDepth }), }, }), ), @@ -340,7 +347,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa Effect.matchEffect({ onFailure: (error) => handleTermination(() => - Effect.succeed(normalizeIncomingError(error, "Codex App Server input stream failed")), + Effect.succeed(normalizeIncomingError(error, "read-input-stream")), ), onSuccess: () => Ref.get(remainder).pipe( @@ -351,12 +358,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa handleTermination( () => options.terminationError ?? - Effect.succeed( - new CodexError.CodexAppServerTransportError({ - detail: "Codex App Server input stream ended", - cause: new Error("Codex App Server input stream ended"), - }), - ), + Effect.succeed(new CodexError.CodexAppServerInputStreamEndedError({})), ), }), ),