mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(code-queue): add workdirs contract checks

Browse Source
This commit is contained in:
Codex
2026-05-19 02:07:41 +00:00
parent 49cf1e57bb
commit 6006017393
9 changed files with 611 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue-mgr/src-rs/main.rs
+265 -1
View File
@@ -18,6 +18,7 @@ const MAX_BODY_BYTES: usize = 512 * 1024;
const MAX_LIST_LIMIT: i64 = 200;
const MAX_OUTPUT_PAGE_LIMIT: usize = 500;
const MAX_PREVIEW_CHARS: usize = 6000;
const WORKDIR_MAX_CHARS: usize = 512;
#[derive(Clone)]
struct Config {
@@ -347,6 +348,14 @@ fn ensure_schema(state: &AppState) -> Result<(), String> {
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL
);
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)
);
ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS queue_id TEXT NOT NULL DEFAULT 'default';
ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS provider_id TEXT NOT NULL DEFAULT 'main-server';
ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default';
@@ -355,6 +364,7 @@ fn ensure_schema(state: &AppState) -> Result<(), String> {
ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS reference_injection JSONB;
ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS read_at TIMESTAMPTZ;
ALTER TABLE unidesk_code_queue_queues ADD COLUMN IF NOT EXISTS name TEXT NOT NULL DEFAULT '';
ALTER TABLE unidesk_code_queue_workdirs ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default';
CREATE INDEX IF NOT EXISTS idx_code_queue_tasks_list ON unidesk_code_queue_tasks (status, updated_at DESC, id DESC);
CREATE INDEX IF NOT EXISTS idx_code_queue_tasks_queue_updated ON unidesk_code_queue_tasks (queue_id, updated_at DESC, id DESC);
",
@@ -370,6 +380,7 @@ fn ensure_schema(state: &AppState) -> Result<(), String> {
&[&DEFAULT_QUEUE_ID.to_string(), &DEFAULT_QUEUE_ID.to_string(), &now],
)
.map_err(|error| error_detail(&error))?;
upsert_default_workdir(&mut client, state, &now)?;
Ok(())
}
@@ -1093,6 +1104,185 @@ fn update_queue(state: &AppState, queue_id_value: &str, request: &mut Request) -
))
}
fn default_workdir_record(state: &AppState, timestamp: &str) -> Result<Value, String> {
let provider_id = normalize_workdir_provider_id(state, &state.config.default_provider_id);
let path = normalize_workdir_path(state, &provider_id, &state.config.default_workdir)?;
Ok(json!({
"providerId": provider_id,
"executionMode": "default",
"path": path,
"createdAt": timestamp,
"updatedAt": timestamp
}))
}
fn workdir_key(record: &Value) -> String {
format!(
"{}\u{0}{}\u{0}{}",
record.get("providerId").and_then(Value::as_str).unwrap_or(""),
record.get("executionMode").and_then(Value::as_str).unwrap_or("default"),
record.get("path").and_then(Value::as_str).unwrap_or("")
)
}
fn sorted_workdir_records(records: Vec<Value>) -> Vec<Value> {
let mut keyed = std::collections::BTreeMap::<String, Value>::new();
for record in records {
keyed.insert(workdir_key(&record), record);
}
keyed.into_values().collect()
}
fn upsert_default_workdir(client: &mut Client, state: &AppState, timestamp: &str) -> Result<(), String> {
let record = default_workdir_record(state, timestamp)?;
let provider_id = record.get("providerId").and_then(Value::as_str).unwrap_or("").to_string();
let execution_mode = record.get("executionMode").and_then(Value::as_str).unwrap_or("default").to_string();
let path = record.get("path").and_then(Value::as_str).unwrap_or("").to_string();
client
.execute(
"
INSERT INTO unidesk_code_queue_workdirs (provider_id, execution_mode, path, created_at, updated_at)
VALUES ($1, $2, $3, $4::text::timestamptz, $4::text::timestamptz)
ON CONFLICT (provider_id, execution_mode, path) DO UPDATE SET updated_at = EXCLUDED.updated_at
",
&[&provider_id, &execution_mode, &path, &timestamp],
)
.map_err(|error| error_detail(&error))?;
Ok(())
}
fn load_workdirs(state: &AppState) -> Result<Vec<Value>, String> {
let mut client = db_client(state, "unidesk-code-queue-mgr-rust-workdirs")?;
let rows = client
.query(
"
SELECT provider_id, execution_mode, path, created_at::text, updated_at::text
FROM unidesk_code_queue_workdirs
ORDER BY provider_id ASC, execution_mode ASC, path ASC
",
&[],
)
.map_err(|error| error_detail(&error))?;
let mut records = Vec::new();
if let Ok(record) = default_workdir_record(state, &now_iso()) {
records.push(record);
}
for row in rows {
let provider_id = normalize_workdir_provider_id(state, row.get::<_, &str>(0));
let execution_mode_raw: &str = row.get(1);
let execution_mode = match execution_mode_raw {
"windows-native" => "windows-native",
_ => "default",
};
let path_raw: &str = row.get(2);
let path = normalize_workdir_path(state, &provider_id, path_raw)?;
records.push(json!({
"providerId": provider_id,
"executionMode": execution_mode,
"path": path,
"createdAt": row.get::<_, String>(3),
"updatedAt": row.get::<_, String>(4)
}));
}
Ok(sorted_workdir_records(records))
}
fn list_workdirs(state: &AppState, url: &RequestUrl) -> Result<(Value, u16), (Value, u16)> {
let provider_filter = query_param(url, "providerId")
.filter(|value| !value.trim().is_empty())
.map(|value| normalize_workdir_provider_id(state, &value));
let mode_filter = query_param(url, "executionMode")
.filter(|value| !value.trim().is_empty())
.map(|value| normalize_execution_mode(Some(&json!(value))));
let records = load_workdirs(state).map_err(|error| (json!({ "ok": false, "error": error }), 500))?;
let workdirs = records
.into_iter()
.filter(|record| {
if let Some(provider_id) = &provider_filter {
if record.get("providerId").and_then(Value::as_str) != Some(provider_id.as_str()) {
return false;
}
}
if let Some(mode) = &mode_filter {
if record.get("executionMode").and_then(Value::as_str) != Some(mode.as_str()) {
return false;
}
}
true
})
.collect::<Vec<_>>();
Ok((json!({
"ok": true,
"workdirs": workdirs,
"defaultProviderId": state.config.default_provider_id,
"defaultWorkdir": state.config.default_workdir,
"remoteDefaultWorkdir": state.config.remote_default_workdir,
"source": "code-queue-mgr-rust-postgres"
}), 200))
}
fn create_workdir(state: &AppState, request: &mut Request) -> Result<(Value, u16), (Value, u16)> {
let body = read_json_body(request)?;
let provider_id = normalize_workdir_provider_id(state, body.get("providerId").and_then(Value::as_str).unwrap_or(""));
let execution_mode = normalize_execution_mode(body.get("executionMode"));
let raw_path = body
.get("path")
.or_else(|| body.get("cwd"))
.or_else(|| body.get("workdir"))
.and_then(Value::as_str)
.unwrap_or("");
let path = normalize_workdir_path(state, &provider_id, raw_path).map_err(|error| (json!({ "ok": false, "error": error }), 400))?;
let now = now_iso();
let mut client = db_client(state, "unidesk-code-queue-mgr-rust-workdir-create").map_err(|error| (json!({ "ok": false, "error": error }), 500))?;
client
.execute(
"
INSERT INTO unidesk_code_queue_workdirs (provider_id, execution_mode, path, created_at, updated_at)
VALUES ($1, $2, $3, $4::text::timestamptz, $4::text::timestamptz)
ON CONFLICT (provider_id, execution_mode, path) DO UPDATE SET updated_at = EXCLUDED.updated_at
",
&[&provider_id, &execution_mode, &path, &now],
)
.map_err(|error| (json!({ "ok": false, "error": error_detail(&error) }), 500))?;
let workdir = json!({ "providerId": provider_id, "executionMode": execution_mode, "path": path, "createdAt": now, "updatedAt": now });
let workdirs = load_workdirs(state).map_err(|error| (json!({ "ok": false, "error": error }), 500))?;
Ok((json!({
"ok": true,
"workdir": workdir,
"workdirs": workdirs,
"ensure": { "ok": true, "skipped": true, "reason": "code-queue-mgr records workdir menu options only; provider directories are created by the execution plane" }
}), 201))
}
fn delete_workdir(state: &AppState, provider_id_value: &str, execution_mode_value: &str, path_value: &str) -> Result<(Value, u16), (Value, u16)> {
let provider_id = normalize_workdir_provider_id(state, provider_id_value);
let execution_mode = normalize_execution_mode(Some(&json!(execution_mode_value)));
let path = normalize_workdir_path(state, &provider_id, path_value).map_err(|error| (json!({ "ok": false, "error": error }), 400))?;
let mut client = db_client(state, "unidesk-code-queue-mgr-rust-workdir-delete").map_err(|error| (json!({ "ok": false, "error": error }), 500))?;
let rows = client
.query(
"
DELETE FROM unidesk_code_queue_workdirs
WHERE provider_id = $1 AND execution_mode = $2 AND path = $3
RETURNING provider_id, execution_mode, path, created_at::text, updated_at::text
",
&[&provider_id, &execution_mode, &path],
)
.map_err(|error| (json!({ "ok": false, "error": error_detail(&error) }), 500))?;
let Some(row) = rows.first() else {
return Ok((json!({ "ok": false, "error": "workdir not found" }), 404));
};
let deleted = json!({
"providerId": row.get::<_, String>(0),
"executionMode": row.get::<_, String>(1),
"path": row.get::<_, String>(2),
"createdAt": row.get::<_, String>(3),
"updatedAt": row.get::<_, String>(4)
});
let workdirs = load_workdirs(state).map_err(|error| (json!({ "ok": false, "error": error }), 500))?;
Ok((json!({ "ok": true, "deleted": deleted, "workdirs": workdirs }), 200))
}
fn normalize_provider_id(state: &AppState, value: Option<&Value>) -> String {
value
.and_then(Value::as_str)
@@ -1102,6 +1292,20 @@ fn normalize_provider_id(state: &AppState, value: Option<&Value>) -> String {
.to_string()
}
fn normalize_workdir_provider_id(state: &AppState, value: &str) -> String {
let text = value.trim();
if !text.is_empty()
&& text.len() <= 64
&& text
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '.' | '-'))
{
text.to_string()
} else {
state.config.default_provider_id.clone()
}
}
fn normalize_cwd(state: &AppState, provider_id: &str, value: Option<&Value>) -> String {
value
.and_then(Value::as_str)
@@ -1111,6 +1315,34 @@ fn normalize_cwd(state: &AppState, provider_id: &str, value: Option<&Value>) ->
.unwrap_or_else(|| if provider_id == "main-server" { state.config.default_workdir.clone() } else { state.config.remote_default_workdir.clone() })
}
fn default_workdir_for_provider(state: &AppState, provider_id: &str) -> String {
if provider_id == state.config.default_provider_id {
state.config.default_workdir.clone()
} else {
state.config.remote_default_workdir.clone()
}
}
fn normalize_workdir_path(state: &AppState, provider_id: &str, value: &str) -> Result<String, String> {
let raw = value.trim();
if raw.is_empty() {
return Err("workdir path is required".to_string());
}
if raw.chars().count() > WORKDIR_MAX_CHARS {
return Err(format!("workdir path must be {WORKDIR_MAX_CHARS} characters or fewer"));
}
if raw.contains('\0') {
return Err("workdir path contains an invalid character".to_string());
}
let path = if raw.starts_with('/') {
raw.to_string()
} else {
format!("{}/{}", default_workdir_for_provider(state, provider_id).trim_end_matches('/'), raw)
};
let trimmed = path.trim_end_matches('/').to_string();
Ok(if trimmed.is_empty() { "/".to_string() } else { trimmed })
}
fn normalize_execution_mode(value: Option<&Value>) -> String {
match value.and_then(Value::as_str).unwrap_or("").trim().to_ascii_lowercase().as_str() {
"windows-native" | "windows" | "win32" | "native-windows" => "windows-native".to_string(),
@@ -1924,7 +2156,7 @@ fn health(state: &AppState) -> (Value, u16) {
"readPath": "column-first"
},
"endpoints": {
"control": ["/api/queues", "/api/tasks", "/api/tasks/:id/(move|retry|read|edit)", "/api/queue-claim-move/self-test"],
"control": ["/api/queues", "/api/tasks", "/api/tasks/:id/(move|retry|read|edit)", "/api/workdirs", "/api/queue-claim-move/self-test"],
"traceRead": ["/api/tasks/overview", "/api/tasks/:id/summary", "/api/tasks/:id/trace-summary", "/api/tasks/:id/trace-steps", "/api/tasks/:id/output"]
}
});
@@ -2015,6 +2247,29 @@ fn handle_request(state: &AppState, request: &mut Request) -> (Value, u16) {
if path == "/api/queue-claim-move/self-test" && (method == Method::Get || method == Method::Post) {
return (json!({ "ok": true, "runtime": "rust", "check": "route-present", "claimMoveSafety": "conditional update requires unclaimed queued/retry_wait task" }), 200);
}
if path == "/api/workdirs" && method == Method::Get {
return list_workdirs(state, &url).unwrap_or_else(|response| response);
}
if path == "/api/workdirs" && method == Method::Post {
return create_workdir(state, request).unwrap_or_else(|response| response);
}
if method == Method::Delete {
if let Some(rest) = path.strip_prefix("/api/workdirs/") {
let mut parts = rest.splitn(3, '/');
let provider_id = parts.next().unwrap_or_default();
let execution_mode = parts.next().unwrap_or_default();
let path_value = parts.next().unwrap_or_default();
if !provider_id.is_empty() && !execution_mode.is_empty() && !path_value.is_empty() {
return delete_workdir(
state,
&percent_decode(provider_id),
&percent_decode(execution_mode),
&percent_decode(path_value),
)
.unwrap_or_else(|response| response);
}
}
}
if let Some(response) = route_active_control(path, &method) {
return response;
}
@@ -2344,4 +2599,13 @@ mod tests {
assert!(preview.get("text").and_then(Value::as_str).unwrap_or("").contains("...<truncated>"));
assert!(!preview.get("text").and_then(Value::as_str).unwrap_or("").contains(marker));
}
#[test]
fn default_workdir_contract_includes_workspace() {
let state = test_state();
let record = default_workdir_record(&state, "2026-01-01T00:00:00.000Z").expect("default workdir");
assert_eq!(record.get("providerId").and_then(Value::as_str), Some("D601"));
assert_eq!(record.get("executionMode").and_then(Value::as_str), Some("default"));
assert_eq!(record.get("path").and_then(Value::as_str), Some("/workspace"));
}
}
src/components/microservices/code-queue-mgr/src/index.ts
+157 -2
View File
@@ -40,6 +40,14 @@ interface QueueRecord {
updatedAt: string;
}
interface WorkdirRecord {
providerId: string;
executionMode: CodeExecutionMode;
path: string;
createdAt: string;
updatedAt: string;
}
interface DailyTaskStatsBucket {
date: string;
executedTasks: number;
@@ -275,6 +283,7 @@ const codeQueueEnvironmentHint = [
"如果当前 Code Queue Docker 容器缺少完成任务所需的环境、系统包或语言依赖，可以先在容器内临时安装以推进当前任务；同时必须把该依赖补到 `src/components/microservices/code-queue/Dockerfile`，让后续任务重建镜像后可直接使用。",
].join("\n");
const maxTaskAttempts = 99;
const workdirMaxLength = 512;
const defaultCodeModels = ["gpt-5.5", "gpt-5.4-mini", "gpt-5.4", "minimax-m2.7"];
const codeExecutionModes: CodeExecutionMode[] = ["default", "windows-native"];
const codexStatsTimeZone = "Asia/Shanghai";
@@ -523,10 +532,140 @@ function normalizeProviderId(value: unknown): string {
return text.length > 0 ? text : config.defaultProviderId;
}
function normalizeWorkdirProviderId(value: unknown): string {
const text = typeof value === "string" ? value.trim() : "";
return /^[A-Za-z0-9_.-]{1,64}$/u.test(text) ? text : config.defaultProviderId;
}
function normalizeCwd(providerId: string, value: unknown): string {
const text = typeof value === "string" && value.trim().length > 0 ? value.trim() : "";
if (text.length > 0) return text;
return providerId === "main-server" ? config.defaultWorkdir : config.remoteDefaultWorkdir;
return providerId === config.defaultProviderId ? config.defaultWorkdir : config.remoteDefaultWorkdir;
}
function defaultWorkdirForProvider(providerId: string): string {
return providerId === config.defaultProviderId ? config.defaultWorkdir : config.remoteDefaultWorkdir;
}
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");
if (raw.startsWith("/")) return raw.replace(/\/+$/u, "") || "/";
return `${defaultWorkdirForProvider(providerId).replace(/\/+$/u, "")}/${raw}`.replace(/\/+$/u, "") || "/";
}
function defaultWorkdirRecords(timestamp = nowIso()): WorkdirRecord[] {
const providerId = normalizeWorkdirProviderId(config.defaultProviderId);
return [{
providerId,
executionMode: "default",
path: normalizeWorkdirPath(config.defaultWorkdir, providerId),
createdAt: timestamp,
updatedAt: timestamp,
}];
}
function workdirRecordKey(record: Pick<WorkdirRecord, "providerId" | "executionMode" | "path">): string {
return `${record.providerId}\u0000${record.executionMode}\u0000${record.path}`;
}
function mergeWorkdirRecords(records: WorkdirRecord[]): WorkdirRecord[] {
const byKey = new Map<string, WorkdirRecord>();
for (const record of records) {
byKey.set(workdirRecordKey(record), record);
}
return Array.from(byKey.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);
});
}
async function loadWorkdirRecords(): Promise<WorkdirRecord[]> {
const rows = await mgrSql<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
`;
const records = rows.map((row) => {
const providerId = normalizeWorkdirProviderId(row.provider_id);
return {
providerId,
executionMode: normalizeExecutionMode(row.execution_mode),
path: normalizeWorkdirPath(row.path, providerId),
createdAt: timestampToIso(row.created_at) ?? nowIso(),
updatedAt: timestampToIso(row.updated_at) ?? nowIso(),
};
});
return mergeWorkdirRecords([...defaultWorkdirRecords(), ...records]);
}
async function upsertWorkdir(record: WorkdirRecord): Promise<void> {
await mgrSql`
INSERT INTO unidesk_code_queue_workdirs (provider_id, execution_mode, path, created_at, updated_at)
VALUES (${record.providerId}, ${record.executionMode}, ${record.path}, ${record.createdAt}, ${record.updatedAt})
ON CONFLICT (provider_id, execution_mode, path) DO UPDATE SET
updated_at = EXCLUDED.updated_at
`;
}
function filteredWorkdirRecords(records: WorkdirRecord[], url: URL): WorkdirRecord[] {
const providerIdParam = url.searchParams.get("providerId");
const executionModeParam = url.searchParams.get("executionMode");
const providerId = providerIdParam === null || providerIdParam.trim().length === 0 ? null : normalizeWorkdirProviderId(providerIdParam);
const executionMode = executionModeParam === null || executionModeParam.trim().length === 0 ? null : normalizeExecutionMode(executionModeParam);
return records.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> {
const records = await loadWorkdirRecords();
return jsonResponse({
ok: true,
workdirs: filteredWorkdirRecords(records, url),
defaultProviderId: config.defaultProviderId,
defaultWorkdir: config.defaultWorkdir,
remoteDefaultWorkdir: config.remoteDefaultWorkdir,
source: "code-queue-mgr-postgres",
});
}
async function createWorkdir(req: Request): Promise<Response> {
const body = asRecord(await readJson(req)) ?? {};
const providerId = normalizeWorkdirProviderId(body.providerId);
const executionMode = normalizeExecutionMode(body.executionMode);
const path = normalizeWorkdirPath(body.path ?? body.cwd ?? body.workdir, providerId);
const now = nowIso();
const workdir: WorkdirRecord = { providerId, executionMode, path, createdAt: now, updatedAt: now };
await upsertWorkdir(workdir);
return jsonResponse({
ok: true,
workdir,
workdirs: await loadWorkdirRecords(),
ensure: { ok: true, skipped: true, reason: "code-queue-mgr records workdir menu options only; provider directories are created by the execution plane" },
}, 201);
}
async function deleteWorkdir(providerIdValue: string, executionModeValue: string, pathValue: string): Promise<Response> {
const providerId = normalizeWorkdirProviderId(providerIdValue);
const executionMode = normalizeExecutionMode(executionModeValue);
const path = normalizeWorkdirPath(pathValue, providerId);
const rows = await mgrSql<WorkdirRecord[]>`
DELETE FROM unidesk_code_queue_workdirs
WHERE provider_id = ${providerId}
AND execution_mode = ${executionMode}
AND path = ${path}
RETURNING provider_id AS "providerId", execution_mode AS "executionMode", path, created_at AS "createdAt", updated_at AS "updatedAt"
`;
if (rows.length === 0) return jsonResponse({ ok: false, error: "workdir not found" }, 404);
return jsonResponse({ ok: true, deleted: rows[0], workdirs: await loadWorkdirRecords() });
}
function normalizeStringArray(value: unknown): string[] {
@@ -1145,6 +1284,16 @@ async function ensureSchema(): Promise<void> {
updated_at TIMESTAMPTZ NOT NULL
)
`;
await mgrSql`
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 mgrSql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS queue_id TEXT NOT NULL DEFAULT 'default'`;
await mgrSql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS provider_id TEXT NOT NULL DEFAULT 'main-server'`;
await mgrSql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`;
@@ -1153,8 +1302,10 @@ async function ensureSchema(): Promise<void> {
await mgrSql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS reference_injection JSONB`;
await mgrSql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS read_at TIMESTAMPTZ`;
await mgrSql`ALTER TABLE unidesk_code_queue_queues ADD COLUMN IF NOT EXISTS name TEXT NOT NULL DEFAULT ''`;
await mgrSql`ALTER TABLE unidesk_code_queue_workdirs ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`;
const now = nowIso();
await upsertQueue({ id: config.defaultQueueId, name: config.defaultQueueId, createdAt: now, updatedAt: now });
await Promise.all(defaultWorkdirRecords(now).map((record) => upsertWorkdir(record)));
schemaReady = true;
schemaLastError = null;
}
@@ -2639,7 +2790,7 @@ async function route(req: Request): Promise<Response> {
noDockerSocket: true,
},
endpoints: {
control: ["/api/queues", "/api/tasks", "/api/tasks/:id/(move|retry|read|edit)", "/api/queue-claim-move/self-test"],
control: ["/api/queues", "/api/tasks", "/api/tasks/:id/(move|retry|read|edit)", "/api/workdirs", "/api/queue-claim-move/self-test"],
traceRead: ["/api/tasks/overview", "/api/tasks/:id/summary", "/api/tasks/:id/trace-summary", "/api/tasks/:id/trace-steps", "/api/tasks/:id/output"],
},
}, schemaReady ? 200 : 503);
@@ -2649,6 +2800,10 @@ async function route(req: Request): Promise<Response> {
if (!schemaReady) return jsonResponse({ ok: false, error: "code-queue-mgr database schema is not ready", schemaLastError }, 503);
return jsonResponse(await runQueueClaimMoveSelfTest(), 200);
}
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] ?? ""));
const activeControl = routeActiveControl(url.pathname, req.method);
if (activeControl !== null) return jsonResponse(activeControl.body, activeControl.status);
if (url.pathname === "/api/queues" && req.method === "GET") {