admin/infinite-canvas
1
0
Fork 0
mirror of https://github.com/basketikun/infinite-canvas.git synced 2026-08-05 00:34:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: enhance Codex agent connection handling with automatic thread recovery and improved error logging

Browse Source
This commit is contained in:
HouYunFei
2026-07-05 23:40:54 +08:00
parent 6113986637
commit 3080d6da10
9 changed files with 154 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
canvas-agent/src/agents.ts
+27 -9
View File
@@ -35,8 +35,16 @@ async function runCodexTurnNow(prompt: string, emit: AgentEmit, attachments: Age
try {
files = await writeAttachmentFiles(attachments);
codexApp ||= await CodexAppClient.start(emit);
const threadId = await ensureCodexThread(codexApp, options);
await codexApp.startTurn(threadId, prompt, files);
let threadId = await ensureCodexThread(codexApp, options, emit);
try {
await codexApp.startTurn(threadId, prompt, files);
} catch (error) {
if (!isRecoverableThreadError(error)) throw error;
emit("agent_log", { text: `Codex thread unavailable, starting a new thread: ${errorMessage(error)}` });
codexThreadId = "";
threadId = await ensureCodexThread(codexApp, { cwd: options.cwd }, emit);
await codexApp.startTurn(threadId, prompt, files);
}
} catch (error) {
emit("agent_error", { message: errorMessage(error) });
} finally {
@@ -96,14 +104,20 @@ export function runClaudeTurn(prompt: string, emit: AgentEmit) {
pipeJsonLines(child, emit, "claude");
}
async function ensureCodexThread(app: CodexAppClient, options: CodexRunOptions) {
async function ensureCodexThread(app: CodexAppClient, options: CodexRunOptions, emit: AgentEmit) {
if (options.threadId) {
const result = await app.readThread(options.threadId, false);
assertThreadWorkspace(field(result, "thread") || {}, options.cwd);
const thread = await app.resumeThread(options.threadId, options.cwd);
assertThreadWorkspace(thread, options.cwd);
codexThreadId = String(field(thread, "id") || options.threadId);
return codexThreadId;
if (options.threadId === codexThreadId) return codexThreadId;
try {
const result = await app.readThread(options.threadId, false);
assertThreadWorkspace(field(result, "thread") || {}, options.cwd);
const thread = await app.resumeThread(options.threadId, options.cwd);
assertThreadWorkspace(thread, options.cwd);
codexThreadId = String(field(thread, "id") || options.threadId);
return codexThreadId;
} catch (error) {
if (!isRecoverableThreadError(error)) throw error;
emit("agent_log", { text: `Codex thread unavailable, starting a new thread: ${errorMessage(error)}` });
}
}
if (!codexThreadId) {
const thread = await app.startThread(options.cwd);
@@ -112,6 +126,10 @@ async function ensureCodexThread(app: CodexAppClient, options: CodexRunOptions)
return codexThreadId;
}
function isRecoverableThreadError(error: unknown) {
return /thread not loaded|no rollout found/i.test(errorMessage(error));
}
class CodexAppClient {
private nextId = 1;
private buffer = "";
canvas-agent/src/canvas-session.ts
+3 -2
View File
@@ -18,13 +18,14 @@ export class CanvasSession {
openEvents(url: URL, res: ServerResponse) {
const clientId = url.searchParams.get("clientId") || crypto.randomUUID();
const statusOnly = url.searchParams.get("role") === "status";
res.writeHead(200, { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", Connection: "keep-alive" });
this.clients.set(clientId, res);
if (!statusOnly) this.clients.set(clientId, res);
sendEvent(res, "hello", { ok: true, clientId });
const timer = setInterval(() => sendEvent(res, "ping", { time: Date.now() }), 15000);
res.on("close", () => {
clearInterval(timer);
this.clients.delete(clientId);
if (!statusOnly) this.clients.delete(clientId);
if (this.canvasState?.clientId === clientId) this.canvasState = null;
});
}