Skip to content
17 changes: 17 additions & 0 deletions apps/server/src/provider/Layers/CodexSessionRuntime.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,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"] {
Expand Down
61 changes: 34 additions & 27 deletions apps/server/src/provider/Layers/CodexSessionRuntime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,15 @@ 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) =>
new CodexErrors.CodexAppServerProtocolParseError({
operation: "decode-request-payload",
method: "turn/start",
detail: formatSchemaIssue(cause.issue),
cause,
}),
),
);
}

Expand Down Expand Up @@ -468,7 +476,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))),
),
);
Expand Down Expand Up @@ -658,16 +666,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);
}
Expand Down Expand Up @@ -760,15 +758,16 @@ export const makeCodexSessionRuntime = (
);
const serverNotifications = yield* Queue.unbounded<CodexServerNotification>();
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 = {
Expand All @@ -788,7 +787,7 @@ export const makeCodexSessionRuntime = (

const emitEvent = (event: Omit<ProviderEvent, "id" | "provider" | "createdAt">) =>
Effect.gen(function* () {
const id = yield* randomUUIDv4;
const id = yield* randomUUIDv4("provider-event");
return yield* offerEvent({
id: EventId.make(id),
provider: PROVIDER,
Expand Down Expand Up @@ -956,7 +955,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<ProviderApprovalDecision>();
Expand Down Expand Up @@ -1012,7 +1011,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<ProviderApprovalDecision>();
Expand Down Expand Up @@ -1068,7 +1069,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<ProviderUserInputAnswers>();
Expand Down Expand Up @@ -1292,8 +1293,14 @@ 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),
Effect.mapError(
(error) =>
new CodexErrors.CodexAppServerProtocolParseError({
operation: "decode-response-payload",
method: "turn/start",
detail: formatSchemaIssue(error.issue),
cause: error,
}),
),
);
const turnId = TurnId.make(response.turn.id);
Expand Down
151 changes: 151 additions & 0 deletions packages/effect-codex-app-server/src/_internal/shared.test.ts
Original file line number Diff line number Diff line change
@@ -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<never, never>("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));
}),
);
36 changes: 13 additions & 23 deletions packages/effect-codex-app-server/src/_internal/shared.ts
Original file line number Diff line number Diff line change
@@ -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({
Expand All @@ -30,16 +27,13 @@ export const decodeOptionalPayload = <A, I>(
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),
),
);
};
Expand All @@ -54,19 +48,13 @@ export const encodeOptionalPayload = <A, I>(
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),
),
);
};
Expand All @@ -77,12 +65,12 @@ export const decodeNotificationPayload = <A, I>(
raw: unknown,
): Effect.Effect<A, CodexError.CodexAppServerProtocolParseError> =>
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,
),
),
);

Expand All @@ -96,6 +84,8 @@ export const runHandler = Effect.fnUntraced(function* <A, B>(
}

return yield* handler(payload).pipe(
Effect.mapError((error) => CodexError.normalizeToRequestError(error)),
Effect.mapError((error) =>
CodexError.CodexAppServerRequestError.fromAppServerError(error, method),
),
);
});
2 changes: 1 addition & 1 deletion packages/effect-codex-app-server/src/_internal/stdio.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 }),
Expand Down
Loading
Loading