mirror of
https://github.com/nexu-io/open-design.git
synced 2026-10-27 09:51:12 +01:00
fix: cancel timed-out batch runs
Generated-By: looper 0.0.0-dev (runner=fixer, agent=opencode)
This commit is contained in:
@@ -6,6 +6,8 @@ import { fileURLToPath } from "node:url";
|
||||
|
||||
import {
|
||||
buildAssistantMessageUpdate,
|
||||
cancelTimedOutRun,
|
||||
createRunTimeoutError,
|
||||
dedupeDesignSystemIds,
|
||||
resolveDryRunDesignSystems,
|
||||
validateExplicitDesignSystemIds,
|
||||
@@ -86,6 +88,52 @@ test("buildAssistantMessageUpdate keeps the daemon run id on final assistant upd
|
||||
);
|
||||
});
|
||||
|
||||
test("cancelTimedOutRun posts to the daemon cancel endpoint for timeout errors", async () => {
|
||||
const originalFetch = globalThis.fetch;
|
||||
const calls: Array<{ url: string; init: RequestInit | undefined }> = [];
|
||||
globalThis.fetch = (async (url: string | URL | Request, init?: RequestInit) => {
|
||||
calls.push({ url: String(url), init });
|
||||
return new Response(JSON.stringify({ ok: true }), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
});
|
||||
}) as typeof fetch;
|
||||
|
||||
try {
|
||||
const canceled = await cancelTimedOutRun("http://127.0.0.1:4321", createRunTimeoutError("run-123", 5000, "running"));
|
||||
assert.equal(canceled, true);
|
||||
assert.deepEqual(calls, [
|
||||
{
|
||||
url: "http://127.0.0.1:4321/api/runs/run-123/cancel",
|
||||
init: {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({}),
|
||||
},
|
||||
},
|
||||
]);
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch;
|
||||
}
|
||||
});
|
||||
|
||||
test("cancelTimedOutRun ignores non-timeout errors", async () => {
|
||||
const originalFetch = globalThis.fetch;
|
||||
let called = false;
|
||||
globalThis.fetch = (async () => {
|
||||
called = true;
|
||||
throw new Error("should not be called");
|
||||
}) as typeof fetch;
|
||||
|
||||
try {
|
||||
const canceled = await cancelTimedOutRun("http://127.0.0.1:4321", new Error("not a timeout"));
|
||||
assert.equal(canceled, false);
|
||||
assert.equal(called, false);
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch;
|
||||
}
|
||||
});
|
||||
|
||||
test("CLI dry-run with explicit design systems succeeds without daemon discovery", () => {
|
||||
const result = spawnSync(
|
||||
process.execPath,
|
||||
|
||||
@@ -128,6 +128,12 @@ interface AssistantMessageUpdate {
|
||||
producedFiles?: unknown[];
|
||||
}
|
||||
|
||||
type RunTimeoutError = Error & {
|
||||
code: 'RUN_TIMEOUT';
|
||||
runId: string;
|
||||
lastStatus?: RunStatusResponse['status'];
|
||||
};
|
||||
|
||||
function buildRunPrompt(prompt: string, skipDiscoveryBrief: boolean): string {
|
||||
if (!skipDiscoveryBrief) return prompt;
|
||||
// The persisted skipDiscoveryBrief flag is understood by newer daemons, but
|
||||
@@ -254,6 +260,29 @@ export function buildAssistantMessageUpdate(params: {
|
||||
};
|
||||
}
|
||||
|
||||
export function createRunTimeoutError(
|
||||
runId: string,
|
||||
timeoutMs: number,
|
||||
lastStatus?: RunStatusResponse['status'],
|
||||
): RunTimeoutError {
|
||||
const error = new Error(
|
||||
`run timed out after ${timeoutMs}ms${lastStatus ? ` (last status: ${lastStatus})` : ''}`,
|
||||
) as RunTimeoutError;
|
||||
error.code = 'RUN_TIMEOUT';
|
||||
error.runId = runId;
|
||||
if (lastStatus !== undefined) error.lastStatus = lastStatus;
|
||||
return error;
|
||||
}
|
||||
|
||||
function isRunTimeoutError(error: unknown): error is RunTimeoutError {
|
||||
return (
|
||||
!!error &&
|
||||
typeof error === 'object' &&
|
||||
(error as { code?: unknown }).code === 'RUN_TIMEOUT' &&
|
||||
typeof (error as { runId?: unknown }).runId === 'string'
|
||||
);
|
||||
}
|
||||
|
||||
function parseJsonObject(value: string, label: string): Record<string, unknown> {
|
||||
const parsed = JSON.parse(value) as unknown;
|
||||
if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
|
||||
@@ -489,7 +518,18 @@ async function waitForRun(daemonUrl: string, runId: string, timeoutMs: number):
|
||||
if (terminal(last.status)) return last;
|
||||
await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS));
|
||||
}
|
||||
throw new Error(`run timed out after ${timeoutMs}ms${last ? ` (last status: ${last.status})` : ''}`);
|
||||
throw createRunTimeoutError(runId, timeoutMs, last?.status);
|
||||
}
|
||||
|
||||
export async function cancelTimedOutRun(daemonUrl: string, error: unknown): Promise<boolean> {
|
||||
if (!isRunTimeoutError(error)) return false;
|
||||
await api<{ ok: true }>(
|
||||
daemonUrl,
|
||||
'POST',
|
||||
`/api/runs/${encodeURIComponent(error.runId)}/cancel`,
|
||||
{},
|
||||
);
|
||||
return true;
|
||||
}
|
||||
|
||||
async function readRunAssistantText(daemonUrl: string, runId: string): Promise<string> {
|
||||
@@ -587,7 +627,22 @@ async function runOne(params: {
|
||||
return { designSystemId, projectId, projectName, conversationId, runId: run.runId, status: 'queued', daemonUrl };
|
||||
}
|
||||
|
||||
const finalStatus = await waitForRun(daemonUrl, run.runId, config.timeoutMs);
|
||||
let finalStatus: RunStatusResponse;
|
||||
try {
|
||||
finalStatus = await waitForRun(daemonUrl, run.runId, config.timeoutMs);
|
||||
} catch (err) {
|
||||
try {
|
||||
const canceled = await cancelTimedOutRun(daemonUrl, err);
|
||||
if (canceled) {
|
||||
process.stderr.write(`warning: canceled timed-out run ${run.runId} for ${designSystemId}\n`);
|
||||
}
|
||||
} catch (cancelErr) {
|
||||
process.stderr.write(
|
||||
`warning: failed to cancel timed-out run ${run.runId} for ${designSystemId}: ${(cancelErr as Error).message || String(cancelErr)}\n`,
|
||||
);
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
const assistantText = await readRunAssistantText(daemonUrl, run.runId).catch(() => '');
|
||||
const producedFiles = await api<ProjectFilesResponse>(daemonUrl, 'GET', `/api/projects/${projectId}/files`)
|
||||
.then((body) => body.files ?? [])
|
||||
|
||||
Reference in New Issue
Block a user