Skip to content

fix(server): fork Codex threads with active writers #7414

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
f-trycua wants to merge 1 commit into from
Closed

fix(server): fork Codex threads with active writers #7414

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 84 additions & 3 deletions apps/server/src/provider/Layers/CodexSessionRuntime.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import {
buildTurnStartParams,
hasConfiguredMcpServer,
isRecoverableThreadResumeError,
isThreadActiveWriterError,
makeMemoryConsolidationNotificationFilter,
openCodexThread,
} from "./CodexSessionRuntime.ts";
Expand Down Expand Up @@ -570,13 +571,93 @@ describe("isRecoverableThreadResumeError", () => {
});
});

describe("isThreadActiveWriterError", () => {
it("matches Codex thread writer conflicts", () => {
NodeAssert.equal(
isThreadActiveWriterError(
new CodexErrors.CodexAppServerRequestError({
code: -32603,
errorMessage: "thread provider-thread already has an active writer",
}),
),
true,
);
});

it("ignores unrelated writer errors", () => {
NodeAssert.equal(
isThreadActiveWriterError(
new CodexErrors.CodexAppServerRequestError({
code: -32603,
errorMessage: "log file already has an active writer",
}),
),
false,
);
});
});

describe("openCodexThread", () => {
it.effect("forks the thread when another Codex process owns its writer", () =>
Effect.gen(function* () {
const calls: Array<{
method: "thread/start" | "thread/resume" | "thread/fork";
payload: unknown;
}> = [];
const forked = makeThreadOpenResponse("forked-thread");
const client = {
request: <M extends "thread/start" | "thread/resume" | "thread/fork">(
method: M,
payload: CodexRpc.ClientRequestParamsByMethod[M],
) => {
calls.push({ method, payload });
if (method === "thread/resume") {
return Effect.fail(
new CodexErrors.CodexAppServerRequestError({
code: -32603,
errorMessage: "thread provider-thread already has an active writer",
}),
);
}
return Effect.succeed(forked as CodexRpc.ClientRequestResponsesByMethod[M]);
},
};

const opened = yield* openCodexThread({
client,
threadId: ThreadId.make("thread-1"),
runtimeMode: "full-access",
cwd: "/tmp/project",
requestedModel: "gpt-5.3-codex",
serviceTier: undefined,
resumeThreadId: "provider-thread",
});

NodeAssert.equal(opened.thread.id, "forked-thread");
NodeAssert.deepStrictEqual(
calls.map((call) => call.method),
["thread/resume", "thread/fork"],
);
NodeAssert.deepStrictEqual(calls[1]?.payload, {
threadId: "provider-thread",
cwd: "/tmp/project",
approvalPolicy: "never",
sandbox: "danger-full-access",
approvalsReviewer: "user",
model: "gpt-5.3-codex",
});
}),
);

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 calls: Array<{
method: "thread/start" | "thread/resume" | "thread/fork";
payload: unknown;
}> = [];
const started = makeThreadOpenResponse("fresh-thread");
const client = {
request: <M extends "thread/start" | "thread/resume">(
request: <M extends "thread/start" | "thread/resume" | "thread/fork">(
method: M,
payload: CodexRpc.ClientRequestParamsByMethod[M],
) => {
Expand Down Expand Up @@ -614,7 +695,7 @@ describe("openCodexThread", () => {
it.effect("propagates non-recoverable resume failures", () =>
Effect.gen(function* () {
const client = {
request: <M extends "thread/start" | "thread/resume">(
request: <M extends "thread/start" | "thread/resume" | "thread/fork">(
method: M,
_payload: CodexRpc.ClientRequestParamsByMethod[M],
) => {
Expand Down
33 changes: 31 additions & 2 deletions apps/server/src/provider/Layers/CodexSessionRuntime.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -447,11 +447,17 @@ export function isRecoverableThreadResumeError(error: unknown): boolean {
return RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS.some((snippet) => message.includes(snippet));
}

export function isThreadActiveWriterError(error: unknown): boolean {
const message = (error instanceof Error ? error.message : String(error)).toLowerCase();
return message.includes("thread") && message.includes("already has an active writer");
}

type CodexThreadOpenResponse =
| CodexRpc.ClientRequestResponsesByMethod["thread/start"]
| CodexRpc.ClientRequestResponsesByMethod["thread/resume"];
| CodexRpc.ClientRequestResponsesByMethod["thread/resume"]
| CodexRpc.ClientRequestResponsesByMethod["thread/fork"];

type CodexThreadOpenMethod = "thread/start" | "thread/resume";
type CodexThreadOpenMethod = "thread/start" | "thread/resume" | "thread/fork";

interface CodexThreadOpenClient {
readonly request: <M extends CodexThreadOpenMethod>(
Expand Down Expand Up @@ -487,6 +493,29 @@ export const openCodexThread = (input: {
...startParams,
})
.pipe(
Effect.catchIf(isThreadActiveWriterError, (error) =>
Effect.logWarning("codex app-server thread resume forked from active writer", {
threadId: input.threadId,
requestedRuntimeMode: input.runtimeMode,
resumeThreadId,
cause: error,
}).pipe(
Effect.andThen(
Effect.suspend(() => {
const config = runtimeModeToThreadConfig(input.runtimeMode);
return input.client.request("thread/fork", {
threadId: resumeThreadId,
cwd: input.cwd,
approvalPolicy: config.approvalPolicy,
approvalsReviewer: config.approvalsReviewer,
sandbox: config.sandbox,
...(input.requestedModel ? { model: input.requestedModel } : {}),
...(input.serviceTier ? { serviceTier: input.serviceTier } : {}),
});
}),
),
),
),
Effect.catchIf(isRecoverableThreadResumeError, (error) =>
Effect.logWarning("codex app-server thread resume fell back to fresh start", {
threadId: input.threadId,
Expand Down
Loading