Add Code Queue workdir presets
This commit is contained in:
Codex
@@ -24,6 +24,7 @@ import type {
|
||||
RunMode,
|
||||
RuntimeConfig,
|
||||
TaskStatus,
|
||||
WorkdirRecord,
|
||||
} from "./types";
|
||||
import {
|
||||
codeAgentPortForModel,
|
||||
@@ -168,6 +169,7 @@ const retryBackoffBaseMs = 1000;
|
||||
const retryBackoffMaxMs = 10 * 60 * 1000;
|
||||
const queueIdPattern = /^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$/u;
|
||||
const queueNameMaxLength = 80;
|
||||
const workdirMaxLength = 512;
|
||||
const config = readConfig();
|
||||
|
||||
const logger = createLogger("code-queue", config.logFile);
|
||||
@@ -210,6 +212,7 @@ let databaseFlushTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
let databaseFlushInFlight = false;
|
||||
const dirtyDatabaseTaskIds = new Set<string>();
|
||||
const dirtyDatabaseQueueIds = new Set<string>();
|
||||
const workdirRecords = new Map<string, WorkdirRecord>();
|
||||
|
||||
function envString(name: string, fallback: string): string {
|
||||
const value = process.env[name];
|
||||
@@ -753,6 +756,61 @@ function safeQueueName(value: unknown, queueId: string): string {
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeWorkdirPath(value: unknown, providerId: string): string {
|
||||
const raw = typeof value === "string" ? value.trim() : "";
|
||||
if (raw.length === 0) throw new Error("workdir path is required");
|
||||
if (raw.length > workdirMaxLength) throw new Error(`workdir path must be ${workdirMaxLength} characters or fewer`);
|
||||
if (raw.includes("\u0000")) throw new Error("workdir path contains an invalid character");
|
||||
return resolveTaskCwd(providerId, raw).replace(/\/+$/u, "") || "/";
|
||||
}
|
||||
|
||||
function workdirRecordKey(providerId: string, executionMode: ReturnType<typeof normalizeCodeExecutionMode>, path: string): string {
|
||||
return `${providerId}\u0000${executionMode}\u0000${path}`;
|
||||
}
|
||||
|
||||
function sortedWorkdirRecords(): WorkdirRecord[] {
|
||||
return Array.from(workdirRecords.values()).sort((left, right) => {
|
||||
const providerDelta = left.providerId.localeCompare(right.providerId);
|
||||
if (providerDelta !== 0) return providerDelta;
|
||||
const modeDelta = left.executionMode.localeCompare(right.executionMode);
|
||||
if (modeDelta !== 0) return modeDelta;
|
||||
return left.path.localeCompare(right.path);
|
||||
});
|
||||
}
|
||||
|
||||
function rememberWorkdir(providerIdValue: unknown, executionModeValue: unknown, pathValue: unknown, timestamp = nowIso()): WorkdirRecord {
|
||||
const providerId = normalizeTaskProviderId(providerIdValue);
|
||||
const executionMode = normalizeCodeExecutionMode(executionModeValue);
|
||||
const path = normalizeWorkdirPath(pathValue, providerId);
|
||||
const key = workdirRecordKey(providerId, executionMode, path);
|
||||
const existing = workdirRecords.get(key);
|
||||
if (existing !== undefined) {
|
||||
existing.updatedAt = timestamp;
|
||||
return existing;
|
||||
}
|
||||
const record: WorkdirRecord = { providerId, executionMode, path, createdAt: timestamp, updatedAt: timestamp };
|
||||
workdirRecords.set(key, record);
|
||||
return record;
|
||||
}
|
||||
|
||||
function ensureDefaultWorkdirRecords(): void {
|
||||
const timestamp = nowIso();
|
||||
for (const provider of executionProviderOptions() as Array<Record<string, JsonValue>>) {
|
||||
const providerId = normalizeProviderId(provider.id);
|
||||
if (providerId === null) continue;
|
||||
rememberWorkdir(providerId, "default", String(provider.defaultWorkdir || defaultWorkdirForProvider(providerId)), timestamp);
|
||||
if (provider.supportsWindowsNativeCodex === true && typeof provider.windowsNativeDefaultWorkdir === "string" && provider.windowsNativeDefaultWorkdir.length > 0) {
|
||||
rememberWorkdir(providerId, "windows-native", provider.windowsNativeDefaultWorkdir, timestamp);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function ensureLocalWorkdir(path: string): { ok: boolean; created: boolean; error: string | null } {
|
||||
const existed = existsSync(path);
|
||||
mkdirSync(path, { recursive: true });
|
||||
return { ok: true, created: !existed, error: null };
|
||||
}
|
||||
|
||||
function emptyState(): PersistedState {
|
||||
const at = nowIso();
|
||||
return { version: 1, updatedAt: at, nextSeq: 1, queues: [{ id: defaultQueueId, name: defaultQueueId, createdAt: at, updatedAt: at }], tasks: [] };
|
||||
@@ -1211,6 +1269,29 @@ async function upsertQueueToDatabase(client: SqlExecutor, queue: QueueRecord): P
|
||||
`;
|
||||
}
|
||||
|
||||
async function upsertWorkdirsToDatabase(records: WorkdirRecord[]): Promise<void> {
|
||||
if (records.length === 0) return;
|
||||
for (const record of records) {
|
||||
await sql`
|
||||
INSERT INTO unidesk_code_queue_workdirs (
|
||||
provider_id,
|
||||
execution_mode,
|
||||
path,
|
||||
created_at,
|
||||
updated_at
|
||||
) VALUES (
|
||||
${record.providerId},
|
||||
${record.executionMode},
|
||||
${record.path},
|
||||
${taskTimestamp(record.createdAt) ?? nowIso()},
|
||||
${taskTimestamp(record.updatedAt) ?? nowIso()}
|
||||
)
|
||||
ON CONFLICT (provider_id, execution_mode, path) DO UPDATE SET
|
||||
updated_at = EXCLUDED.updated_at
|
||||
`;
|
||||
}
|
||||
}
|
||||
|
||||
interface DatabaseTaskRow {
|
||||
id: string;
|
||||
updated_at: Date | string;
|
||||
@@ -1595,6 +1676,16 @@ async function initDatabasePersistence(): Promise<void> {
|
||||
updated_at TIMESTAMPTZ NOT NULL
|
||||
)
|
||||
`;
|
||||
await sql`
|
||||
CREATE TABLE IF NOT EXISTS unidesk_code_queue_workdirs (
|
||||
provider_id TEXT NOT NULL,
|
||||
execution_mode TEXT NOT NULL DEFAULT 'default',
|
||||
path TEXT NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL,
|
||||
PRIMARY KEY (provider_id, execution_mode, path)
|
||||
)
|
||||
`;
|
||||
await sql`
|
||||
CREATE TABLE IF NOT EXISTS unidesk_code_queue_notifications (
|
||||
id TEXT PRIMARY KEY,
|
||||
@@ -1626,6 +1717,7 @@ async function initDatabasePersistence(): Promise<void> {
|
||||
AND (task_json->>'readAt') ~ '^\\d{4}-\\d{2}-\\d{2}T'
|
||||
`;
|
||||
await sql`ALTER TABLE unidesk_code_queue_queues ADD COLUMN IF NOT EXISTS name TEXT NOT NULL DEFAULT ''`;
|
||||
await sql`ALTER TABLE unidesk_code_queue_workdirs ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`;
|
||||
|
||||
const countRows = await sql<Array<{ count: string | number }>>`SELECT COUNT(*) AS count FROM unidesk_code_queue_tasks`;
|
||||
const hotTasks = await loadTasksFromDatabase("hot");
|
||||
@@ -1665,6 +1757,21 @@ async function initDatabasePersistence(): Promise<void> {
|
||||
}
|
||||
}
|
||||
state.queues.splice(0, state.queues.length, ...Array.from(queueMap.values()).sort((left, right) => left.id.localeCompare(right.id)));
|
||||
const workdirRows = await sql<Array<{ provider_id: string; execution_mode: string; path: string; created_at: Date | string; updated_at: Date | string }>>`
|
||||
SELECT provider_id, execution_mode, path, created_at, updated_at
|
||||
FROM unidesk_code_queue_workdirs
|
||||
ORDER BY provider_id ASC, execution_mode ASC, path ASC
|
||||
`;
|
||||
for (const row of workdirRows) {
|
||||
rememberWorkdir(row.provider_id, row.execution_mode, row.path, taskTimestamp(String(row.updated_at)) ?? nowIso());
|
||||
const record = workdirRecords.get(workdirRecordKey(normalizeTaskProviderId(row.provider_id), normalizeCodeExecutionMode(row.execution_mode), normalizeWorkdirPath(row.path, normalizeTaskProviderId(row.provider_id))));
|
||||
if (record !== undefined) {
|
||||
record.createdAt = taskTimestamp(String(row.created_at)) ?? record.createdAt;
|
||||
record.updatedAt = taskTimestamp(String(row.updated_at)) ?? record.updatedAt;
|
||||
}
|
||||
}
|
||||
ensureDefaultWorkdirRecords();
|
||||
await upsertWorkdirsToDatabase(sortedWorkdirRecords());
|
||||
databaseReady = true;
|
||||
scheduleStartupDatabaseMaintenance();
|
||||
runGarbageCollection();
|
||||
@@ -1902,6 +2009,7 @@ function createTask(request: QueueTaskRequest): QueueTask {
|
||||
const executionMode = normalizeCodeExecutionMode(request.executionMode);
|
||||
const cwd = resolveTaskCwd(providerId, request.cwd);
|
||||
validateExecutionModeForTask(providerId, cwd, model, executionMode);
|
||||
rememberWorkdir(providerId, executionMode, cwd, at);
|
||||
const queueId = normalizeQueueId(request.queueId);
|
||||
ensureQueue(queueId);
|
||||
return {
|
||||
@@ -2137,6 +2245,7 @@ configureProviderRuntime({
|
||||
config,
|
||||
safePreview,
|
||||
});
|
||||
ensureDefaultWorkdirRecords();
|
||||
|
||||
configureTaskOutput({
|
||||
config,
|
||||
@@ -3318,9 +3427,12 @@ function requestErrorResponse(error: unknown): Response | null {
|
||||
|| error.message === "a task cannot reference itself while editing prompt"
|
||||
|| error.message === "sourceQueueId is required"
|
||||
|| error.message === "source queue must be different from target queue"
|
||||
|| error.message === "workdir path is required"
|
||||
|| error.message === "workdir path contains an invalid character"
|
||||
|| error.message.startsWith("referenceTaskIds supports at most ")
|
||||
|| error.message.startsWith("queueId must match ")
|
||||
|| error.message.startsWith("queue name must be ")
|
||||
|| error.message.startsWith("workdir path must be ")
|
||||
|| error.message.startsWith("windows-native executionMode ")
|
||||
)) {
|
||||
return jsonResponse({ ok: false, error: error.message }, 400);
|
||||
@@ -3337,6 +3449,76 @@ function parseLimit(url: URL): number {
|
||||
return Number.isInteger(value) && value > 0 ? Math.min(500, value) : 100;
|
||||
}
|
||||
|
||||
function workdirRowsForResponse(providerIdValue: string | null, executionModeValue: string | null): WorkdirRecord[] {
|
||||
const providerId = normalizeProviderId(providerIdValue) ?? null;
|
||||
const executionMode = executionModeValue === null ? null : normalizeCodeExecutionMode(executionModeValue);
|
||||
return sortedWorkdirRecords().filter((record) => {
|
||||
if (providerId !== null && record.providerId !== providerId) return false;
|
||||
if (executionMode !== null && record.executionMode !== executionMode) return false;
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
async function listWorkdirs(url: URL): Promise<Response> {
|
||||
ensureDefaultWorkdirRecords();
|
||||
return jsonResponse({
|
||||
ok: true,
|
||||
workdirs: workdirRowsForResponse(url.searchParams.get("providerId"), url.searchParams.get("executionMode")),
|
||||
defaultProviderId: config.mainProviderId,
|
||||
defaultWorkdir: config.defaultWorkdir,
|
||||
remoteDefaultWorkdir: config.remoteDefaultWorkdir,
|
||||
windowsNativeCodexDefaultWorkdir: config.windowsNativeCodexDefaultWorkdir,
|
||||
});
|
||||
}
|
||||
|
||||
async function createWorkdir(req: Request): Promise<Response> {
|
||||
const body = await readJson(req);
|
||||
const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
|
||||
const providerId = normalizeTaskProviderId(record.providerId);
|
||||
const executionMode = normalizeCodeExecutionMode(record.executionMode);
|
||||
const path = normalizeWorkdirPath(record.path ?? record.cwd ?? record.workdir, providerId);
|
||||
validateExecutionModeForTask(providerId, path, config.defaultModel, executionMode);
|
||||
const previous = workdirRecords.get(workdirRecordKey(providerId, executionMode, path));
|
||||
let ensureResult: JsonValue = { ok: false, skipped: true, reason: "remote provider workdirs are created when a task starts" };
|
||||
if (providerIsMain(providerId)) {
|
||||
ensureResult = ensureLocalWorkdir(path) as unknown as JsonValue;
|
||||
} else if (record.ensure === true || record.createOnProvider === true) {
|
||||
const command = await runCodeQueueSsh(providerId, `set -euo pipefail\nmkdir -p ${shellQuote(path)}\ntest -d ${shellQuote(path)}\nprintf 'workdir_ready path=%s\\n' ${shellQuote(path)}`, 30_000, "workdir-create");
|
||||
ensureResult = {
|
||||
ok: command.exitCode === 0,
|
||||
exitCode: command.exitCode,
|
||||
stdout: safePreview(command.stdout, 800),
|
||||
stderr: safePreview(command.stderr, 800),
|
||||
durationMs: command.durationMs,
|
||||
} as unknown as JsonValue;
|
||||
if (command.exitCode !== 0) return jsonResponse({ ok: false, error: `failed to create workdir on provider ${providerId}`, ensure: ensureResult }, 502);
|
||||
}
|
||||
const workdir = rememberWorkdir(providerId, executionMode, path);
|
||||
await upsertWorkdirsToDatabase([workdir]);
|
||||
logger("info", "workdir_saved", { providerId, executionMode, path, existed: previous !== undefined, ensure: ensureResult });
|
||||
return jsonResponse({ ok: true, workdir, workdirs: sortedWorkdirRecords(), ensure: ensureResult }, previous === undefined ? 201 : 200);
|
||||
}
|
||||
|
||||
async function deleteWorkdir(providerIdValue: string, executionModeValue: string, pathValue: string): Promise<Response> {
|
||||
const providerId = normalizeTaskProviderId(providerIdValue);
|
||||
const executionMode = normalizeCodeExecutionMode(executionModeValue);
|
||||
const path = normalizeWorkdirPath(pathValue, providerId);
|
||||
const key = workdirRecordKey(providerId, executionMode, path);
|
||||
const existing = workdirRecords.get(key) ?? null;
|
||||
if (existing === null) return jsonResponse({ ok: false, error: "workdir not found" }, 404);
|
||||
workdirRecords.delete(key);
|
||||
if (databaseReady) {
|
||||
await sql`
|
||||
DELETE FROM unidesk_code_queue_workdirs
|
||||
WHERE provider_id = ${providerId}
|
||||
AND execution_mode = ${executionMode}
|
||||
AND path = ${path}
|
||||
`;
|
||||
}
|
||||
logger("info", "workdir_deleted", { providerId, executionMode, path });
|
||||
return jsonResponse({ ok: true, deleted: existing, workdirs: sortedWorkdirRecords() });
|
||||
}
|
||||
|
||||
async function createTasks(req: Request): Promise<Response> {
|
||||
const body = await readJson(req);
|
||||
const batchRecord = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
|
||||
@@ -3358,6 +3540,7 @@ async function createTasks(req: Request): Promise<Response> {
|
||||
logger("info", "tasks_enqueued", { count: tasks.length, ids: tasks.map((task) => task.id), queueIds: Array.from(new Set(tasks.map(queueIdOf))), providerIds: Array.from(new Set(tasks.map((task) => task.providerId))), executionModes: Array.from(new Set(tasks.map((task) => task.executionMode))) });
|
||||
scheduleQueue();
|
||||
await flushDirtyTasksToDatabase(true);
|
||||
await upsertWorkdirsToDatabase(sortedWorkdirRecords());
|
||||
return jsonResponse({ ok: true, tasks: tasks.map((task) => taskForResponse(task)), queue: await queueSummaryForResponse() }, 202);
|
||||
}
|
||||
|
||||
@@ -3964,6 +4147,16 @@ async function route(req: Request): Promise<Response> {
|
||||
const tasks = await loadAllTasksForRead();
|
||||
return jsonResponse({ ok: true, queues: perQueueSummaries(tasks), queue: queueSummary(false, tasks) });
|
||||
}
|
||||
if (url.pathname === "/api/workdirs" && req.method === "GET") return await listWorkdirs(url);
|
||||
if (url.pathname === "/api/workdirs" && req.method === "POST") return await createWorkdir(req);
|
||||
const workdirMatch = url.pathname.match(/^\/api\/workdirs\/([^/]+)\/([^/]+)\/(.+)$/u);
|
||||
if (workdirMatch !== null && req.method === "DELETE") {
|
||||
return await deleteWorkdir(
|
||||
decodeURIComponent(workdirMatch[1] ?? ""),
|
||||
decodeURIComponent(workdirMatch[2] ?? ""),
|
||||
decodeURIComponent(workdirMatch[3] ?? ""),
|
||||
);
|
||||
}
|
||||
if (url.pathname === "/api/queues" && req.method === "POST") return await createQueue(req);
|
||||
if (url.pathname === "/api/queues/merge" && req.method === "POST") return await mergeQueues(null, req);
|
||||
const queueMergeMatch = url.pathname.match(/^\/api\/queues\/([^/]+)\/merge$/u);
|
||||
|
||||
@@ -385,6 +385,14 @@ export interface QueueRecord {
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
export interface WorkdirRecord {
|
||||
providerId: string;
|
||||
executionMode: CodeExecutionMode;
|
||||
path: string;
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
export interface AppServerExit {
|
||||
code: number | null;
|
||||
signal: string | null;
|
||||
|
||||
Reference in New Issue
Block a user