fix: stabilize code queue runtime and trace flow

This commit is contained in:
Codex
2026-05-14 17:11:45 +00:00
parent 9f483b002c
commit cbbed004a6
30 changed files with 1655 additions and 250 deletions
File diff suppressed because one or more lines are too long
+9
View File
@@ -3181,6 +3181,15 @@ input:focus, select:focus, textarea:focus { border-color: var(--accent-2); }
color: var(--accent-2);
white-space: nowrap;
}
.codex-trace-step-inline-summary {
min-width: 80px;
flex: 1 1 180px;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
color: #a7c7c3;
font-size: 11px;
}
.codex-trace-step.error > summary .codex-output-channel {
color: var(--danger);
border-color: rgba(207, 106, 84, 0.52);
+100 -6
View File
@@ -655,6 +655,40 @@ function objectRecord(value: any): AnyRecord | null {
return value && typeof value === "object" && !Array.isArray(value) ? value : null;
}
function judgeFailureDetailsText(judge: any): string {
const raw = objectRecord(judge?.raw);
const details = objectRecord(judge?.failureDetails) || objectRecord(raw?.minimaxFailure);
if (details === null) return "";
const repairAttempt = details.repairAttempt === undefined ? "" : `${details.repairAttempt}/${details.maxRepairAttempts ?? "?"}`;
const rows = [
["provider", details.provider || "minimax"],
["stage", details.stage],
["model", details.model],
["timedOut", details.timedOut],
["durationMs", details.durationMs],
["timeoutMs", details.timeoutMs],
["promptChars", details.promptChars],
["promptLines", details.promptLines],
["payloadBytes", details.payloadBytes],
["responseStatus", details.responseStatus],
["repairAttempt", repairAttempt],
["errorName", details.errorName],
["error", details.errorMessage],
["responseContentPreview", details.responseContentPreview],
["responseTextPreview", details.responseTextPreview],
].filter(([, value]) => value !== undefined && value !== null && String(value).length > 0);
return rows.map(([key, value]) => `${key}: ${String(value)}`).join("\n");
}
function JudgeFailureDetails({ judge, testId = "codex-judge-failure-details" }: AnyRecord) {
const text = judgeFailureDetailsText(judge);
if (text.length === 0) return null;
return h("details", { className: "codex-judge-failure-details", "data-testid": testId },
h("summary", null, "MiniMax failure details"),
h("pre", null, text),
);
}
function taskProgressiveAttempts(task: any): any[] {
const summaryAttempts = taskTraceSummary(task)?.attempts;
if (Array.isArray(summaryAttempts) && summaryAttempts.length > 0) return summaryAttempts;
@@ -1214,6 +1248,8 @@ function codexProviderOptions(queue: any, currentProviderId: string): any[] {
id: String(item?.id || "").trim(),
label: String(item?.label || item?.id || "").trim(),
defaultWorkdir: String(item?.defaultWorkdir || "").trim(),
supportsWindowsNativeCodex: item?.supportsWindowsNativeCodex === true,
windowsNativeDefaultWorkdir: String(item?.windowsNativeDefaultWorkdir || "").trim(),
kind: String(item?.kind || "").trim(),
}))
.filter((item: any) => item.id.length > 0);
@@ -1221,14 +1257,44 @@ function codexProviderOptions(queue: any, currentProviderId: string): any[] {
const byId = new Map<string, any>();
for (const item of [
...rows,
{ id: fallbackMain, label: `${fallbackMain} (master)`, defaultWorkdir: String(queue?.defaultWorkdir || "/root/unidesk"), kind: "local" },
currentProviderId ? { id: currentProviderId, label: currentProviderId, defaultWorkdir: providerDefaultWorkdir(queue, currentProviderId), kind: "" } : null,
{ id: fallbackMain, label: `${fallbackMain} (master)`, defaultWorkdir: String(queue?.defaultWorkdir || "/root/unidesk"), supportsWindowsNativeCodex: false, windowsNativeDefaultWorkdir: "", kind: "local" },
currentProviderId ? { id: currentProviderId, label: currentProviderId, defaultWorkdir: providerDefaultWorkdir(queue, currentProviderId), supportsWindowsNativeCodex: currentProviderId !== fallbackMain, windowsNativeDefaultWorkdir: String(queue?.windowsNativeCodexDefaultWorkdir || "/mnt/f/Work/ConStart"), kind: "" } : null,
].filter(Boolean) as any[]) {
if (!byId.has(item.id)) byId.set(item.id, item);
}
return Array.from(byId.values());
}
function codexExecutionModeOptions(queue: any, currentMode: string): any[] {
const configured = Array.isArray(queue?.executionModes) ? queue.executionModes : [];
const rows = configured
.map((item: any) => ({
id: String(item?.id || item?.kind || "").trim(),
label: String(item?.label || item?.id || item?.kind || "").trim(),
description: String(item?.description || "").trim(),
defaultWorkdir: String(item?.defaultWorkdir || "").trim(),
requiresProvider: item?.requiresProvider === true,
requiresWindowsCwd: item?.requiresWindowsCwd === true,
}))
.filter((item: any) => item.id.length > 0);
const fallback = [
{ id: "default", label: "默认容器/本机", description: "主 server 用本机 Codex;远程 Provider 用执行容器 Codex。", defaultWorkdir: "", requiresProvider: false, requiresWindowsCwd: false },
{ id: "windows-native", label: "Windows 原生 Codex", description: "启动执行容器,但容器只做 stdio relayCodex 运行在 Provider 的 Windows 宿主。", defaultWorkdir: String(queue?.windowsNativeCodexDefaultWorkdir || "/mnt/f/Work/ConStart"), requiresProvider: true, requiresWindowsCwd: true },
];
const byId = new Map<string, any>();
for (const item of [...rows, ...fallback, currentMode ? { id: currentMode, label: currentMode, description: "", defaultWorkdir: "", requiresProvider: currentMode === "windows-native", requiresWindowsCwd: currentMode === "windows-native" } : null].filter(Boolean) as any[]) {
if (!byId.has(item.id)) byId.set(item.id, item);
}
return Array.from(byId.values());
}
function executionModeDefaultWorkdir(queue: any, mode: string, providerId: string): string {
if (mode !== "windows-native") return providerDefaultWorkdir(queue, providerId);
const option = Array.isArray(queue?.executionModes) ? queue.executionModes.find((item: any) => String(item?.id || item?.kind || "") === "windows-native") : null;
const provider = Array.isArray(queue?.executionProviders) ? queue.executionProviders.find((item: any) => String(item?.id || "") === providerId) : null;
return String(provider?.windowsNativeDefaultWorkdir || option?.defaultWorkdir || queue?.windowsNativeCodexDefaultWorkdir || "/mnt/f/Work/ConStart");
}
function providerDefaultWorkdir(queue: any, providerId: string): string {
const id = String(providerId || "").trim();
const map = queue?.defaultWorkdirByProvider && typeof queue.defaultWorkdirByProvider === "object" ? queue.defaultWorkdirByProvider : {};
@@ -1310,6 +1376,7 @@ function TaskCard({ task, selected, onSelect, onCopy, onReference, onMarkRead, c
h("div", { className: "codex-task-meta" },
h("span", null, `queue=${taskQueueLabel(task)}`),
h("span", null, `provider=${task?.providerId || "main-server"}`),
h("span", null, `mode=${task?.executionMode || "default"}`),
h("span", null, task?.model || "--"),
h("span", null, taskDurationLabel(task)),
),
@@ -1622,6 +1689,7 @@ function ProgressiveExecutionSummary({ task, attempt, attemptIndex, loading, onL
const seq = String(step?.seq ?? "");
const detail = stepDetails[seq];
const summaryLines = Array.isArray(step?.summaryLines) ? step.summaryLines.slice(0, 4) : [];
const inlineSummary = summaryLines.find((line: any) => String(line || "").trim().length > 0);
return h("details", {
key: seq || `${step?.title}-${step?.at}`,
className: `codex-trace-step ${String(step?.kind || "message")} ${traceStepIsError(step) ? "error" : ""}`,
@@ -1634,6 +1702,7 @@ function ProgressiveExecutionSummary({ task, attempt, attemptIndex, loading, onL
h("span", { className: "codex-output-channel" }, traceStepKindLabel(step?.kind)),
h("strong", null, String(step?.title || "Trace step")),
step?.status ? h("code", null, String(step.status)) : null,
inlineSummary ? h("span", { className: "codex-trace-step-inline-summary", title: String(inlineSummary) }, String(inlineSummary)) : null,
h("time", null, fmtDate(step?.at)),
),
h("div", { className: "codex-trace-step-summary" },
@@ -1695,6 +1764,7 @@ function ProgressiveJudge({ task, attempt, attemptIndex, testId = "codex-progres
h(StatusBadge, { status: judge.decision }, judge.decision),
h("strong", null, `${Math.round(Number(judge.confidence || 0) * 100)}% confidence`),
h("p", { "data-testid": `${testId}-reason` }, judge.reason || "--"),
h(JudgeFailureDetails, { judge, testId: `${testId}-failure-details` }),
judge.continuePrompt ? h("pre", { "data-testid": `${testId}-continue-prompt` }, String(judge.continuePrompt || "")) : null,
),
);
@@ -1889,6 +1959,7 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
const [mergeDialogOpen, setMergeDialogOpen] = useState(false);
const [mergeSourceQueueId, setMergeSourceQueueId] = useState("");
const [providerId, setProviderId] = useState("main-server");
const [executionMode, setExecutionMode] = useState("default");
const [model, setModel] = useState("gpt-5.5");
const [cwd, setCwd] = useState("/root/unidesk");
const [maxAttempts, setMaxAttempts] = useState(99);
@@ -1967,7 +2038,8 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
const submitDisabled = submitting || busy || enqueueCount === 0 || batchNeedsConfirmation;
const codexModels = codexModelOptions(queue, model);
const providerOptions = codexProviderOptions(queue, providerId);
const currentProviderDefaultWorkdir = providerDefaultWorkdir(queue, providerId);
const executionModeRows = codexExecutionModeOptions(queue, executionMode);
const currentProviderDefaultWorkdir = executionModeDefaultWorkdir(queue, executionMode, providerId);
const selectedCanSteer = selectedTask?.id && selectedTask?.activeTurnId && String(selectedTask?.status) === "running";
const selectedCanInterrupt = selectedTask?.id && !["succeeded", "failed", "canceled"].includes(String(selectedTask?.status || ""));
const selectedCanRetry = selectedTask?.id && ["succeeded", "failed", "canceled"].includes(String(selectedTask?.status || ""));
@@ -2038,7 +2110,21 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
function changeSubmitProvider(nextProviderId: string): void {
const next = String(nextProviderId || queue?.mainProviderId || "main-server").trim() || "main-server";
setProviderId(next);
setCwd(providerDefaultWorkdir(queue, next));
setCwd(executionModeDefaultWorkdir(queue, executionMode, next));
}
function changeSubmitExecutionMode(nextMode: string): void {
const next = String(nextMode || "default").trim() || "default";
let nextProvider = providerId;
if (next === "windows-native") {
const current = providerOptions.find((provider: any) => provider.id === providerId);
if (!current?.supportsWindowsNativeCodex) {
nextProvider = String(providerOptions.find((provider: any) => provider.supportsWindowsNativeCodex)?.id || providerId || "D601");
setProviderId(nextProvider);
}
}
setExecutionMode(next);
setCwd(executionModeDefaultWorkdir(queue, next, nextProvider));
}
function patchLoadedReadState(taskIds: string[], readAt: string, queuePatch: any = null, taskPatch: any = null): void {
@@ -2872,6 +2958,7 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
prompt: text,
queueId: submitQueueId,
providerId,
executionMode,
model,
cwd,
maxAttempts: Number(maxAttempts),
@@ -3298,6 +3385,7 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
h("span", { className: `codex-trace-status-chip unread ${overallUnreadTerminalCount > 0 ? "warn" : ""}` }, h("b", null, "结束未读"), String(overallUnreadTerminalCount)),
h("span", { className: "codex-trace-status-chip service" }, h("b", null, "服务"), `${runtime.providerStatus || "unknown"} · ${service?.providerId || "main-server"} · ${backend.public ? "公网暴露" : "仅 UniDesk frontend 代理访问"}`),
h("span", { className: "codex-trace-status-chip" }, h("b", null, "执行节点"), providerOptions.map((provider: any) => provider.id).join(" / ")),
h("span", { className: "codex-trace-status-chip" }, h("b", null, "执行模式"), executionModeRows.map((mode: any) => mode.id).join(" / ")),
h("span", { className: "codex-trace-status-chip" }, h("b", null, "模型"), codexModels.join(" / ")),
h("span", { className: "codex-trace-status-chip" }, h("b", null, "加载"), loadStats?.phase === "complete" ? fmtPreciseMs(loadStats?.totalMs) : String(loadStats?.phase || "idle")),
h("span", { className: "codex-trace-status-chip" }, h("b", null, "刷新"), refreshedAt ? fmtClock(refreshedAt) : "--"),
@@ -3305,7 +3393,7 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
const sessionPanel = h(Panel, {
title: selectedTask ? `Trace ${String(selectedTask.id).slice(0, 22)}` : "Trace 输出",
eyebrow: selectedTask ? `${selectedTask.status} / view=${selectedQueueName} / task queue=${taskQueueLabel(selectedTask)} / provider=${selectedTask.providerId || "main-server"} / ${selectedTask.model} / agent loop trace` : `Agent loop trace / view=${selectedQueueName}`,
eyebrow: selectedTask ? `${selectedTask.status} / view=${selectedQueueName} / task queue=${taskQueueLabel(selectedTask)} / provider=${selectedTask.providerId || "main-server"} / mode=${selectedTask.executionMode || "default"} / ${selectedTask.model} / agent loop trace` : `Agent loop trace / view=${selectedQueueName}`,
summary: traceStatusSummary,
loading: selectedDetailLoading || loadingMoreTasks || searchLoading || searchLoadingMoreTasks || loadStats?.phase === "loading",
actions: h("div", { className: "panel-actions" },
@@ -3433,7 +3521,12 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
),
h("label", null, "执行 Provider",
h("select", { value: providerId, disabled: submitting, onChange: (event: any) => changeSubmitProvider(String(event.target.value || "main-server")), "data-testid": "codex-provider-select" },
providerOptions.map((provider: any) => h("option", { key: provider.id, value: provider.id }, `${provider.label || provider.id} · ${provider.defaultWorkdir || providerDefaultWorkdir(queue, provider.id)}`)),
providerOptions.map((provider: any) => h("option", { key: provider.id, value: provider.id }, `${provider.label || provider.id} · ${provider.defaultWorkdir || providerDefaultWorkdir(queue, provider.id)}${provider.supportsWindowsNativeCodex ? " · Windows native" : ""}`)),
),
),
h("label", null, "执行模式",
h("select", { value: executionMode, disabled: submitting, onChange: (event: any) => changeSubmitExecutionMode(String(event.target.value || "default")), "data-testid": "codex-execution-mode-select" },
executionModeRows.map((mode: any) => h("option", { key: mode.id, value: mode.id }, `${mode.label || mode.id}${mode.id === "windows-native" ? " · 宿主 Codex" : ""}`)),
),
),
h("label", null, "工作目录", h("input", { value: cwd, disabled: submitting, onChange: (event: any) => setCwd(event.target.value), placeholder: currentProviderDefaultWorkdir || queue?.defaultWorkdir || "/root/unidesk", "data-testid": "codex-cwd-input" })),
@@ -3530,6 +3623,7 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
h(StatusBadge, { status: selectedTask.lastJudge.decision }, selectedTask.lastJudge.decision),
h("strong", null, `${Math.round(Number(selectedTask.lastJudge.confidence || 0) * 100)}% confidence`),
h("p", { "data-testid": "codex-task-judge-reason" }, shortText(selectedTask.lastJudge.reason || "--", 180)),
h(JudgeFailureDetails, { judge: selectedTask.lastJudge, testId: "codex-task-judge-failure-details" }),
selectedTask.lastJudge.continuePrompt ? h("code", { "data-testid": "codex-task-judge-continue-prompt" }, shortText(selectedTask.lastJudge.continuePrompt, 160)) : null,
) : h(EmptyState, { title: "尚未判定", text: "Codex turn 结束后会由 MiniMax M2.7 或 fallback judge 判定 complete/retry/failretry 会在已有 thread 追加继续执行 prompt。" }),
),
+24 -2
View File
@@ -864,6 +864,28 @@ function renderToolGroup(item: TraceItem): any {
);
}
function traceSystemItemIsError(item: TraceItem): boolean {
const text = [
item.title,
item.status,
item.bodyPreview,
item.commandPreview,
item.stderrPreview,
item.stdoutPreview,
].map((value) => String(value || "")).join("\n");
return /\b(error|failed|failure|interrupt|interrupted|cancell?ed|watchdog|timeout|closed|refused|aborted|exception)\b/iu.test(text);
}
function filterTraceSystemEvents(items: TraceItem[], showSystemEvents: boolean): TraceItem[] {
if (showSystemEvents) return items;
return items.flatMap((item) => {
if (String(item.kind || "") === "system" && !traceSystemItemIsError(item)) return [];
if (String(item.kind || "") !== "toolGroup" || !Array.isArray(item.items)) return [item];
const visibleItems = filterTraceSystemEvents(item.items, showSystemEvents);
return [{ ...item, items: visibleItems }];
});
}
const traceAutoScrollBottomThresholdPx = 16;
function traceIsScrolledToBottom(element: HTMLElement): boolean {
@@ -871,10 +893,10 @@ function traceIsScrolledToBottom(element: HTMLElement): boolean {
return distanceToBottom <= traceAutoScrollBottomThresholdPx;
}
export function TraceView({ items, input, port, autoScroll = false, loading = false, hasDetail = true, emptyText = "等待 Trace 输出...", loadingText = "正在加载完整 Trace...", testId = "trace-output", className = "codex-transcript", keepRecentToolCalls = 3, collapseTools = true }: AnyRecord) {
export function TraceView({ items, input, port, autoScroll = false, loading = false, hasDetail = true, emptyText = "等待 Trace 输出...", loadingText = "正在加载完整 Trace...", testId = "trace-output", className = "codex-transcript", keepRecentToolCalls = 3, collapseTools = true, showSystemEvents = false }: AnyRecord) {
const ref = useRef<HTMLDivElement | null>(null);
const shouldFollowTailRef = useRef(true);
const rawTrace = coalesceEditedFileChanges(port ? traceFromPort(port, input) : normalizeTraceItems(items));
const rawTrace = filterTraceSystemEvents(coalesceEditedFileChanges(port ? traceFromPort(port, input) : normalizeTraceItems(items)), Boolean(showSystemEvents));
const trace = collapseTools ? collapseToolTraceRuns(rawTrace, keepRecentToolCalls) : rawTrace;
const maxSeq = traceMaxSeq(rawTrace);
useEffect(() => {
@@ -20,6 +20,7 @@ export interface CodexPortContext {
recordNumberField: (record: Record<string, unknown> | null, keys: string[]) => number | null;
recordStringField: (record: Record<string, unknown> | null, keys: string[]) => string;
remoteAppServerCommand: (task: QueueTask) => string;
windowsNativeAppServerCommand: (task: QueueTask) => string;
resolveReasoningEffort: (model: string, explicit?: string | null) => string | null;
safePreview: (value: string, max?: number) => string;
nowIso: () => string;
@@ -36,6 +37,18 @@ function ctx(): CodexPortContext {
return context;
}
function windowsPathFromWslMount(path: string): string {
const match = path.match(/^\/mnt\/([a-zA-Z])(?:\/(.*))?$/u);
if (match === null) return path;
const drive = String(match[1]).toUpperCase();
const rest = String(match[2] ?? "").replace(/\//gu, "\\");
return rest.length > 0 ? `${drive}:\\${rest}` : `${drive}:\\`;
}
function appServerCwdForTask(task: QueueTask): string {
return task.executionMode === "windows-native" ? windowsPathFromWslMount(task.cwd) : task.cwd;
}
class AppServerClient {
private child: ChildProcessWithoutNullStreams;
private nextId = 1;
@@ -48,13 +61,16 @@ class AppServerClient {
constructor(private readonly task: QueueTask, private readonly onNotification: (message: Record<string, unknown>) => void) {
this.closedPromise = new Promise((resolveClosed) => { this.closeResolve = resolveClosed; });
const remoteCommand = task.executionMode === "windows-native"
? ctx().windowsNativeAppServerCommand(task)
: ctx().remoteAppServerCommand(task);
this.child = ctx().providerIsMain(task.providerId)
? spawn("codex", ["app-server", "--listen", "stdio://"], {
cwd: task.cwd,
env: { ...process.env, CODEX_HOME: ctx().config.codexHome, CODEX_INTERNAL_ORIGINATOR_OVERRIDE: "unidesk_code_queue" },
stdio: "pipe",
})
: spawn("bun", ["scripts/cli.ts", "ssh", task.providerId, ctx().remoteAppServerCommand(task)], {
: spawn("bun", ["scripts/cli.ts", "ssh", task.providerId, remoteCommand], {
cwd: ctx().config.defaultWorkdir,
env: process.env,
stdio: "pipe",
@@ -83,7 +99,7 @@ class AppServerClient {
const response = await this.request("thread/resume", {
threadId: this.task.codexThreadId,
model: this.task.model,
cwd: this.task.cwd,
cwd: appServerCwdForTask(this.task),
approvalPolicy: ctx().config.approvalPolicy,
sandbox: ctx().config.sandbox,
});
@@ -94,7 +110,7 @@ class AppServerClient {
}
const response = await this.request("thread/start", {
model: this.task.model,
cwd: this.task.cwd,
cwd: appServerCwdForTask(this.task),
approvalPolicy: ctx().config.approvalPolicy,
sandbox: ctx().config.sandbox,
serviceName: "unidesk-code-queue",
@@ -108,7 +124,7 @@ class AppServerClient {
const params: Record<string, JsonValue> = {
threadId,
input: textInput(prompt),
cwd: this.task.cwd,
cwd: appServerCwdForTask(this.task),
approvalPolicy: ctx().config.approvalPolicy,
model: this.task.model,
};
@@ -324,6 +340,9 @@ export async function runCodexTurn(task: QueueTask, prompt: string): Promise<Cod
let terminalResolve!: () => void;
const terminalPromise = new Promise<void>((resolveTerminal) => { terminalResolve = resolveTerminal; });
await ctx().ensureTaskExecutionContainer(task);
if (task.executionMode === "windows-native") {
ctx().appendOutput(task, "system", "windows-native execution: container stdio relay will connect to Codex on the provider Windows host\n", "execution/mode");
}
const app = new AppServerClient(task, (message) => {
const method = typeof message.method === "string" ? message.method : "unknown";
if (method !== "thread/started" && method !== "turn/started") lastAppActivityAt = Date.now();
@@ -1,6 +1,6 @@
// 重构前 index.ts 只读参考:commit 6a04144d3f5103014f75b637d7e6bc2f45bf007fblob 56e590c1a6b5ca7ad128bf2c992f60e46c355a58;可用 `git show 6a04144d3f5103014f75b637d7e6bc2f45bf007f:src/components/microservices/code-queue/src/index.ts` 查看。
import type { JsonValue } from "../types";
import type { CodeExecutionMode, JsonValue } from "../types";
export type CodeAgentPortKind = "codex" | "opencode";
@@ -29,6 +29,8 @@ export interface ActiveRunSlotWaiter {
export const minimaxM27Model = "minimax-m2.7";
export const defaultCodeModels = ["gpt-5.5", "gpt-5.4-mini", "gpt-5.4", minimaxM27Model];
export const opencodeNpmPackage = "opencode-ai@1.14.48";
export const defaultCodeExecutionMode: CodeExecutionMode = "default";
export const codeExecutionModes: CodeExecutionMode[] = ["default", "windows-native"];
export function normalizeCodeModel(value: string): string {
const raw = String(value || "").trim();
@@ -43,6 +45,33 @@ export function codeAgentPortForModel(model: string): CodeAgentPortKind {
return normalizeCodeModel(model) === minimaxM27Model ? "opencode" : "codex";
}
export function normalizeCodeExecutionMode(value: unknown): CodeExecutionMode {
const raw = typeof value === "string" ? value.trim().toLowerCase() : "";
if (raw === "windows-native" || raw === "windows" || raw === "win32" || raw === "native-windows") return "windows-native";
return defaultCodeExecutionMode;
}
export function codeExecutionModeInfo(mode: CodeExecutionMode): Record<string, JsonValue> {
if (mode === "windows-native") {
return {
kind: mode,
label: "Windows native Codex",
description: "Run Codex on the provider Windows host while the execution container relays stdio.",
codexRunsInContainer: false,
requiresProvider: true,
requiresWindowsCwd: true,
};
}
return {
kind: mode,
label: "Default Codex runtime",
description: "Use the main-server Codex runtime or the provider execution container.",
codexRunsInContainer: true,
requiresProvider: false,
requiresWindowsCwd: false,
};
}
export function opencodeModels(models: string[]): string[] {
return models.filter((model) => codeAgentPortForModel(model) === "opencode");
}
@@ -103,7 +103,7 @@ export async function ensureTaskExecutionContainer(task: QueueTask): Promise<voi
const result = await startDevContainerPlan(plan, {
forceRecreate: false,
verifyPing: false,
prepareRuntime: true,
prepareRuntime: task.executionMode !== "windows-native",
onCommandStart: (command) => {
ctx().appendOutput(task, "system", `provider prepare start name=${command.name} target=${command.providerId} timeoutMs=${command.timeoutMs}\n`, "provider/container");
},
@@ -117,6 +117,7 @@ export async function ensureTaskExecutionContainer(task: QueueTask): Promise<voi
taskId: task.id,
providerId: plan.providerId,
containerName: plan.containerName,
executionMode: task.executionMode,
workdir: task.cwd,
hostWorkdir: plan.workdir,
});
@@ -30,6 +30,7 @@ import {
defaultCodeModels,
extractRecord,
extractString,
normalizeCodeExecutionMode,
normalizeCodeModel,
terminalStatus,
} from "./code-agent/common";
@@ -43,6 +44,7 @@ import {
explicitUserInterrupt,
judgeFailContinuationPrompt,
judgeReasonForPrompt,
judgeTaskInputDiagnostics,
judgeTask,
queueRecoveryRetryPrompt,
retryPrompt,
@@ -57,6 +59,7 @@ import {
containerTunnelStartScript,
defaultWorkdirForProvider,
devContainerPingScript,
executionModeOptions,
executionProviderOptions,
masterKeyReadScript,
masterKeySetupScript,
@@ -76,6 +79,7 @@ import {
runCodeQueueSsh,
shellQuote,
throwIfCommandFailed,
windowsNativeAppServerCommand,
} from "./provider-runtime";
import {
armIdleNotification,
@@ -313,7 +317,7 @@ function readConfig(): RuntimeConfig {
minimaxApiKey: envString("MINIMAX_API_KEY", ""),
minimaxApiBase: envString("MINIMAX_API_BASE", "https://api.minimaxi.com/v1").replace(/\/+$/u, ""),
minimaxModel: envString("MINIMAX_MODEL", "MiniMax-M2.7"),
judgeTimeoutMs: envNumber("MINIMAX_JUDGE_TIMEOUT_MS", 60_000),
judgeTimeoutMs: envNumber("MINIMAX_JUDGE_TIMEOUT_MS", 90_000),
judgeRepairAttempts: Math.max(0, Math.min(5, envNumber("MINIMAX_JUDGE_REPAIR_ATTEMPTS", 2))),
judgeMaxTokens: Math.max(800, Math.min(4000, envNumber("MINIMAX_JUDGE_MAX_TOKENS", 1800))),
turnNoActivityTimeoutMs: Math.max(60_000, Math.min(30 * 60_000, envNumber("CODEX_TURN_NO_ACTIVITY_TIMEOUT_MS", 6 * 60_000))),
@@ -337,6 +341,11 @@ function readConfig(): RuntimeConfig {
devContainerDefaultProviderId,
devContainerImage: envString("CODE_QUEUE_DEV_CONTAINER_IMAGE", ""),
devContainerWorkdir: envString("CODE_QUEUE_DEV_CONTAINER_WORKDIR", remoteDefaultWorkdir),
windowsNativeCodexBridgeDir: envString("CODE_QUEUE_WINDOWS_NATIVE_CODEX_BRIDGE_DIR", "/home/ubuntu/.unidesk/code-queue/windows-native-codex"),
windowsNativeCodexCommand: envString("CODE_QUEUE_WINDOWS_NATIVE_CODEX_COMMAND", "codex app-server --listen stdio://"),
windowsNativeCodexConnectHost: envString("CODE_QUEUE_WINDOWS_NATIVE_CODEX_CONNECT_HOST", "host.docker.internal"),
windowsNativeCodexDefaultWorkdir: envString("CODE_QUEUE_WINDOWS_NATIVE_CODEX_DEFAULT_WORKDIR", "/mnt/f/Work/ConStart"),
windowsNativeCodexIdleTimeoutMs: Math.max(30_000, Math.min(24 * 60 * 60_000, envNumber("CODE_QUEUE_WINDOWS_NATIVE_CODEX_IDLE_TIMEOUT_MS", 10 * 60_000))),
};
}
@@ -562,9 +571,15 @@ function memoryWatchdogThreshold(): number | null {
function activeRunMemoryPressure(): (CgroupMemoryUsage & { thresholdBytes: number }) | null {
const threshold = memoryWatchdogThreshold();
if (threshold === null || threshold <= 0) return null;
const usage = readCgroupMemoryUsage();
const swapHasHeadroom = usage?.swapMaxBytes === null || (usage !== null && usage.swapMaxBytes > 0 && usage.swapCurrentBytes < Math.floor(usage.swapMaxBytes * 0.9));
return usage !== null && usage.workingSetBytes >= threshold && !swapHasHeadroom ? { ...usage, thresholdBytes: threshold } : null;
let usage = readCgroupMemoryUsage();
if (usage === null || usage.workingSetBytes < threshold) return null;
try {
(Bun as unknown as { gc?: (force?: boolean) => void }).gc?.(true);
usage = readCgroupMemoryUsage();
} catch (error) {
logger("debug", "memory_watchdog_gc_failed", { error: errorToJson(error) });
}
return usage !== null && usage.workingSetBytes >= threshold ? { ...usage, thresholdBytes: threshold } : null;
}
function startMemoryWatchdog(): void {
@@ -749,6 +764,7 @@ function normalizeTask(task: QueueTask): QueueTask {
task.activeTurnId ??= null;
task.providerId = normalizeTaskProviderId(task.providerId);
task.model ||= config.defaultModel;
task.executionMode = normalizeCodeExecutionMode(task.executionMode);
task.cwd = resolveTaskCwd(task.providerId, task.cwd);
task.reasoningEffort = resolveReasoningEffort(task.model, task.reasoningEffort);
task.basePrompt ||= userPromptForDisplay(task.prompt);
@@ -816,6 +832,7 @@ function isOpenCodeStepBoundaryMethod(method: string | undefined): boolean {
function outputCanChangeStepCount(output: LiveOutput): boolean {
if (output.channel === "user" && output.method === "enqueue") return false;
if (output.channel === "system") return false;
return !isOpenCodeStepBoundaryMethod(output.method);
}
@@ -827,7 +844,8 @@ function commandStartedBeforeIn(outputs: LiveOutput[], output: LiveOutput): bool
function outputStartsTraceStepInHistory(outputs: LiveOutput[], output: LiveOutput): boolean {
if (output.channel === "user" && output.method === "enqueue") return false;
if (isOpenCodeStepBoundaryMethod(output.method)) return false;
if (output.channel === "diff" || output.channel === "tool" || output.channel === "error" || output.channel === "assistant" || output.channel === "reasoning" || output.channel === "system") return true;
if (output.channel === "system") return false;
if (output.channel === "diff" || output.channel === "tool" || output.channel === "error" || output.channel === "assistant" || output.channel === "reasoning") return true;
if (output.channel === "user") return true;
if (output.channel !== "command") return true;
const method = String(output.method || "");
@@ -900,6 +918,24 @@ function errorToJson(error: unknown): JsonValue {
return String(error);
}
function judgeFailureDetailsForOutput(judge: JudgeResult): string {
const details = judge.failureDetails;
if (details === undefined || details === null) return "";
return [
"",
`MiniMax failure details: stage=${details.stage}`,
`timedOut=${details.timedOut}`,
`durationMs=${details.durationMs}`,
`timeoutMs=${details.timeoutMs}`,
details.promptChars === undefined ? "" : `promptChars=${details.promptChars}`,
details.promptLines === undefined ? "" : `promptLines=${details.promptLines}`,
details.payloadBytes === undefined ? "" : `payloadBytes=${details.payloadBytes}`,
details.responseStatus === undefined || details.responseStatus === null ? "" : `responseStatus=${details.responseStatus}`,
`errorName=${details.errorName}`,
`error=${safePreview(details.errorMessage, 220)}`,
].filter((part) => part.length > 0).join(" ");
}
function databaseErrorMessage(error: unknown): string {
if (error instanceof Error) return error.message;
return String(error);
@@ -966,6 +1002,7 @@ async function upsertTaskToDatabase(client: SqlExecutor, task: QueueTask): Promi
queue_id,
status,
provider_id,
execution_mode,
model,
cwd,
prompt,
@@ -995,6 +1032,7 @@ async function upsertTaskToDatabase(client: SqlExecutor, task: QueueTask): Promi
${queueIdOf(task)},
${task.status},
${task.providerId},
${task.executionMode},
${task.model},
${task.cwd},
${task.prompt},
@@ -1024,6 +1062,7 @@ async function upsertTaskToDatabase(client: SqlExecutor, task: QueueTask): Promi
status = EXCLUDED.status,
queue_id = EXCLUDED.queue_id,
provider_id = EXCLUDED.provider_id,
execution_mode = EXCLUDED.execution_mode,
model = EXCLUDED.model,
cwd = EXCLUDED.cwd,
prompt = EXCLUDED.prompt,
@@ -1352,6 +1391,7 @@ async function initDatabasePersistence(): Promise<void> {
queue_id TEXT NOT NULL DEFAULT 'default',
status TEXT NOT NULL,
provider_id TEXT NOT NULL DEFAULT 'main-server',
execution_mode TEXT NOT NULL DEFAULT 'default',
model TEXT NOT NULL,
cwd TEXT NOT NULL,
prompt TEXT NOT NULL,
@@ -1403,6 +1443,7 @@ async function initDatabasePersistence(): Promise<void> {
`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS queue_id TEXT NOT NULL DEFAULT 'default'`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS provider_id TEXT NOT NULL DEFAULT 'main-server'`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS base_prompt TEXT NOT NULL DEFAULT ''`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS reference_task_ids JSONB NOT NULL DEFAULT '[]'::jsonb`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS reference_injection JSONB`;
@@ -1419,6 +1460,7 @@ async function initDatabasePersistence(): Promise<void> {
await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_status_updated ON unidesk_code_queue_tasks(status, updated_at DESC)`;
await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_queue_status_updated ON unidesk_code_queue_tasks(queue_id, status, updated_at DESC)`;
await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_provider_updated ON unidesk_code_queue_tasks(provider_id, updated_at DESC)`;
await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_execution_mode_updated ON unidesk_code_queue_tasks(execution_mode, updated_at DESC)`;
await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_created ON unidesk_code_queue_tasks(created_at DESC)`;
await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_queue_created ON unidesk_code_queue_tasks(queue_id, created_at DESC, id DESC)`;
await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_unread_terminal ON unidesk_code_queue_tasks(queue_id, updated_at DESC) WHERE read_at IS NULL AND status IN ('succeeded', 'failed', 'canceled')`;
@@ -1669,18 +1711,29 @@ function normalizeRequest(value: unknown): QueueTaskRequest {
if (typeof record.cwd === "string" && record.cwd.length > 0) request.cwd = record.cwd;
if (typeof record.model === "string" && record.model.length > 0) request.model = record.model;
if (typeof record.reasoningEffort === "string" && record.reasoningEffort.length > 0) request.reasoningEffort = record.reasoningEffort;
if (typeof record.executionMode === "string" && record.executionMode.length > 0) request.executionMode = normalizeCodeExecutionMode(record.executionMode);
if (typeof record.maxAttempts === "number" && Number.isInteger(record.maxAttempts) && record.maxAttempts > 0) request.maxAttempts = Math.min(maxTaskAttempts, record.maxAttempts);
const referenceTaskIds = collectReferenceTaskIds(record, record.prompt);
if (referenceTaskIds.length > 0) request.referenceTaskIds = referenceTaskIds;
return request;
}
function validateExecutionModeForTask(providerId: string, cwd: string, model: string, executionMode: ReturnType<typeof normalizeCodeExecutionMode>): void {
if (executionMode !== "windows-native") return;
if (providerIsMain(providerId)) throw new Error("windows-native executionMode requires a non-main WSL provider");
if (codeAgentPortForModel(model) !== "codex") throw new Error("windows-native executionMode only supports Codex models");
if (!cwd.startsWith("/mnt/")) throw new Error("windows-native executionMode requires cwd under /mnt/<drive>");
}
function createTask(request: QueueTaskRequest): QueueTask {
const at = nowIso();
const basePrompt = request.basePrompt ?? userPromptForDisplay(request.prompt);
const referenceTaskIds = request.referenceTaskIds ?? [];
const providerId = normalizeTaskProviderId(request.providerId);
const model = normalizeCodeModel(request.model ?? config.defaultModel);
const executionMode = normalizeCodeExecutionMode(request.executionMode);
const cwd = resolveTaskCwd(providerId, request.cwd);
validateExecutionModeForTask(providerId, cwd, model, executionMode);
const queueId = normalizeQueueId(request.queueId);
ensureQueue(queueId);
return {
@@ -1692,9 +1745,10 @@ function createTask(request: QueueTaskRequest): QueueTask {
referenceTaskIds,
referenceInjection: request.referenceInjection ?? null,
providerId,
cwd: resolveTaskCwd(providerId, request.cwd),
cwd,
model,
reasoningEffort: resolveReasoningEffort(model, request.reasoningEffort),
executionMode,
maxAttempts: request.maxAttempts ?? config.defaultMaxAttempts,
status: "queued",
createdAt: at,
@@ -2067,6 +2121,7 @@ configureCodexPort({
recordNumberField,
recordStringField,
remoteAppServerCommand,
windowsNativeAppServerCommand,
resolveReasoningEffort,
safePreview,
nowIso,
@@ -2149,6 +2204,7 @@ function taskForJudgeProbe(probe: JudgeProbeCase): QueueTask {
cwd: config.defaultWorkdir,
model: config.defaultModel,
reasoningEffort: resolveReasoningEffort(config.defaultModel, config.defaultReasoningEffort),
executionMode: "default",
maxAttempts: 3,
status: "judging",
createdAt: at,
@@ -2239,6 +2295,231 @@ async function runJudgeProbe(): Promise<Response> {
});
}
function cloneJson<T>(value: T): T {
return JSON.parse(JSON.stringify(value)) as T;
}
function numberOrNull(value: unknown): number | null {
const parsed = Number(value);
return Number.isFinite(parsed) ? parsed : null;
}
function maxOutputSeqBefore(output: LiveOutput[], exclusiveSeq: number): number | null {
let maxSeq: number | null = null;
for (const item of output) {
const seq = numberOrNull(item.seq);
if (seq === null || seq >= exclusiveSeq) continue;
maxSeq = maxSeq === null ? seq : Math.max(maxSeq, seq);
}
return maxSeq;
}
function outputSeqAtOrBeforeTime(output: LiveOutput[], at: string | null): number | null {
const atMs = timestampMs(at);
if (atMs === null) return null;
let maxSeq: number | null = null;
for (const item of output) {
const itemMs = timestampMs(item.at);
if (itemMs === null || itemMs > atMs) continue;
const seq = numberOrNull(item.seq);
if (seq === null) continue;
maxSeq = maxSeq === null ? seq : Math.max(maxSeq, seq);
}
return maxSeq;
}
function preJudgeOutputEndSeq(attempt: AttemptSummary, output: LiveOutput[]): { endSeq: number | null; source: string } {
const judgeSeq = numberOrNull(attempt.judgeSeq);
if (judgeSeq !== null) return { endSeq: maxOutputSeqBefore(output, judgeSeq), source: "before-judgeSeq" };
const explicitEnd = numberOrNull(attempt.outputEndSeq);
if (explicitEnd !== null) {
const explicitItem = output.find((item) => numberOrNull(item.seq) === explicitEnd);
if (explicitItem?.method === "judge") return { endSeq: maxOutputSeqBefore(output, explicitEnd), source: "before-outputEndSeq-judge" };
return { endSeq: explicitEnd, source: "outputEndSeq" };
}
const byFinishedAt = outputSeqAtOrBeforeTime(output, attempt.finishedAt);
return { endSeq: byFinishedAt, source: byFinishedAt === null ? "unbounded" : "finishedAt" };
}
function outputBeforeJudge(output: LiveOutput[], endSeq: number | null): LiveOutput[] {
if (endSeq === null) return output;
return output.filter((item) => numberOrNull(item.seq) !== null && Number(item.seq) <= endSeq);
}
function currentAttemptOutputCountForReplay(attempt: AttemptSummary, output: LiveOutput[], hotOutput: LiveOutput[], endSeq: number | null): number {
const startSeq = numberOrNull(attempt.outputStartSeq);
if (startSeq === null) return hotOutput.slice(-80).length;
return output.filter((item) => {
const seq = numberOrNull(item.seq);
return seq !== null && seq >= startSeq && (endSeq === null || seq <= endSeq);
}).length;
}
function finalResponseForTaskBeforeJudge(attempts: AttemptSummary[]): string {
for (let index = attempts.length - 1; index >= 0; index -= 1) {
const finalResponse = String(attempts[index]?.finalResponse ?? "");
if (finalResponse.trim().length > 0) return finalResponse;
}
return "";
}
function retainedEventsForAttempt(task: QueueTask, attempt: AttemptSummary): CodexRunResult["events"] {
const startedMs = timestampMs(attempt.startedAt);
const finishedMs = timestampMs(attempt.finishedAt);
return task.events.filter((event) => {
const eventMs = timestampMs(event.at);
if (eventMs === null) return false;
if (startedMs !== null && eventMs < startedMs) return false;
if (finishedMs !== null && eventMs > finishedMs + 1000) return false;
return true;
});
}
function attemptEventsForReplay(task: QueueTask, attempt: AttemptSummary): { events: CodexRunResult["events"]; source: string; exact: boolean } {
if (Array.isArray(attempt.events)) return { events: cloneJson(attempt.events), source: "attempt.events", exact: true };
return { events: cloneJson(retainedEventsForAttempt(task, attempt)), source: "retained-task-events", exact: false };
}
function replayAttemptIndexFromUrl(task: QueueTask, url: URL): number {
const raw = url.searchParams.get("attempt") ?? url.searchParams.get("attemptId") ?? url.searchParams.get("attemptIndex");
if (raw !== null) {
const value = Number(raw);
if (!Number.isInteger(value) || value <= 0) throw new Error("judge replay attempt must be a positive integer");
return value;
}
const latest = task.attempts.at(-1)?.index ?? task.currentAttempt;
if (!Number.isInteger(latest) || latest <= 0) throw new Error("task has no completed attempt to judge");
return latest;
}
function buildJudgeReplay(task: QueueTask, attemptIndex: number): {
task: QueueTask;
result: CodexRunResult;
attempt: AttemptSummary;
replay: Record<string, JsonValue>;
} {
const attempt = task.attempts.find((item) => item.index === attemptIndex);
if (attempt === undefined) throw new Error(`attempt ${attemptIndex} not found on task ${task.id}`);
const fullOutput = taskFullOutput(task);
const preJudge = preJudgeOutputEndSeq(attempt, fullOutput);
const beforeJudge = outputBeforeJudge(fullOutput, preJudge.endSeq);
const hotOutput = config.maxInMemoryOutputRecords > 0 ? beforeJudge.slice(-config.maxInMemoryOutputRecords) : beforeJudge;
const attempts = task.attempts
.filter((item) => item.index <= attempt.index)
.map((item) => cloneJson(item));
const latestAttempt = attempts[attempts.length - 1];
if (latestAttempt === undefined) throw new Error(`attempt ${attemptIndex} could not be prepared`);
latestAttempt.judge = null;
latestAttempt.judgeAt = null;
latestAttempt.judgeSeq = null;
latestAttempt.outputEndSeq = preJudge.endSeq;
const eventReplay = attemptEventsForReplay(task, attempt);
const previousAttempts = attempts.filter((item) => item.index < attempt.index);
const result: CodexRunResult = {
threadId: task.codexThreadId,
turnId: null,
finalResponse: String(attempt.finalResponse ?? ""),
terminalStatus: attempt.terminalStatus,
terminalError: attempt.error,
transportClosedBeforeTerminal: attempt.transportClosedBeforeTerminal,
appServerExit: {
code: attempt.appServerExitCode,
signal: attempt.appServerSignal,
stderrTail: String(attempt.stderrTail ?? ""),
},
events: eventReplay.events,
};
const replayTask: QueueTask = {
...cloneJson(task),
status: "judging",
currentAttempt: attempt.index,
currentMode: attempt.mode,
attempts,
output: numberOrNull(attempt.outputStartSeq) === null ? hotOutput : beforeJudge,
events: eventReplay.events,
finalResponse: finalResponseForTaskBeforeJudge(attempts),
lastJudge: previousAttempts.at(-1)?.judge ?? null,
judgeFailCount: previousAttempts.filter((item) => item.judge?.decision === "fail").length,
activeTurnId: null,
finishedAt: null,
readAt: null,
nextPrompt: null,
nextMode: null,
};
return {
task: replayTask,
result,
attempt,
replay: {
sameJudgeCodePath: true,
replayExact: eventReplay.exact,
taskId: task.id,
attemptIndex: attempt.index,
attemptMode: attempt.mode,
outputStartSeq: attempt.outputStartSeq ?? null,
storedOutputEndSeq: attempt.outputEndSeq ?? null,
replayOutputEndSeq: preJudge.endSeq,
replayOutputEndSeqSource: preJudge.source,
judgeSeq: attempt.judgeSeq ?? null,
fullOutputBeforeJudgeCount: beforeJudge.length,
hotOutputCount: hotOutput.length,
currentAttemptOutputCount: currentAttemptOutputCountForReplay(attempt, beforeJudge, hotOutput, preJudge.endSeq),
eventSource: eventReplay.source,
eventReplayExact: eventReplay.exact,
eventCount: eventReplay.events.length,
note: eventReplay.exact
? "Replay uses attempt-stored event summaries plus the same judgeTask context builder and MiniMax call path."
: "Historical attempt did not store per-attempt events; replay falls back to retained task.events. Output, final response, terminal status, stderr, attempt window, prompt compaction, and MiniMax call path are reconstructed from persisted task/output archives.",
},
};
}
async function runSingleTaskJudge(task: QueueTask, url: URL): Promise<Response> {
const attemptIndex = replayAttemptIndexFromUrl(task, url);
const dryRun = truthyParam(url, "dryRun") || truthyParam(url, "noCall");
const includePrompt = truthyParam(url, "includePrompt");
const replay = buildJudgeReplay(task, attemptIndex);
const context = judgeTaskInputDiagnostics(replay.task, replay.result, includePrompt);
const startedAt = Date.now();
const judge = dryRun ? null : await judgeTask(replay.task, replay.result);
const durationMs = Date.now() - startedAt;
const storedJudge = replay.attempt.judge ?? null;
const storedFailure = storedJudge?.failureDetails ?? null;
const promptChars = Number(context.promptChars);
const payloadBytes = Number(context.payloadBytes);
return jsonResponse({
ok: true,
dryRun,
configured: config.minimaxApiKey.length > 0,
model: config.minimaxModel,
taskId: task.id,
attempt: {
index: replay.attempt.index,
mode: replay.attempt.mode,
terminalStatus: replay.attempt.terminalStatus,
finalResponseChars: replay.attempt.finalResponseChars ?? String(replay.attempt.finalResponse ?? "").length,
startedAt: replay.attempt.startedAt,
finishedAt: replay.attempt.finishedAt,
},
replay: replay.replay,
context,
judge,
durationMs,
storedJudge,
comparison: {
storedDecision: storedJudge?.decision ?? null,
storedSource: storedJudge?.source ?? null,
decisionMatchesStored: judge === null || storedJudge === null ? null : judge.decision === storedJudge.decision,
storedFailureTimedOut: storedFailure?.timedOut ?? null,
storedFailureStage: storedFailure?.stage ?? null,
storedPromptChars: storedFailure?.promptChars ?? null,
promptCharsDeltaFromStored: storedFailure?.promptChars === undefined || !Number.isFinite(promptChars) ? null : promptChars - storedFailure.promptChars,
storedPayloadBytes: storedFailure?.payloadBytes ?? null,
payloadBytesDeltaFromStored: storedFailure?.payloadBytes === undefined || !Number.isFinite(payloadBytes) ? null : payloadBytes - storedFailure.payloadBytes,
},
});
}
function attemptFromResult(task: QueueTask, mode: RunMode, startedAt: string, finishedAt: string, result: CodexRunResult, outputStartSeq: number | null = null, outputEndSeq: number | null = null, inputPrompt: string | null = null): AttemptSummary {
const finalResponse = result.finalResponse;
const attempt: AttemptSummary = {
@@ -2246,11 +2527,14 @@ function attemptFromResult(task: QueueTask, mode: RunMode, startedAt: string, fi
mode,
startedAt,
finishedAt,
providerId: task.providerId,
executionMode: task.executionMode,
terminalStatus: result.terminalStatus,
transportClosedBeforeTerminal: result.transportClosedBeforeTerminal,
appServerExitCode: result.appServerExit.code,
appServerSignal: result.appServerExit.signal,
error: result.terminalError,
events: result.events,
finalResponse,
finalResponsePreview: safePreview(finalResponse, 3000),
finalResponseChars: finalResponse.length,
@@ -2323,7 +2607,7 @@ function failTaskForFallbackRetryLimit(task: QueueTask, judge: JudgeResult | nul
}
async function runTask(task: QueueTask): Promise<void> {
logger("info", "task_processor_start", { taskId: task.id, queueId: queueIdOf(task), providerId: task.providerId, cwd: task.cwd, maxAttempts: task.maxAttempts, model: task.model, agentPort: codeAgentPortForModel(task.model), promptPreview: safePreview(task.prompt, 240) });
logger("info", "task_processor_start", { taskId: task.id, queueId: queueIdOf(task), providerId: task.providerId, executionMode: task.executionMode, cwd: task.cwd, maxAttempts: task.maxAttempts, model: task.model, agentPort: codeAgentPortForModel(task.model), promptPreview: safePreview(task.prompt, 240) });
if (task.status === "retry_wait" && task.lastJudge?.source === "fallback" && task.lastJudge.decision === "retry" && fallbackJudgeRetryCount(task) >= fallbackJudgeRetryLimit) {
failTaskForFallbackRetryLimit(task, task.lastJudge);
return;
@@ -2354,8 +2638,8 @@ async function runTask(task: QueueTask): Promise<void> {
task.readAt = null;
task.finishedAt = null;
task.updatedAt = startedAt;
logger("info", "task_run_start", { taskId: task.id, queueId: queueIdOf(task), attempt: task.currentAttempt, mode, providerId: task.providerId, cwd: task.cwd, maxAttempts: task.maxAttempts, model: task.model, agentPort: codeAgentPortForModel(task.model), freshRecovery: needsFreshRecoveryPrompt });
const attemptStartOutput = appendOutput(task, "system", `attempt ${task.currentAttempt}/${task.maxAttempts} queue=${queueIdOf(task)} provider=${task.providerId} cwd=${task.cwd} mode=${mode} model=${task.model} port=${codeAgentPortForModel(task.model)}\n`, "queue");
logger("info", "task_run_start", { taskId: task.id, queueId: queueIdOf(task), attempt: task.currentAttempt, mode, providerId: task.providerId, executionMode: task.executionMode, cwd: task.cwd, maxAttempts: task.maxAttempts, model: task.model, agentPort: codeAgentPortForModel(task.model), freshRecovery: needsFreshRecoveryPrompt });
const attemptStartOutput = appendOutput(task, "system", `attempt ${task.currentAttempt}/${task.maxAttempts} queue=${queueIdOf(task)} provider=${task.providerId} executionMode=${task.executionMode} cwd=${task.cwd} mode=${mode} model=${task.model} port=${codeAgentPortForModel(task.model)}\n`, "queue");
let result: CodexRunResult;
try {
@@ -2374,7 +2658,7 @@ async function runTask(task: QueueTask): Promise<void> {
if (task.cancelRequested) break;
const judge = await judgeTask(task, result);
task.lastJudge = judge;
const judgeOutput = appendOutput(task, judge.decision === "complete" ? "system" : judge.decision === "fail" ? "error" : "system", `judge=${judge.decision} confidence=${judge.confidence.toFixed(2)} source=${judge.source}: ${judge.reason}\n`, "judge");
const judgeOutput = appendOutput(task, judge.decision === "complete" ? "system" : judge.decision === "fail" ? "error" : "system", `judge=${judge.decision} confidence=${judge.confidence.toFixed(2)} source=${judge.source}: ${judge.reason}${judgeFailureDetailsForOutput(judge)}\n`, "judge");
const latestAttempt = task.attempts.at(-1);
if (latestAttempt !== undefined && latestAttempt.index === task.currentAttempt) {
latestAttempt.judge = judge;
@@ -2382,7 +2666,7 @@ async function runTask(task: QueueTask): Promise<void> {
latestAttempt.judgeSeq = judgeOutput?.seq ?? null;
latestAttempt.outputEndSeq = judgeOutput?.seq ?? latestAttempt.outputEndSeq ?? null;
}
logger("info", "task_judged", { taskId: task.id, attempt: task.currentAttempt, decision: judge.decision, confidence: judge.confidence, source: judge.source, reason: safePreview(judge.reason, 500) });
logger("info", "task_judged", { taskId: task.id, attempt: task.currentAttempt, decision: judge.decision, confidence: judge.confidence, source: judge.source, reason: safePreview(judge.reason, 500), failureDetails: judge.failureDetails ?? null } as unknown as JsonValue);
if (judge.decision === "complete") {
task.status = "succeeded";
@@ -2856,6 +3140,7 @@ function requestErrorResponse(error: unknown): Response | null {
|| error.message.startsWith("referenceTaskIds supports at most ")
|| error.message.startsWith("queueId must match ")
|| error.message.startsWith("queue name must be ")
|| error.message.startsWith("windows-native executionMode ")
)) {
return jsonResponse({ ok: false, error: error.message }, 400);
}
@@ -2889,7 +3174,7 @@ async function createTasks(req: Request): Promise<Response> {
for (const task of tasks) publishTaskOaEvent(task, "enqueue");
for (const id of new Set(tasks.map(queueIdOf))) publishQueueEvent("enqueue", id);
persistState();
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))) });
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);
return jsonResponse({ ok: true, tasks: tasks.map((task) => taskForResponse(task)), queue: await queueSummaryForResponse() }, 202);
@@ -3402,7 +3687,7 @@ async function route(req: Request): Promise<Response> {
const url = new URL(req.url);
if (req.method === "OPTIONS") return jsonResponse({ ok: true });
try {
if (url.pathname === "/" || url.pathname === "/health") return jsonResponse({ ok: true, service: "code-queue", queue: await queueSummaryForHealth(), startedAt: serviceStartedAt });
if (url.pathname === "/" || url.pathname === "/health") return jsonResponse({ ok: true, service: "code-queue", queue: await queueSummaryForHealth(false), startedAt: serviceStartedAt });
if (url.pathname === "/logs") return jsonResponse({ ok: true, logs: recentLogs.slice(-parseLimit(url)) });
if (url.pathname === "/api/events" && req.method === "GET") return jsonResponse({ ok: false, error: "Code Queue private SSE was removed; subscribe to oa-event-flow /api/events/stream with service:code-queue tags." }, 410);
if (url.pathname === "/api/dev-ready" && req.method === "GET") return jsonResponse({ ok: true, devReady: collectDevReady() });
@@ -3546,6 +3831,12 @@ async function route(req: Request): Promise<Response> {
if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
return await taskTraceStepDetailResponse(task, url);
}
const judgeTaskMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/judge$/u);
if (judgeTaskMatch !== null && (req.method === "GET" || req.method === "POST")) {
const task = await findTaskForRead(decodeURIComponent(judgeTaskMatch[1] ?? ""));
if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
return await runSingleTaskJudge(task, url);
}
const summaryMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/summary$/u);
if (summaryMatch !== null && req.method === "GET") {
const task = await findTaskForRead(decodeURIComponent(summaryMatch[1] ?? ""));
@@ -4,6 +4,7 @@ import type {
CodexRunResult,
FeedbackPromptRecord,
JudgeDecision,
JudgeFailureDetails,
JudgeProbeCase,
JudgeResult,
JsonValue,
@@ -80,7 +81,13 @@ function promptLineCount(text: string): number {
return ctx().promptLineCount(text);
}
function fallbackJudge(result: CodexRunResult, minimaxError?: string): JudgeResult {
function failureDetailReason(detail: JudgeFailureDetails): string {
const timeout = detail.timedOut ? `timed out after ${detail.timeoutMs}ms` : detail.errorMessage;
const promptSize = detail.promptChars === undefined ? "" : `prompt=${detail.promptChars} chars/${detail.promptLines ?? "?"} lines`;
return `MiniMax judge ${timeout}${promptSize}`;
}
function fallbackJudge(result: CodexRunResult, minimaxError?: string | JudgeFailureDetails): JudgeResult {
if (result.transportClosedBeforeTerminal || result.terminalStatus === null) {
return { decision: "retry", confidence: 0.75, reason: "Codex app-server 在 turn/completed 之前关闭。", continuePrompt: retryInstruction, source: "fallback" };
}
@@ -94,11 +101,17 @@ function fallbackJudge(result: CodexRunResult, minimaxError?: string): JudgeResu
return { decision: "retry", confidence: 0.78, reason: "Codex turn 没有返回最终 assistant response,不能视为任务已完成。", continuePrompt: retryInstruction, source: "fallback" };
}
if (minimaxError !== undefined) {
const failureDetails = typeof minimaxError === "string" ? null : minimaxError;
const errorText = typeof minimaxError === "string" ? minimaxError : failureDetailReason(minimaxError);
return {
decision: "retry",
confidence: 0.65,
reason: `MiniMax judge 失败(${safePreview(minimaxError, 240)});安全 fallback 将继续现有 session,而不是把 turn/completed 当作任务已完成。`,
reason: `MiniMax judge 失败(${safePreview(errorText, 300)});安全 fallback 将继续现有 session,而不是把 turn/completed 当作任务已完成。`,
source: "fallback",
failureDetails,
raw: failureDetails === null
? { minimaxFailure: { errorMessage: String(minimaxError) } }
: { minimaxFailure: failureDetails as unknown as JsonValue },
};
}
return { decision: "complete", confidence: 0.65, reason: "Codex 输出了 completed 状态的 turn/completed,且未配置 MiniMax judge。", source: "fallback" };
@@ -135,21 +148,99 @@ function judgeEvidenceText(task: QueueTask, result: CodexRunResult): string {
].join("\n");
}
function encodedBytes(value: string): number {
return new TextEncoder().encode(value).length;
}
function compactJudgeOutputItems(output: LiveOutput[]): { items: Array<{ channel: string; text: string; method?: string; seq?: number; at?: string }>; omitted: number; totalTextChars: number } {
const important = output.filter((item) => {
const method = String(item.method || "");
const text = String(item.text || "");
return item.channel === "error"
|| method === "turn/completed"
|| method === "item/completed"
|| method === "item/started"
|| method === "provider/container"
|| method === "execution/mode"
|| method.includes("watchdog")
|| /(error|failed||timeout|timed out|aborted|rebuild|health|succeeded|verified|passed|judge)/iu.test(text);
});
const bySeq = new Map<number, LiveOutput>();
for (const item of [...important.slice(-32), ...output.slice(-28)]) {
const seq = Number(item.seq);
if (Number.isFinite(seq)) bySeq.set(seq, item);
}
const selected = Array.from(bySeq.values()).sort((left, right) => Number(left.seq) - Number(right.seq));
const items: Array<{ channel: string; text: string; method?: string; seq?: number; at?: string }> = [];
let remainingChars = 14_000;
let totalTextChars = 0;
for (const item of selected) {
const rawText = String(item.text || "");
totalTextChars += rawText.length;
if (remainingChars <= 0) break;
const perItem = item.channel === "error" ? 700 : 360;
const max = Math.max(120, Math.min(perItem, remainingChars));
const text = safePreview(rawText, max);
remainingChars -= text.length;
items.push({
channel: item.channel,
method: item.method,
seq: item.seq,
at: item.at,
text,
});
}
return { items, omitted: Math.max(0, output.length - items.length), totalTextChars };
}
function compactJudgeEvents(events: CodexRunResult["events"]): { events: CodexRunResult["events"]; omitted: number } {
const selected = events.slice(-30).map((event) => ({
...event,
message: event.message === undefined ? undefined : safePreview(event.message, 300),
textPreview: event.textPreview === undefined ? undefined : safePreview(event.textPreview, 300),
}));
return { events: selected, omitted: Math.max(0, events.length - selected.length) };
}
function hasServiceLimitOrTransportError(text: string): boolean {
return /(429|Too Many Requests|rate limit|quota exceeded|overloaded|exceeded retry limit|stream disconnected|no activity timeout|app-server closed|ECONNRESET|ETIMEDOUT)/iu.test(text);
}
function compactLatestAttemptForJudge(attempt: QueueTask["attempts"][number] | null): Record<string, unknown> | null {
if (attempt === null) return null;
return {
index: attempt.index,
mode: attempt.mode,
startedAt: attempt.startedAt,
finishedAt: attempt.finishedAt,
providerId: attempt.providerId,
executionMode: attempt.executionMode,
terminalStatus: attempt.terminalStatus,
transportClosedBeforeTerminal: attempt.transportClosedBeforeTerminal,
appServerExitCode: attempt.appServerExitCode,
appServerSignal: attempt.appServerSignal,
error: attempt.error,
finalResponsePreview: safePreview(String(attempt.finalResponsePreview || attempt.finalResponse || ""), 1200),
finalResponseChars: Number(attempt.finalResponseChars ?? String(attempt.finalResponse || "").length),
stderrTail: safePreview(String(attempt.stderrTail || ""), 1200),
outputStartSeq: attempt.outputStartSeq ?? null,
outputEndSeq: attempt.outputEndSeq ?? null,
};
}
function judgePrompt(task: QueueTask, result: CodexRunResult): string {
const latestAttempt = task.attempts[task.attempts.length - 1] ?? null;
const originalUserTask = task.basePrompt || userPromptForDisplay(task.prompt);
const resolvedPromptForCodex = task.prompt === originalUserTask ? null : safePreview(task.prompt, 12000);
const resolvedPromptForCodex = task.prompt === originalUserTask ? null : safePreview(task.prompt, 6000);
const currentAttemptOutput = currentAttemptOutputForJudge(task);
const outputEvidence = compactJudgeOutputItems(currentAttemptOutput);
const eventEvidence = compactJudgeEvents(result.events);
// Keep this record factual. Do not add local completion gates here; MiniMax
// is authoritative whenever it returns a valid judge JSON.
return JSON.stringify({
instruction: "请判定一个 Codex 编码任务是否真正完成、是否应通过向现有 Codex thread 追加 continuation prompt 继续重试,或是否应作为不可重试失败处理。只能返回 JSON,不要输出 Markdown fence、解释性正文或注释。所有自然语言字段必须使用中文,尤其是 reason 和 continuePrompt。重要:普通的 Codex turn/completed 状态只表示传输/session 终止事件,不等于用户任务已完成。决策前必须检查当前尝试的 transcript、最终回复、命令/文件变更事件、stderr 和原始任务。请严格判定:如果用户任务中的任一显式验收项缺少已完成证据,就选择 retry。最常见错误是把未完成或跑偏的工作标成 fail;未完成工作必须选择 retry,让同一个 session 继续。不要把更早 attempt 的限流/中断证据自动当作当前 attempt 的完成门禁;如果当前最新 attempt 已经提供完整完成证据,可以判定 complete。",
schema: { decision: "complete|retry|fail", confidence: "0..1", reason: "中文短句", continuePrompt: "decision=retry 时必填,除非确实没有可用的继续提示;内容必须是中文,保持简洁,不要粘贴原始任务、引用上下文、transcript 或 JSON" },
originalTask: originalUserTask,
originalTask: safePreview(originalUserTask, 6000),
resolvedPromptForCodex,
attempt: task.currentAttempt,
maxAttempts: task.maxAttempts,
@@ -163,12 +254,15 @@ function judgePrompt(task: QueueTask, result: CodexRunResult): string {
finalResponse: safePreview(result.finalResponse, 6000),
finalResponseChars: result.finalResponse.length,
finalResponseMissing: result.finalResponse.trim().length === 0,
latestAttempt,
latestAttempt: compactLatestAttemptForJudge(latestAttempt),
judgeFailCount: task.judgeFailCount,
judgeFailRetryLimit: ctx().judgeFailRetryLimit,
cancelRequested: task.cancelRequested,
currentAttemptOutput: currentAttemptOutput.slice(-80).map((item) => ({ channel: item.channel, text: safePreview(item.text, 500), method: item.method })),
currentAttemptEvents: result.events.slice(-60),
currentAttemptOutput: outputEvidence.items,
currentAttemptOutputOmitted: outputEvidence.omitted,
currentAttemptOutputTotalTextChars: outputEvidence.totalTextChars,
currentAttemptEvents: eventEvidence.events,
currentAttemptEventsOmitted: eventEvidence.omitted,
},
policy: {
complete: "仅当 transcript/最终回复证明当前任务确实完成,并且每个显式验收项都有证据时使用;队列 worker 随后会推进到下一个 queued 任务。",
@@ -513,6 +607,102 @@ export function miniMaxJudgeMessages(userContent: string, repair: JudgeRepairCon
return messages;
}
class MiniMaxJudgeFailure extends Error {
readonly details: JudgeFailureDetails;
constructor(details: JudgeFailureDetails) {
super(details.errorMessage);
this.name = "MiniMaxJudgeFailure";
this.details = details;
}
}
function errorName(error: unknown): string {
return error instanceof Error ? error.name || "Error" : typeof error;
}
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}
function minimaxFailureDetails(input: {
stage: JudgeFailureDetails["stage"];
error: unknown;
startedAt: number;
finishedAt?: number;
timedOut?: boolean;
repairAttempt?: number;
maxRepairAttempts?: number;
promptChars?: number;
promptLines?: number;
payloadBytes?: number;
responseStatus?: number | null;
responseText?: string;
responseContent?: string;
}): JudgeFailureDetails {
const finishedAt = input.finishedAt ?? Date.now();
return {
provider: "minimax",
stage: input.stage,
model: config().minimaxModel,
apiBase: config().minimaxApiBase,
occurredAt: new Date(finishedAt).toISOString(),
durationMs: Math.max(0, finishedAt - input.startedAt),
timeoutMs: config().judgeTimeoutMs,
timedOut: input.timedOut === true,
errorName: errorName(input.error),
errorMessage: errorMessage(input.error),
repairAttempt: input.repairAttempt,
maxRepairAttempts: input.maxRepairAttempts,
promptChars: input.promptChars,
promptLines: input.promptLines,
payloadBytes: input.payloadBytes,
responseStatus: input.responseStatus,
responseTextPreview: input.responseText === undefined ? undefined : safePreview(input.responseText, 1000),
responseTextChars: input.responseText?.length,
responseContentPreview: input.responseContent === undefined ? undefined : safePreview(input.responseContent, 1000),
responseContentChars: input.responseContent?.length,
};
}
function messagePromptStats(messages: Array<{ role: string; content: string }>): { promptChars: number; promptLines: number } {
return messages.reduce((stats, message) => {
const content = String(message.content || "");
stats.promptChars += content.length;
stats.promptLines += promptLineCount(content);
return stats;
}, { promptChars: 0, promptLines: 0 });
}
export function judgeTaskInputDiagnostics(task: QueueTask, result: CodexRunResult, includePrompt = false): Record<string, JsonValue> {
const judgePromptContent = judgePrompt(task, result);
const messages = miniMaxJudgeMessages(judgePromptContent, null);
const promptStats = messagePromptStats(messages);
const body = JSON.stringify({
model: config().minimaxModel,
temperature: 0,
max_tokens: config().judgeMaxTokens,
messages,
});
return {
provider: "minimax",
model: config().minimaxModel,
apiBase: config().minimaxApiBase,
configured: config().minimaxApiKey.length > 0,
timeoutMs: config().judgeTimeoutMs,
repairAttempts: config().judgeRepairAttempts,
maxTokens: config().judgeMaxTokens,
messageCount: messages.length,
judgePromptChars: judgePromptContent.length,
judgePromptLines: promptLineCount(judgePromptContent),
promptChars: promptStats.promptChars,
promptLines: promptStats.promptLines,
payloadBytes: encodedBytes(body),
promptPreview: safePreview(judgePromptContent, 1200),
...(includePrompt ? { prompt: judgePromptContent } : {}),
};
}
function judgeScopeText(task: QueueTask, result: CodexRunResult): string {
return [
task.basePrompt,
@@ -673,12 +863,13 @@ function applyJudgeSafetyOverrides(task: QueueTask, result: CodexRunResult, judg
export async function judgeTask(task: QueueTask, result: CodexRunResult): Promise<JudgeResult> {
if (config().minimaxApiKey.length === 0) return applyJudgeSafetyOverrides(task, result, fallbackJudge(result));
const judgePromptContent = judgePrompt(task, result);
const basePromptStats = { promptChars: judgePromptContent.length, promptLines: promptLineCount(judgePromptContent) };
try {
let lastParseError: string | null = null;
let repairContext: JudgeRepairContext | null = null;
for (let repairAttempt = 0; repairAttempt <= config().judgeRepairAttempts; repairAttempt += 1) {
const messages = miniMaxJudgeMessages(judgePromptContent, repairContext);
const response = await requestMiniMaxJudge(messages);
const response = await requestMiniMaxJudge(messages, repairAttempt);
const preDenoiseContent = response.content;
try {
const parsedResult = parseJudgeJson(preDenoiseContent);
@@ -695,12 +886,32 @@ export async function judgeTask(task: QueueTask, result: CodexRunResult): Promis
reason: typeof parsed.reason === "string" ? parsed.reason : "MiniMax judge returned a decision.",
continuePrompt,
source: "minimax",
raw: { ...(parsed as Record<string, JsonValue>), _parseSource: parsedResult.source, _repairAttempt: repairAttempt },
raw: {
...(parsed as Record<string, JsonValue>),
_parseSource: parsedResult.source,
_repairAttempt: repairAttempt,
_request: response.diagnostics as unknown as JsonValue,
},
};
return applyJudgeSafetyOverrides(task, result, judge);
} catch (error) {
lastParseError = error instanceof Error ? error.message : String(error);
if (repairAttempt >= config().judgeRepairAttempts) throw new Error(lastParseError);
if (repairAttempt >= config().judgeRepairAttempts) {
throw new MiniMaxJudgeFailure(minimaxFailureDetails({
stage: /continuePrompt exceeds source budget/iu.test(lastParseError) ? "validation" : "parse",
error: new Error(lastParseError),
startedAt: Date.now() - response.diagnostics.durationMs,
finishedAt: Date.now(),
repairAttempt,
maxRepairAttempts: config().judgeRepairAttempts,
promptChars: response.diagnostics.promptChars,
promptLines: response.diagnostics.promptLines,
payloadBytes: response.diagnostics.payloadBytes,
responseStatus: response.diagnostics.responseStatus,
responseText: response.rawText,
responseContent: preDenoiseContent,
}));
}
logger("warn", "judge_json_parse_retry", {
taskId: task.id,
repairAttempt: repairAttempt + 1,
@@ -713,29 +924,99 @@ export async function judgeTask(task: QueueTask, result: CodexRunResult): Promis
}
throw new Error(lastParseError ?? "MiniMax judge exhausted JSON repair attempts");
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
logger("warn", "judge_failed_fallback", { taskId: task.id, error: message });
return applyJudgeSafetyOverrides(task, result, fallbackJudge(result, message));
const detail = error instanceof MiniMaxJudgeFailure
? error.details
: minimaxFailureDetails({ stage: "unknown", error, startedAt: Date.now(), promptChars: basePromptStats.promptChars, promptLines: basePromptStats.promptLines });
logger("warn", "judge_failed_fallback", {
taskId: task.id,
error: detail.errorMessage,
errorName: detail.errorName,
stage: detail.stage,
timedOut: detail.timedOut,
durationMs: detail.durationMs,
timeoutMs: detail.timeoutMs,
promptChars: detail.promptChars,
promptLines: detail.promptLines,
payloadBytes: detail.payloadBytes,
responseStatus: detail.responseStatus,
repairAttempt: detail.repairAttempt,
maxRepairAttempts: detail.maxRepairAttempts,
} as unknown as JsonValue);
return applyJudgeSafetyOverrides(task, result, fallbackJudge(result, detail));
}
}
async function requestMiniMaxJudge(messages: Array<{ role: "system" | "user" | "assistant"; content: string }>): Promise<MiniMaxJudgeResponse> {
async function requestMiniMaxJudge(messages: Array<{ role: "system" | "user" | "assistant"; content: string }>, repairAttempt: number): Promise<MiniMaxJudgeResponse> {
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), config().judgeTimeoutMs);
let timedOut = false;
const startedAt = Date.now();
const promptStats = messagePromptStats(messages);
const body = JSON.stringify({
model: config().minimaxModel,
temperature: 0,
max_tokens: config().judgeMaxTokens,
messages,
});
const payloadBytes = encodedBytes(body);
const timer = setTimeout(() => {
timedOut = true;
controller.abort();
}, config().judgeTimeoutMs);
try {
const response = await fetch(`${config().minimaxApiBase}/chat/completions`, {
method: "POST",
headers: { authorization: `Bearer ${config().minimaxApiKey}`, "content-type": "application/json" },
body: JSON.stringify({
model: config().minimaxModel,
temperature: 0,
max_tokens: config().judgeMaxTokens,
messages,
}),
signal: controller.signal,
});
const rawText = await response.text();
if (!response.ok) throw new Error(`MiniMax HTTP ${response.status}: ${safePreview(rawText, 1000)}`);
let response: Response;
try {
response = await fetch(`${config().minimaxApiBase}/chat/completions`, {
method: "POST",
headers: { authorization: `Bearer ${config().minimaxApiKey}`, "content-type": "application/json" },
body,
signal: controller.signal,
});
} catch (error) {
throw new MiniMaxJudgeFailure(minimaxFailureDetails({
stage: "request",
error,
startedAt,
timedOut,
repairAttempt,
maxRepairAttempts: config().judgeRepairAttempts,
promptChars: promptStats.promptChars,
promptLines: promptStats.promptLines,
payloadBytes,
}));
}
let rawText: string;
try {
rawText = await response.text();
} catch (error) {
throw new MiniMaxJudgeFailure(minimaxFailureDetails({
stage: "request",
error,
startedAt,
timedOut,
repairAttempt,
maxRepairAttempts: config().judgeRepairAttempts,
promptChars: promptStats.promptChars,
promptLines: promptStats.promptLines,
payloadBytes,
responseStatus: response.status,
}));
}
const finishedAt = Date.now();
if (!response.ok) {
throw new MiniMaxJudgeFailure(minimaxFailureDetails({
stage: "http",
error: new Error(`MiniMax HTTP ${response.status}: ${safePreview(rawText, 1000)}`),
startedAt,
finishedAt,
repairAttempt,
maxRepairAttempts: config().judgeRepairAttempts,
promptChars: promptStats.promptChars,
promptLines: promptStats.promptLines,
payloadBytes,
responseStatus: response.status,
responseText: rawText,
}));
}
let content = rawText;
try {
const payload = JSON.parse(rawText) as Record<string, unknown>;
@@ -749,7 +1030,23 @@ async function requestMiniMaxJudge(messages: Array<{ role: "system" | "user" | "
} catch {
content = rawText;
}
return { rawText, content };
return {
rawText,
content,
diagnostics: {
provider: "minimax",
model: config().minimaxModel,
apiBase: config().minimaxApiBase,
durationMs: finishedAt - startedAt,
timeoutMs: config().judgeTimeoutMs,
promptChars: promptStats.promptChars,
promptLines: promptStats.promptLines,
payloadBytes,
responseStatus: response.status,
responseTextChars: rawText.length,
responseContentChars: content.length,
},
};
} finally {
clearTimeout(timer);
}
@@ -184,6 +184,7 @@ function taskStatusPayload(task: QueueTask, queueId: string, reason: string, ste
finishedAt: task.finishedAt,
readAt: task.readAt,
providerId: task.providerId,
executionMode: task.executionMode,
model: task.model,
currentAttempt: task.currentAttempt,
stepCount,
@@ -3,12 +3,12 @@
import { spawn } from "node:child_process";
import { existsSync, readFileSync } from "node:fs";
import { resolve } from "node:path";
import { opencodeNpmPackage } from "./code-agent/common";
import { codeExecutionModeInfo, opencodeNpmPackage } from "./code-agent/common";
import type { DevContainerCommandLog, DevContainerPlan, JsonValue, QueueTask, RuntimeConfig } from "./types";
export interface ProviderRuntimeContext {
config: Pick<RuntimeConfig,
"codexHome" | "defaultWorkdir" | "devContainerDefaultProviderId" | "devContainerImage" | "devContainerMasterHost" | "devContainerWorkdir" | "executionProviderIds" | "mainProviderId" | "remoteCodexEnvKeys" | "remoteDefaultWorkdir" | "sourceCodexConfig"
"codexHome" | "defaultWorkdir" | "devContainerDefaultProviderId" | "devContainerImage" | "devContainerMasterHost" | "devContainerWorkdir" | "executionProviderIds" | "mainProviderId" | "remoteCodexEnvKeys" | "remoteDefaultWorkdir" | "sourceCodexConfig" | "windowsNativeCodexBridgeDir" | "windowsNativeCodexCommand" | "windowsNativeCodexConnectHost" | "windowsNativeCodexDefaultWorkdir" | "windowsNativeCodexIdleTimeoutMs"
>;
safePreview: (value: string, max?: number) => string;
}
@@ -73,10 +73,29 @@ function executionProviderOptions(): JsonValue[] {
label: providerIsMain(providerId) ? `${providerId} (master)` : providerId,
kind: providerIsMain(providerId) ? "local" : "remote-dev-container",
defaultWorkdir: defaultWorkdirForProvider(providerId),
supportsWindowsNativeCodex: !providerIsMain(providerId),
windowsNativeDefaultWorkdir: providerIsMain(providerId) ? null : ctx().config.windowsNativeCodexDefaultWorkdir,
containerName: providerIsMain(providerId) ? null : buildDevContainerPlan(providerId, {}).containerName,
}));
}
function executionModeOptions(): JsonValue[] {
return [
{
...codeExecutionModeInfo("default"),
id: "default",
defaultWorkdir: null,
},
{
...codeExecutionModeInfo("windows-native"),
id: "windows-native",
defaultWorkdir: ctx().config.windowsNativeCodexDefaultWorkdir,
bridgeDir: ctx().config.windowsNativeCodexBridgeDir,
connectHost: ctx().config.windowsNativeCodexConnectHost,
},
];
}
function numericIdFromProvider(providerId: string): number {
const digits = providerId.match(/\d+/u)?.[0] ?? "";
if (digits.length > 0) {
@@ -457,6 +476,321 @@ CONTAINER=${shellQuote(plan.containerName)}
docker exec "$CONTAINER" bash -lc 'export PATH=/tmp/unidesk-tools:$PATH; echo route=$(ip route show default | head -1); echo resolv=$(tr "\\n" " " </etc/resolv.conf); ping -c 1 -W 5 google.com'`;
}
function windowsNativeBridgeServerSource(): string {
return String.raw`#!/usr/bin/env python3
import argparse
import os
import socket
import subprocess
import sys
import threading
import time
def write_text(path, text):
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as handle:
handle.write(text)
os.replace(tmp, path)
def main():
parser = argparse.ArgumentParser(description="UniDesk Windows native Codex stdio bridge")
parser.add_argument("--session", required=True)
parser.add_argument("--cwd", required=True)
parser.add_argument("--codex-command", required=True)
parser.add_argument("--bind-host", default="0.0.0.0")
parser.add_argument("--port", type=int, default=0)
parser.add_argument("--port-file", required=True)
parser.add_argument("--pid-file", required=True)
parser.add_argument("--idle-timeout", type=float, default=600)
parser.add_argument("--initial-connect-timeout", type=float, default=90)
parser.add_argument("--stdout-log", required=True)
parser.add_argument("--stderr-log", required=True)
args = parser.parse_args()
if not args.cwd.startswith("/mnt/"):
raise SystemExit("windows-native Codex cwd must be under /mnt/<drive>")
if not os.path.isdir(args.cwd):
raise SystemExit("windows-native Codex cwd does not exist: " + args.cwd)
env = os.environ.copy()
home = os.path.expanduser("~")
env["PATH"] = os.path.join(home, ".local", "bin") + os.pathsep + env.get("PATH", "")
env["CODEX_INTERNAL_ORIGINATOR_OVERRIDE"] = "unidesk_code_queue_windows_native"
bridge_env_keys = [
"OPENAI_API_KEY",
"CRS_OAI_KEY",
"OPENAI_BASE_URL",
"OPENAI_API_BASE",
"MINIMAX_API_KEY",
"MINIMAX_API_BASE",
"MINIMAX_MODEL",
"CODEX_INTERNAL_ORIGINATOR_OVERRIDE",
]
existing_wsl_env = [item for item in env.get("WSLENV", "").split(":") if item]
existing_wsl_names = {item.split("/", 1)[0] for item in existing_wsl_env}
for key in bridge_env_keys:
if key in env and key not in existing_wsl_names:
existing_wsl_env.append(key)
env["WSLENV"] = ":".join(existing_wsl_env)
child = subprocess.Popen(
["win-cmd", args.codex_command],
cwd=args.cwd,
env=env,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
bufsize=0,
)
clients = set()
lock = threading.Lock()
last_empty_at = time.time()
ever_connected = False
stop_event = threading.Event()
def log(message):
print(message, file=sys.stderr, flush=True)
def close_client(conn):
try:
conn.close()
except OSError:
pass
with lock:
clients.discard(conn)
def broadcast(data):
dead = []
with lock:
snapshot = list(clients)
for conn in snapshot:
try:
conn.sendall(data)
except OSError:
dead.append(conn)
for conn in dead:
close_client(conn)
def pump_stdout():
with open(args.stdout_log, "ab", buffering=0) as log_file:
while True:
data = child.stdout.readline() if child.stdout is not None else b""
if not data:
break
log_file.write(data)
broadcast(data)
stop_event.set()
with lock:
snapshot = list(clients)
for conn in snapshot:
close_client(conn)
def pump_stderr():
with open(args.stderr_log, "ab", buffering=0) as log_file:
while True:
data = child.stderr.readline() if child.stderr is not None else b""
if not data:
break
log_file.write(data)
threading.Thread(target=pump_stdout, daemon=True).start()
threading.Thread(target=pump_stderr, daemon=True).start()
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((args.bind_host, args.port))
server.listen(8)
server.settimeout(1)
port = server.getsockname()[1]
os.makedirs(os.path.dirname(args.port_file), exist_ok=True)
write_text(args.port_file, str(port))
write_text(args.pid_file, str(os.getpid()))
log("windows_native_codex_bridge_ready session=%s pid=%s port=%s child=%s cwd=%s" % (args.session, os.getpid(), port, child.pid, args.cwd))
def handle_client(conn):
nonlocal last_empty_at
try:
while True:
data = conn.recv(65536)
if not data:
break
if child.stdin is not None:
child.stdin.write(data)
child.stdin.flush()
except (BrokenPipeError, OSError):
pass
finally:
close_client(conn)
with lock:
if len(clients) == 0:
last_empty_at = time.time()
try:
while not stop_event.is_set():
if child.poll() is not None:
break
with lock:
empty_for = time.time() - last_empty_at if len(clients) == 0 else 0
no_client_yet = not ever_connected and len(clients) == 0
if no_client_yet and empty_for >= args.initial_connect_timeout:
log("windows_native_codex_bridge_initial_connect_timeout session=%s seconds=%.1f" % (args.session, empty_for))
break
if not no_client_yet and empty_for >= args.idle_timeout:
log("windows_native_codex_bridge_idle_timeout session=%s seconds=%.1f" % (args.session, empty_for))
break
try:
conn, _addr = server.accept()
except socket.timeout:
continue
with lock:
clients.add(conn)
ever_connected = True
threading.Thread(target=handle_client, args=(conn,), daemon=True).start()
finally:
try:
server.close()
except OSError:
pass
if child.poll() is None:
child.terminate()
try:
child.wait(timeout=5)
except subprocess.TimeoutExpired:
child.kill()
try:
os.remove(args.port_file)
except OSError:
pass
log("windows_native_codex_bridge_exit session=%s childExit=%s" % (args.session, child.poll()))
if __name__ == "__main__":
main()
`;
}
function windowsNativeContainerClientSource(): string {
return String.raw`const fs = require("node:fs");
const dns = require("node:dns");
const net = require("node:net");
const { spawnSync } = require("node:child_process");
const host = process.env.UNIDESK_WINDOWS_NATIVE_CODEX_HOST || "host.docker.internal";
const port = Number(process.env.UNIDESK_WINDOWS_NATIVE_CODEX_PORT);
const extraHosts = String(process.env.UNIDESK_WINDOWS_NATIVE_CODEX_EXTRA_HOSTS || "").split(/[,\s]+/u).filter(Boolean);
const connectDeadlineMs = Math.max(5000, Number(process.env.UNIDESK_WINDOWS_NATIVE_CODEX_CONNECT_TIMEOUT_MS || 60000));
if (!Number.isInteger(port) || port <= 0) throw new Error("UNIDESK_WINDOWS_NATIVE_CODEX_PORT is invalid");
function gatewayFromHex(value) {
const bytes = value.match(/../g);
return bytes ? bytes.reverse().map((item) => Number.parseInt(item, 16)).join(".") : "";
}
function routeGateways() {
try {
return fs.readFileSync("/proc/net/route", "utf8").trim().split(/\n/u).slice(1).flatMap((line) => {
const parts = line.trim().split(/\s+/u);
if (parts.length < 3 || parts[0].startsWith("tun") || parts[2] === "00000000") return [];
const gateway = gatewayFromHex(parts[2]);
return gateway.length > 0 ? [{ iface: parts[0], gateway }] : [];
});
} catch {
return [];
}
}
function lookup4(name) {
return new Promise((resolve) => {
dns.lookup(name, { family: 4, all: true }, (error, addresses) => {
resolve(error ? [] : [...new Set(addresses.map((item) => item.address))]);
});
});
}
function ensureHostRoutes(ips) {
const ipBin = fs.existsSync("/tmp/unidesk-tools/ip") ? "/tmp/unidesk-tools/ip" : "ip";
for (const { iface, gateway } of routeGateways()) {
for (const ip of ips) {
spawnSync(ipBin, ["route", "replace", ip + "/32", "via", gateway, "dev", iface], { stdio: "ignore" });
}
}
}
function connectOnce(candidate) {
return new Promise((resolve, reject) => {
const socket = net.createConnection({ host: candidate, port });
const timer = setTimeout(() => socket.destroy(new Error("connect timeout")), 2500);
socket.once("connect", () => {
clearTimeout(timer);
resolve(socket);
});
socket.once("error", (error) => {
clearTimeout(timer);
reject(error);
});
});
}
function sleep(ms) {
return new Promise((resolve) => setTimeout(resolve, ms));
}
async function connect() {
const hostIps = await lookup4(host);
ensureHostRoutes(hostIps);
const candidates = [...new Set([...hostIps, host, ...extraHosts, ...routeGateways().map((item) => item.gateway), "172.17.0.1"].filter(Boolean))];
const errors = [];
const deadline = Date.now() + connectDeadlineMs;
let attempt = 0;
while (Date.now() < deadline) {
attempt += 1;
for (const candidate of candidates) {
try {
return await connectOnce(candidate);
} catch (error) {
errors.push(candidate + "=" + (error && error.message ? error.message : String(error)));
while (errors.length > 24) errors.shift();
}
}
await sleep(500);
}
throw new Error("could not connect to Windows native Codex bridge after " + attempt + " attempts in " + connectDeadlineMs + "ms: " + errors.join("; "));
}
(async () => {
const socket = await connect();
socket.setNoDelay(true);
socket.on("data", (chunk) => {
if (!process.stdout.write(chunk)) socket.pause();
});
process.stdout.on("drain", () => socket.resume());
socket.on("end", () => process.exit(0));
socket.on("close", () => process.exit(0));
socket.on("error", (error) => {
console.error(error && error.stack ? error.stack : String(error));
process.exit(1);
});
process.stdin.on("data", (chunk) => {
if (!socket.write(chunk)) process.stdin.pause();
});
socket.on("drain", () => process.stdin.resume());
process.stdin.on("end", () => socket.end());
})().catch((error) => {
console.error(error && error.stack ? error.stack : String(error));
process.exit(1);
});`;
}
function windowsNativeBridgeInstallScript(): string {
const bridgeBase64 = base64Text(windowsNativeBridgeServerSource());
return `mkdir -p ${shellQuote(ctx().config.windowsNativeCodexBridgeDir)}
printf %s ${shellQuote(bridgeBase64)} | base64 -d > ${shellQuote(resolve(ctx().config.windowsNativeCodexBridgeDir, "bridge.py"))}
chmod 700 ${shellQuote(resolve(ctx().config.windowsNativeCodexBridgeDir, "bridge.py"))}`;
}
function remoteCodexRuntimePrepareScript(plan: DevContainerPlan): string {
return `set -euo pipefail
CONTAINER=${shellQuote(plan.containerName)}
@@ -500,12 +834,56 @@ function remoteAppServerCommand(task: QueueTask): string {
return `docker exec -i ${shellQuote(plan.containerName)} bash -lc ${shellQuote(inner)}`;
}
function windowsNativeSessionName(task: QueueTask): string {
return safeDockerName(`${task.id}-attempt-${task.currentAttempt || 0}`).slice(0, 120);
}
function windowsNativeAppServerCommand(task: QueueTask): string {
const plan = buildDevContainerPlan(task.providerId, { workdir: remoteHostWorkdirForTask(task) });
const session = windowsNativeSessionName(task);
const sessionDir = resolve(ctx().config.windowsNativeCodexBridgeDir, "sessions", session);
const portFile = resolve(sessionDir, "port");
const pidFile = resolve(sessionDir, "pid");
const stdoutLog = resolve(sessionDir, "codex.stdout.jsonl");
const stderrLog = resolve(sessionDir, "codex.stderr.log");
const serverLog = resolve(sessionDir, "bridge.log");
const bridgePy = resolve(ctx().config.windowsNativeCodexBridgeDir, "bridge.py");
const clientSource = windowsNativeContainerClientSource();
const install = windowsNativeBridgeInstallScript();
const idleSeconds = Math.max(30, Math.floor(ctx().config.windowsNativeCodexIdleTimeoutMs / 1000));
const initialConnectSeconds = Math.max(45, Math.min(120, idleSeconds));
return [
"set -euo pipefail",
`export PATH="$HOME/.local/bin:$PATH"`,
`CWD=${shellQuote(task.cwd)}`,
`case "$CWD" in /mnt/*) ;; *) echo "windows-native executionMode requires cwd under /mnt/<drive>, got $CWD" >&2; exit 64 ;; esac`,
`test -d "$CWD" || { echo "windows-native cwd does not exist: $CWD" >&2; exit 64; }`,
install,
`SESSION_DIR=${shellQuote(sessionDir)}`,
`PORT_FILE=${shellQuote(portFile)}`,
`PID_FILE=${shellQuote(pidFile)}`,
`KEY_DIR=${shellQuote(plan.keyDir)}`,
`[ ! -r "$KEY_DIR/codex-env" ] || { set -a; . "$KEY_DIR/codex-env"; set +a; }`,
`mkdir -p "$SESSION_DIR"`,
`if [ -s "$PID_FILE" ] && kill -0 "$(cat "$PID_FILE")" >/dev/null 2>&1 && [ -s "$PORT_FILE" ]; then :; else rm -f "$PORT_FILE" "$PID_FILE"; nohup python3 ${shellQuote(bridgePy)} --session ${shellQuote(session)} --cwd "$CWD" --codex-command ${shellQuote(ctx().config.windowsNativeCodexCommand)} --bind-host 0.0.0.0 --port 0 --port-file "$PORT_FILE" --pid-file "$PID_FILE" --idle-timeout ${shellQuote(String(idleSeconds))} --initial-connect-timeout ${shellQuote(String(initialConnectSeconds))} --stdout-log ${shellQuote(stdoutLog)} --stderr-log ${shellQuote(stderrLog)} > ${shellQuote(serverLog)} 2>&1 & fi`,
`for i in $(seq 1 120); do [ -s "$PORT_FILE" ] && break; sleep 0.25; done`,
`if [ ! -s "$PORT_FILE" ]; then echo "windows-native Codex bridge did not become ready; log follows:" >&2; tail -80 ${shellQuote(serverLog)} >&2 2>/dev/null || true; exit 65; fi`,
`PORT=$(cat "$PORT_FILE")`,
`WSL_HOSTS=$(hostname -I 2>/dev/null | tr ' ' ',' | sed 's/,,*/,/g; s/^,//; s/,$//')`,
`echo "windows_native_codex_stdio provider=${task.providerId} session=${session} container=${plan.containerName} port=$PORT cwd=$CWD" >&2`,
`CLIENT_SOURCE=${shellQuote(clientSource)}`,
`CLIENT_CMD='if command -v node >/dev/null 2>&1; then exec node -e "$UNIDESK_WINDOWS_NATIVE_CODEX_CLIENT_SOURCE"; fi; if command -v bun >/dev/null 2>&1; then exec bun -e "$UNIDESK_WINDOWS_NATIVE_CODEX_CLIENT_SOURCE"; fi; echo "windows-native relay container requires node or bun" >&2; exit 127'`,
`docker exec -i -e UNIDESK_WINDOWS_NATIVE_CODEX_HOST=${shellQuote(ctx().config.windowsNativeCodexConnectHost)} -e UNIDESK_WINDOWS_NATIVE_CODEX_PORT="$PORT" -e UNIDESK_WINDOWS_NATIVE_CODEX_EXTRA_HOSTS="$WSL_HOSTS" -e UNIDESK_WINDOWS_NATIVE_CODEX_CONNECT_TIMEOUT_MS=60000 -e UNIDESK_WINDOWS_NATIVE_CODEX_CLIENT_SOURCE="$CLIENT_SOURCE" ${shellQuote(plan.containerName)} bash -lc "$CLIENT_CMD"`,
].join("; ");
}
export {
buildDevContainerPlan,
containerTunnelStartScript,
defaultWorkdirForProvider,
devContainerPingScript,
executionModeOptions,
executionProviderOptions,
masterKeyReadScript,
masterKeySetupScript,
@@ -525,4 +903,5 @@ export {
runCodeQueueSsh,
shellQuote,
throwIfCommandFailed,
windowsNativeAppServerCommand,
};
@@ -1,9 +1,9 @@
// 重构前 index.ts 只读参考:commit 6a04144d3f5103014f75b637d7e6bc2f45bf007fblob 56e590c1a6b5ca7ad128bf2c992f60e46c355a58;可用 `git show 6a04144d3f5103014f75b637d7e6bc2f45bf007f:src/components/microservices/code-queue/src/index.ts` 查看。
import postgres from "postgres";
import { codeAgentPortForModel, codeAgentPortInfo, codeModelPorts as codeModelPortsFor, opencodeModels as opencodeModelsFor } from "./code-agent/common";
import { codeAgentPortForModel, codeAgentPortInfo, codeExecutionModeInfo, codeExecutionModes, codeModelPorts as codeModelPortsFor, opencodeModels as opencodeModelsFor } from "./code-agent/common";
import { claudeQqNotificationOutboxStats, notificationTargetConfigured, notificationTargetLabel } from "./notifications";
import { executionProviderOptions } from "./provider-runtime";
import { executionModeOptions, executionProviderOptions } from "./provider-runtime";
import { taskFullOutput } from "./task-output";
import { applyOaTraceStatsToTaskJson, taskScopeId, type OaTraceStats } from "./oa-events";
import { buildCompactTaskTranscript, buildTaskTranscript, cachedPreviewTranscript, fullTranscript, prefixPreview, safePreview, statsDaysFromUrl, taskForCompactMetaResponse, taskForMetaResponse, taskStatisticsSummary, taskTiming, timestampMs } from "./task-view";
@@ -12,7 +12,7 @@ import type { ActiveRun, ActiveRunSlotWaiter } from "./code-agent/common";
import type { JsonValue, QueueRecord, QueuedStatusReason, QueueTask, RuntimeConfig, TaskStatus, TranscriptLine } from "./types";
export interface QueueApiContext {
config: Pick<RuntimeConfig, "codeModels" | "codexModels" | "codexSqliteLogExportBatchSize" | "codexSqliteLogExportEnabled" | "codexSqliteLogExportIntervalMs" | "codexSqliteLogMaxBytes" | "defaultModel" | "defaultReasoningEffort" | "defaultWorkdir" | "judgeMaxTokens" | "judgeRepairAttempts" | "mainProviderId" | "maxActiveQueues" | "maxInMemoryEventRecords" | "maxInMemoryOutputRecords" | "minimaxApiKey" | "minimaxModel" | "modelReasoningEfforts" | "notifyClaudeQqBaseUrl" | "notifyClaudeQqEnabled" | "notifyClaudeQqMaxResponseChars" | "notifyClaudeQqRetryIntervalMs" | "notifyClaudeQqSendAttempts" | "notifyClaudeQqTargetType" | "notifyClaudeQqTimeoutMs" | "outputArchiveDir" | "remoteDefaultWorkdir">;
config: Pick<RuntimeConfig, "codeModels" | "codexModels" | "codexSqliteLogExportBatchSize" | "codexSqliteLogExportEnabled" | "codexSqliteLogExportIntervalMs" | "codexSqliteLogMaxBytes" | "defaultModel" | "defaultReasoningEffort" | "defaultWorkdir" | "judgeMaxTokens" | "judgeRepairAttempts" | "mainProviderId" | "maxActiveQueues" | "maxInMemoryEventRecords" | "maxInMemoryOutputRecords" | "minimaxApiKey" | "minimaxModel" | "modelReasoningEfforts" | "notifyClaudeQqBaseUrl" | "notifyClaudeQqEnabled" | "notifyClaudeQqMaxResponseChars" | "notifyClaudeQqRetryIntervalMs" | "notifyClaudeQqSendAttempts" | "notifyClaudeQqTargetType" | "notifyClaudeQqTimeoutMs" | "outputArchiveDir" | "remoteDefaultWorkdir" | "windowsNativeCodexDefaultWorkdir">;
activeRunSlotQueueIds: () => string[];
activeRunSlotWaiterSummaries: () => JsonValue[];
activeRuns: Map<string, ActiveRun>;
@@ -186,6 +186,8 @@ function taskForListResponse(task: QueueTask, lite = false, queueTasks?: QueueTa
truncated: task.referenceInjection.truncated,
},
providerId: task.providerId,
executionMode: task.executionMode,
executionModeInfo: codeExecutionModeInfo(task.executionMode),
cwd: task.cwd,
model: task.model,
agentPort: codeAgentPortForModel(task.model),
@@ -240,6 +242,8 @@ function taskForListResponse(task: QueueTask, lite = false, queueTasks?: QueueTa
referenceTaskIds: task.referenceTaskIds,
referenceInjection: task.referenceInjection,
providerId: task.providerId,
executionMode: task.executionMode,
executionModeInfo: codeExecutionModeInfo(task.executionMode),
cwd: task.cwd,
model: task.model,
agentPort: codeAgentPortForModel(task.model),
@@ -376,6 +380,8 @@ function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks()
codexModels: ctx().config.codexModels,
opencodeModels: opencodeModelsFor(ctx().config.codeModels),
modelPorts: codeModelPortsFor(ctx().config.codeModels) as unknown as JsonValue,
executionModes: executionModeOptions(),
executionModeInfo: Object.fromEntries(codeExecutionModes.map((mode) => [mode, codeExecutionModeInfo(mode)])) as unknown as JsonValue,
agentPorts: {
codex: codeAgentPortInfo("codex"),
opencode: codeAgentPortInfo("opencode"),
@@ -386,6 +392,7 @@ function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks()
mainProviderId: ctx().config.mainProviderId,
defaultWorkdir: ctx().config.defaultWorkdir,
remoteDefaultWorkdir: ctx().config.remoteDefaultWorkdir,
windowsNativeCodexDefaultWorkdir: ctx().config.windowsNativeCodexDefaultWorkdir,
maxActiveQueues: ctx().config.maxActiveQueues,
executionProviders: executionProviderOptions(),
defaultWorkdirByProvider: Object.fromEntries((executionProviderOptions() as Array<Record<string, JsonValue>>).map((provider) => [String(provider.id), provider.defaultWorkdir ?? ctx().config.defaultWorkdir])) as JsonValue,
@@ -586,6 +593,7 @@ function taskMatchesSearch(task: QueueTask, terms: string[]): boolean {
queued.queuedReason?.code ?? "",
queued.queuedReason?.message ?? "",
task.providerId,
task.executionMode,
task.cwd,
task.model,
task.reasoningEffort ?? "",
@@ -56,6 +56,7 @@ function referenceSummaryItem(task: QueueTask, round: number, roundIndex: number
viaTaskId,
status: task.status,
providerId: task.providerId,
executionMode: task.executionMode,
model: task.model,
cwd: task.cwd,
createdAt: task.createdAt,
@@ -56,6 +56,7 @@ function testTask(id: string, prompt: string, finalResponse: string, referenceTa
cwd: ctx().config.defaultWorkdir,
model: ctx().config.defaultModel,
reasoningEffort: ctx().resolveReasoningEffort(ctx().config.defaultModel, ctx().config.defaultReasoningEffort),
executionMode: "default",
maxAttempts: 1,
status: "succeeded",
createdAt,
@@ -344,6 +345,28 @@ function runTracePortSelfTest(): JsonValue {
assertReferenceTest(String(edited.bodyPreview || "").includes("+const after = true;"), "opencode edit should use metadata diff for line diff display");
assertReferenceTest(!transcript.some((line) => line.status === "opencode/step-start" || line.status === "opencode/step-finish"), "opencode step boundaries should stay out of trace");
assertReferenceTest(!transcript.some((line) => String(line.bodyPreview || "").includes("<think>hidden reasoning</think>")), "reasoning-only opencode assistant text should not duplicate reasoning");
const codexTask = testTask("codex_5002_interleaved_command", "codex command prompt", "", [], "2026-05-12T00:02:00.000Z");
codexTask.output = [
{ seq: 10, at: "2026-05-12T00:02:00.000Z", channel: "command", method: "item/started", itemId: "call_long", text: "item/started: /bin/bash -lc \"python3 - <<'PY'\\nprint('hello')\\nPY\" status=inProgress\n" },
{ seq: 11, at: "2026-05-12T00:02:01.000Z", channel: "command", method: "item/commandExecution/outputDelta", itemId: "call_long", text: "first line\n" },
{ seq: 12, at: "2026-05-12T00:02:02.000Z", channel: "system", method: "startup", text: "Service restarted while task was active; task queued for retry\n" },
{ seq: 13, at: "2026-05-12T00:02:03.000Z", channel: "command", method: "item/commandExecution/outputDelta", itemId: "call_long", text: "second line\n" },
{ seq: 14, at: "2026-05-12T00:02:04.000Z", channel: "command", method: "item/started", itemId: "call_other", text: "item/started: /bin/bash -lc \"git status --short\" status=inProgress\n" },
{ seq: 15, at: "2026-05-12T00:02:05.000Z", channel: "command", method: "item/completed", itemId: "call_other", text: "item/completed: /bin/bash -lc \"git status --short\" status=completed\n" },
{ seq: 16, at: "2026-05-12T00:02:06.000Z", channel: "command", method: "item/commandExecution/outputDelta", itemId: "call_long", text: "third line\n" },
{ seq: 17, at: "2026-05-12T00:02:07.000Z", channel: "command", method: "item/completed", itemId: "call_long", text: "item/completed: /bin/bash -lc \"python3 - <<'PY'\\nprint('hello')\\nPY\" status=completed\n" },
];
const codexTranscript = buildTaskTranscript(codexTask, 20, 0);
const longCommandLines = codexTranscript.filter((line) => line.rawSeqs.includes(11) || line.rawSeqs.includes(13) || line.rawSeqs.includes(16));
assertReferenceTest(longCommandLines.length === 1, "interleaved command output deltas should stay in one trace command line");
const longCommand = longCommandLines[0];
if (longCommand === undefined) throw new Error("interleaved command trace line missing");
assertReferenceTest(longCommand.kind === "ran", "interleaved shell command should be a ran trace line");
assertReferenceTest(String(longCommand.commandPreview || "").includes("python3 - <<"), "interleaved command trace line should retain command preview");
assertReferenceTest(["first line", "second line", "third line"].every((part) => String(longCommand.bodyPreview || "").includes(part)), "interleaved command trace line should aggregate all output chunks");
assertReferenceTest(transcriptLineSummaryLines(longCommand).some((line) => line.includes("$ python3 - <<")), "interleaved command summary should expose the command before expansion");
const remoteTask = testTask("codex_5001_remote_opencode", "remote command prompt", "", [], "2026-05-12T00:01:00.000Z");
remoteTask.providerId = "D601";
remoteTask.cwd = "/home/ubuntu";
@@ -363,6 +386,8 @@ function runTracePortSelfTest(): JsonValue {
{ name: "partial_opencode_tool_to_explored", ok: true, title: partial?.title ?? null },
{ name: "step_boundaries_filtered", ok: true },
{ name: "reasoning_duplicate_filtered", ok: true },
{ name: "interleaved_command_output_single_trace_line", ok: true, rawSeqs: longCommand?.rawSeqs ?? [] },
{ name: "interleaved_command_summary_has_command", ok: true, summaryLines: longCommand ? transcriptLineSummaryLines(longCommand) : [] },
{ name: "duration_preserved", ok: true, durationMs: explored?.durationMs ?? null },
{ name: "remote_opencode_exec_includes_binary", ok: true },
{ name: "opencode_exit0_final_without_step_finish_is_terminal", ok: true },
@@ -16,7 +16,13 @@ export interface TaskOutputContext {
}
const outputArchiveSeededTasks = new Set<string>();
const archivedOutputCache = new Map<string, { signature: string; output: LiveOutput[] }>();
const archivedOutputCache = new Map<string, { signature: string; output: LiveOutput[]; bytes: number }>();
const archivedOutputCacheMaxEntries = 16;
const archivedOutputCacheMaxBytes = 32 * 1024 * 1024;
const archivedOutputCacheMaxSingleBytes = 8 * 1024 * 1024;
const outputGcByteInterval = 8 * 1024 * 1024;
let archivedOutputCacheBytes = 0;
let outputBytesSinceGc = 0;
let context: TaskOutputContext | null = null;
export function configureTaskOutput(runtimeContext: TaskOutputContext): void {
@@ -58,14 +64,60 @@ function ensureTaskOutputArchiveSeeded(task: QueueTask): void {
function appendOutputArchive(task: QueueTask, output: LiveOutput, op: ArchivedLiveOutput["op"], text: string): void {
try {
archivedOutputCache.delete(task.id);
deleteArchivedOutputCache(task.id);
mkdirSync(ctx().config.outputArchiveDir, { recursive: true });
appendFileSync(taskOutputArchivePath(task.id), serializeArchivedOutput(output, op, text), "utf8");
maybeCollectGarbageAfterOutput(text.length);
} catch (error) {
ctx().logger("error", "codex_output_archive_write_failed", { taskId: task.id, error: ctx().errorToJson(error) });
}
}
function estimateCachedOutputBytes(output: LiveOutput[]): number {
return output.reduce((total, item) => total + String(item.text || "").length * 2 + 256, 0);
}
function deleteArchivedOutputCache(taskId: string): void {
const cached = archivedOutputCache.get(taskId);
if (cached !== undefined) {
archivedOutputCacheBytes = Math.max(0, archivedOutputCacheBytes - cached.bytes);
archivedOutputCache.delete(taskId);
}
}
function pruneArchivedOutputCache(requiredBytes = 0): void {
while (
archivedOutputCache.size > archivedOutputCacheMaxEntries
|| archivedOutputCacheBytes + requiredBytes > archivedOutputCacheMaxBytes
) {
const firstKey = archivedOutputCache.keys().next().value;
if (typeof firstKey !== "string") break;
deleteArchivedOutputCache(firstKey);
}
}
function setArchivedOutputCache(taskId: string, signature: string, output: LiveOutput[]): void {
const bytes = estimateCachedOutputBytes(output);
if (bytes > archivedOutputCacheMaxSingleBytes) return;
deleteArchivedOutputCache(taskId);
pruneArchivedOutputCache(bytes);
if (bytes > archivedOutputCacheMaxBytes) return;
archivedOutputCache.set(taskId, { signature, output, bytes });
archivedOutputCacheBytes += bytes;
pruneArchivedOutputCache();
}
function maybeCollectGarbageAfterOutput(textBytes: number): void {
outputBytesSinceGc += Math.max(0, textBytes);
if (outputBytesSinceGc < outputGcByteInterval) return;
outputBytesSinceGc = 0;
try {
(Bun as unknown as { gc?: (force?: boolean) => void }).gc?.(false);
} catch (error) {
ctx().logger("debug", "codex_output_gc_failed", { error: ctx().errorToJson(error) });
}
}
function archiveRecordToOutput(value: unknown): ArchivedLiveOutput | null {
if (typeof value !== "object" || value === null || Array.isArray(value)) return null;
const record = value as Record<string, unknown>;
@@ -114,11 +166,7 @@ function archivedTaskOutput(task: QueueTask): LiveOutput[] {
return [];
}
const output = Array.from(bySeq.values()).sort((left, right) => Number(left.seq) - Number(right.seq));
archivedOutputCache.set(task.id, { signature, output });
while (archivedOutputCache.size > 80) {
const firstKey = archivedOutputCache.keys().next().value;
if (typeof firstKey === "string") archivedOutputCache.delete(firstKey);
}
setArchivedOutputCache(task.id, signature, output);
return output;
}
@@ -18,7 +18,7 @@ import type {
TranscriptKind,
TranscriptLine,
} from "./types";
import { codeAgentPortForModel, codeAgentPortInfo, extractRecord } from "./code-agent/common";
import { codeAgentPortForModel, codeAgentPortInfo, codeExecutionModeInfo, extractRecord } from "./code-agent/common";
import { currentTaskPromptMarker, resolvedReferenceContextTitle, stripCodeQueueEnvironmentHint, userPromptForDisplay } from "./prompts";
import { outputArchiveSignature, taskFullOutput } from "./task-output";
import { retryPrompt } from "./judge";
@@ -948,25 +948,53 @@ function buildTaskTranscript(task: QueueTask, limit = 180, rawOutputWindow = 0,
const promptHistoryLines = promptHistoryTranscriptLines(task, fullText);
const promptHistorySeqs = new Set(promptHistoryLines.map((line) => line.seq));
entries.push(...promptHistoryLines);
let activeCommand: { seq: number; at: string; command: string; status?: string; body: string; rawSeqs: number[]; itemId?: string } | null = null;
type ActiveCommand = { seq: number; at: string; command: string; status?: string; body: string; rawSeqs: number[]; itemId?: string };
let activeCommand: ActiveCommand | null = null;
const activeCommandsByItemId = new Map<string, ActiveCommand>();
const flushCommand = (): void => {
if (activeCommand === null) return;
const kind = commandKind(activeCommand.command);
const output = activeCommand.itemId === undefined ? null : commandOutputs.get(activeCommand.itemId) ?? null;
const body = activeCommand.body.length > 0 ? activeCommand.body : formatCommandOutput(output);
const pushRawSeq = (command: ActiveCommand, seq: number): void => {
if (!command.rawSeqs.includes(seq)) command.rawSeqs.push(seq);
};
const flushCommand = (command: ActiveCommand | null = activeCommand): void => {
if (command === null) return;
const kind = commandKind(command.command);
const output = command.itemId === undefined ? null : commandOutputs.get(command.itemId) ?? null;
const body = command.body.length > 0 ? command.body : formatCommandOutput(output);
const title = command.command.trim().length > 0 ? shortCommandTitle(command.command) : "Command output";
entries.push(addCommandOutputStreams(transcriptLine(
kind,
activeCommand.at,
activeCommand.seq,
shortCommandTitle(activeCommand.command),
activeCommand.rawSeqs,
command.at,
command.seq,
title,
command.rawSeqs,
body,
activeCommand.command,
activeCommand.status,
command.command,
command.status,
fullText,
), output, fullText));
activeCommand = null;
if (command === activeCommand) activeCommand = null;
if (command.itemId !== undefined && activeCommandsByItemId.get(command.itemId) === command) {
activeCommandsByItemId.delete(command.itemId);
}
};
const itemIdCommand = (item: LiveOutput, parsed: { command: string; status?: string } | null = null): ActiveCommand | null => {
if (typeof item.itemId !== "string" || item.itemId.length === 0) return null;
let command = activeCommandsByItemId.get(item.itemId);
if (command === undefined) {
command = {
seq: item.seq,
at: item.at,
command: parsed?.command ?? "",
status: parsed?.status,
body: "",
rawSeqs: [],
itemId: item.itemId,
};
activeCommandsByItemId.set(item.itemId, command);
}
return command;
};
const outputSource = rawOutputWindow > 0 ? task.output : taskFullOutput(task);
@@ -984,8 +1012,18 @@ function buildTaskTranscript(task: QueueTask, limit = 180, rawOutputWindow = 0,
if (item.channel === "user" && item.method === "turn/steer" && promptHistorySeqs.has(item.seq)) continue;
if (isOpenCodeStepBoundaryMethod(item.method)) continue;
if (item.channel === "command" && item.method === "item/started") {
flushCommand();
const parsed = parseCommandLine(item.text);
const groupedCommand = itemIdCommand(item, parsed);
if (groupedCommand !== null) {
flushCommand();
groupedCommand.seq = item.seq;
groupedCommand.at = item.at;
groupedCommand.command = parsed?.command || item.text;
groupedCommand.status = parsed?.status;
pushRawSeq(groupedCommand, item.seq);
continue;
}
flushCommand();
activeCommand = {
seq: item.seq,
at: item.at,
@@ -998,9 +1036,13 @@ function buildTaskTranscript(task: QueueTask, limit = 180, rawOutputWindow = 0,
continue;
}
if (item.channel === "command" && item.method === "item/commandExecution/outputDelta") {
if (activeCommand !== null) {
const groupedCommand = itemIdCommand(item);
if (groupedCommand !== null) {
groupedCommand.body += item.text;
pushRawSeq(groupedCommand, item.seq);
} else if (activeCommand !== null) {
activeCommand.body += item.text;
activeCommand.rawSeqs.push(item.seq);
pushRawSeq(activeCommand, item.seq);
} else {
const output = typeof item.itemId === "string" ? commandOutputs.get(item.itemId) ?? null : null;
entries.push(addCommandOutputStreams(transcriptLine("ran", item.at, item.seq, "Command output", [item.seq], item.text || formatCommandOutput(output), "", undefined, fullText), output, fullText));
@@ -1009,9 +1051,15 @@ function buildTaskTranscript(task: QueueTask, limit = 180, rawOutputWindow = 0,
}
if (item.channel === "command" && item.method === "item/completed") {
const parsed = parseCommandLine(item.text);
if (activeCommand !== null) {
const groupedCommand = itemIdCommand(item, parsed);
if (groupedCommand !== null) {
if (parsed?.command) groupedCommand.command = parsed.command;
groupedCommand.status = parsed?.status ?? groupedCommand.status;
pushRawSeq(groupedCommand, item.seq);
flushCommand(groupedCommand);
} else if (activeCommand !== null) {
activeCommand.status = parsed?.status ?? activeCommand.status;
activeCommand.rawSeqs.push(item.seq);
pushRawSeq(activeCommand, item.seq);
flushCommand();
} else {
const command = parsed?.command || item.text;
@@ -1049,73 +1097,14 @@ function buildTaskTranscript(task: QueueTask, limit = 180, rawOutputWindow = 0,
}
}
flushCommand();
for (const command of Array.from(activeCommandsByItemId.values()).sort((left, right) => left.seq - right.seq)) {
flushCommand(command);
}
return boundedTranscript(entries, limit);
}
function compactTaskTranscriptLine(item: LiveOutput, title: string, kind: TranscriptKind): TranscriptLine {
return {
seq: item.seq,
at: item.at,
kind,
title,
status: item.method,
bodyPreview: prefixPreview(item.text.replace(/\u001b\[[0-9;]*m/gu, ""), 900),
rawSeqs: [item.seq],
};
}
function buildCompactTaskTranscript(task: QueueTask, limit = 12, rawOutputWindow = 24): TranscriptLine[] {
const entries: TranscriptLine[] = [];
for (const item of task.promptHistory.slice(-2)) {
entries.push({
seq: item.seq,
at: item.at,
kind: "message",
title: "Steer prompt",
status: item.method,
bodyPreview: prefixPreview(item.text, 900),
rawSeqs: [item.seq],
});
}
const outputItems = task.output.slice(-rawOutputWindow);
const fileChangeInputs = outputItems.some((item) => item.channel === "diff" && item.method === "item/fileChange/outputDelta" && typeof item.itemId === "string")
? codexSessionFileChangesByCallId(task)
: new Map<string, SessionFileChange>();
for (const item of outputItems) {
if (item.channel === "user" && item.method === "enqueue") continue;
if (isOpenCodeStepBoundaryMethod(item.method)) continue;
if (item.channel === "command") {
const isOutput = item.method === "item/commandExecution/outputDelta";
entries.push({
seq: item.seq,
at: item.at,
kind: isOutput ? "ran" : item.method === "item/started" ? "ran" : "system",
title: isOutput ? "Command output" : item.method === "item/started" ? "Command started" : "Command completed",
status: item.method,
commandPreview: isOutput ? undefined : prefixPreview(item.text, 900),
bodyPreview: isOutput ? prefixPreview(item.text, 900) : undefined,
rawSeqs: [item.seq],
});
} else if (item.channel === "diff") {
entries.push(compactTaskTranscriptLine({ ...item, text: fileChangeTextWithInlinePatch(item, fileChangeInputs) }, "Edited files", "edited"));
} else if (item.channel === "error") {
entries.push(compactTaskTranscriptLine(item, "Error", "error"));
} else if (item.channel === "assistant" || item.channel === "reasoning" || item.channel === "user") {
const body = item.channel === "assistant" && String(item.method || "").startsWith("opencode/") ? openCodeVisibleAssistantText(item.text) : item.text;
if (body.length > 0) entries.push(compactTaskTranscriptLine({ ...item, text: body }, item.channel === "assistant" ? "Assistant message" : item.channel === "reasoning" ? "Reasoning" : "User prompt", "message"));
} else if (item.channel === "tool" && String(item.method || "").startsWith("opencode/")) {
const line = openCodeToolTranscriptLine(item, false);
entries.push(line ?? compactTaskTranscriptLine(item, "OpenCode tool", "system"));
} else {
const title = item.method === "judge"
? "Judge result"
: item.method === "startup" || item.method === "shutdown"
? "Recovered thread execution"
: "System";
entries.push(compactTaskTranscriptLine(item, title, "system"));
}
}
return boundedTranscript(entries, limit);
return buildTaskTranscript(task, limit, rawOutputWindow, false);
}
function transcriptSignature(task: QueueTask): string {
@@ -1269,6 +1258,8 @@ function taskForMetaResponse(task: QueueTask): JsonValue {
referenceTaskIds: task.referenceTaskIds,
referenceInjection: task.referenceInjection,
providerId: task.providerId,
executionMode: task.executionMode,
executionModeInfo: codeExecutionModeInfo(task.executionMode),
cwd: task.cwd,
model: task.model,
agentPort: codeAgentPortForModel(task.model),
@@ -1337,6 +1328,8 @@ function taskForCompactMetaResponse(task: QueueTask): JsonValue {
truncated: task.referenceInjection.truncated,
},
providerId: task.providerId,
executionMode: task.executionMode,
executionModeInfo: codeExecutionModeInfo(task.executionMode),
cwd: task.cwd,
model: task.model,
agentPort: codeAgentPortForModel(task.model),
@@ -1578,11 +1571,37 @@ function parseJudgeLine(text: string): JudgeResult | null {
if (match === null) return null;
const confidence = Number(match[2]);
const source = match[3] === "minimax" ? "minimax" : "fallback";
const rawReason = (match[4] ?? "").trim();
const detailsMatch = /\bMiniMax failure details:\s+([\s\S]*)$/u.exec(rawReason);
const detailsText = detailsMatch?.[1] ?? "";
const detailValue = (key: string): string | null => new RegExp(`\\b${key}=([^\\s]+)`, "u").exec(detailsText)?.[1] ?? null;
const detailNumber = (key: string): number | undefined => {
const value = Number(detailValue(key));
return Number.isFinite(value) ? value : undefined;
};
const errorMessage = /\berror=([\s\S]*)$/u.exec(detailsText)?.[1]?.trim() ?? "";
const failureDetails = detailsText.length === 0 ? null : {
provider: "minimax" as const,
stage: (detailValue("stage") || "unknown") as JudgeResult["failureDetails"] extends infer T ? T extends { stage: infer S } ? S : never : never,
model: "",
apiBase: "",
occurredAt: "",
durationMs: detailNumber("durationMs") ?? 0,
timeoutMs: detailNumber("timeoutMs") ?? 0,
timedOut: detailValue("timedOut") === "true",
errorName: detailValue("errorName") || "",
errorMessage,
promptChars: detailNumber("promptChars"),
promptLines: detailNumber("promptLines"),
payloadBytes: detailNumber("payloadBytes"),
responseStatus: detailNumber("responseStatus") ?? null,
};
return {
decision: match[1] as JudgeDecision,
confidence: Number.isFinite(confidence) ? confidence : 0,
source,
reason: (match[4] ?? "").trim(),
reason: detailsMatch === null ? rawReason : rawReason.slice(0, detailsMatch.index).trim(),
failureDetails,
};
}
@@ -1680,6 +1699,23 @@ function isRecoveredThreadLine(line: TranscriptLine): boolean {
);
}
function traceSystemLineIsError(line: TranscriptLine): boolean {
const text = [
line.title,
line.status,
line.bodyPreview,
line.commandPreview,
line.stderrPreview,
line.stdoutPreview,
].map((value) => String(value || "")).join("\n");
return /\b(error|failed|failure|interrupt|interrupted|cancell?ed|watchdog|timeout|closed|refused|aborted|exception)\b/iu.test(text);
}
function traceLineVisibleInTraceView(line: TranscriptLine): boolean {
if (line.title === "Submitted prompt") return false;
return line.kind !== "system" || traceSystemLineIsError(line);
}
function mergeTraceWindowLines(left: TranscriptLine[], right: TranscriptLine[]): TranscriptLine[] {
const seen = new Set<string>();
const merged: TranscriptLine[] = [];
@@ -1990,6 +2026,8 @@ function taskTraceSummaryResponse(task: QueueTask, oaTraceStats: JsonValue | nul
queueId: ctx().queueIdOf(task),
status: task.status,
providerId: task.providerId,
executionMode: task.executionMode,
executionModeInfo: codeExecutionModeInfo(task.executionMode),
model: task.model,
agentPort: codeAgentPortForModel(task.model),
agentPortInfo: codeAgentPortInfo(codeAgentPortForModel(task.model)),
@@ -2083,7 +2121,7 @@ function taskTraceStepsResponse(task: QueueTask, url: URL): Response {
const previewTranscript = cachedPreviewTranscript(task);
const attemptWindow = attemptIndex === null ? null : traceAttemptWindows(task, previewTranscript).find((window) => window.index === attemptIndex) ?? null;
const sourceTranscript = attemptWindow === null ? previewTranscript : executionLinesForAttempt(attemptWindow.lines);
const transcript = sourceTranscript.filter((line) => line.title !== "Submitted prompt");
const transcript = sourceTranscript.filter(traceLineVisibleInTraceView);
const page = ctx().pageBySeq(transcript, url, limit);
return ctx().jsonResponse({
ok: true,
@@ -2121,7 +2159,7 @@ function taskTraceStepDetailResponse(task: QueueTask, url: URL): Response {
const seq = Number(url.searchParams.get("seq"));
if (!Number.isFinite(seq)) return ctx().jsonResponse({ ok: false, error: "seq is required" }, 400);
const agentPort = codeAgentPortForModel(task.model);
const transcript = fullTranscript(task).filter((line) => line.title !== "Submitted prompt");
const transcript = fullTranscript(task).filter(traceLineVisibleInTraceView);
const line = transcript.find((item) => Number(item.seq) === seq || item.rawSeqs.includes(seq));
if (line === undefined) return ctx().jsonResponse({ ok: false, error: "trace step not found", seq }, 404);
return ctx().jsonResponse({
@@ -2145,6 +2183,8 @@ function taskSummaryResponse(task: QueueTask, url: URL): JsonValue {
queueId: ctx().queueIdOf(task),
status: task.status,
providerId: task.providerId,
executionMode: task.executionMode,
executionModeInfo: codeExecutionModeInfo(task.executionMode),
model: task.model,
agentPort: codeAgentPortForModel(task.model),
agentPortInfo: codeAgentPortInfo(codeAgentPortForModel(task.model)),
@@ -20,6 +20,8 @@ export type RunMode = "initial" | "retry";
export type JudgeDecision = "complete" | "retry" | "fail";
export type CodeExecutionMode = "default" | "windows-native";
export type OutputChannel = "system" | "user" | "assistant" | "reasoning" | "command" | "diff" | "tool" | "error";
export type TerminalStatus = "completed" | "interrupted" | "failed" | null;
@@ -81,6 +83,11 @@ export interface RuntimeConfig {
devContainerDefaultProviderId: string;
devContainerImage: string;
devContainerWorkdir: string;
windowsNativeCodexBridgeDir: string;
windowsNativeCodexCommand: string;
windowsNativeCodexConnectHost: string;
windowsNativeCodexDefaultWorkdir: string;
windowsNativeCodexIdleTimeoutMs: number;
}
export interface QueueTaskRequest {
@@ -90,6 +97,7 @@ export interface QueueTaskRequest {
cwd?: string;
model?: string;
reasoningEffort?: string;
executionMode?: CodeExecutionMode;
maxAttempts?: number;
referenceTaskIds?: string[];
basePrompt?: string;
@@ -145,11 +153,14 @@ export interface AttemptSummary {
mode: RunMode;
startedAt: string;
finishedAt: string;
providerId?: string;
executionMode?: CodeExecutionMode;
terminalStatus: TerminalStatus;
transportClosedBeforeTerminal: boolean;
appServerExitCode: number | null;
appServerSignal: string | null;
error: string | null;
events?: CodexEventSummary[];
inputPrompt?: string;
inputPromptPreview?: string;
inputPromptChars?: number;
@@ -178,9 +189,33 @@ export interface JudgeResult {
reason: string;
continuePrompt?: string;
source: "minimax" | "fallback";
failureDetails?: JudgeFailureDetails | null;
raw?: JsonValue;
}
export interface JudgeFailureDetails {
provider: "minimax";
stage: "request" | "http" | "parse" | "validation" | "unknown";
model: string;
apiBase: string;
occurredAt: string;
durationMs: number;
timeoutMs: number;
timedOut: boolean;
errorName: string;
errorMessage: string;
repairAttempt?: number;
maxRepairAttempts?: number;
promptChars?: number;
promptLines?: number;
payloadBytes?: number;
responseStatus?: number | null;
responseTextPreview?: string;
responseTextChars?: number;
responseContentPreview?: string;
responseContentChars?: number;
}
export interface FeedbackPromptRecord {
text: string;
preview: string;
@@ -199,6 +234,19 @@ export interface ParsedJudgeJson {
export interface MiniMaxJudgeResponse {
rawText: string;
content: string;
diagnostics: {
provider: "minimax";
model: string;
apiBase: string;
durationMs: number;
timeoutMs: number;
promptChars: number;
promptLines: number;
payloadBytes: number;
responseStatus: number;
responseTextChars: number;
responseContentChars: number;
};
}
export interface ReferenceInjectionSummaryItem {
@@ -208,6 +256,7 @@ export interface ReferenceInjectionSummaryItem {
viaTaskId: string | null;
status: TaskStatus;
providerId: string;
executionMode?: CodeExecutionMode;
model: string;
cwd: string;
createdAt: string;
@@ -250,6 +299,7 @@ export interface QueueTask {
cwd: string;
model: string;
reasoningEffort: string | null;
executionMode: CodeExecutionMode;
maxAttempts: number;
status: TaskStatus;
createdAt: string;
@@ -643,13 +643,13 @@ async function applyTraceSnapshot(executor: SqlExecutor, event: OaEventRecord, s
subject_kind = EXCLUDED.subject_kind,
subject_id = EXCLUDED.subject_id,
stats_revision = oa_trace_stats.stats_revision + 1,
step_count = GREATEST(oa_trace_stats.step_count, EXCLUDED.step_count),
llm_step_count = GREATEST(oa_trace_stats.llm_step_count, EXCLUDED.llm_step_count),
read_count = GREATEST(oa_trace_stats.read_count, EXCLUDED.read_count),
edit_count = GREATEST(oa_trace_stats.edit_count, EXCLUDED.edit_count),
run_count = GREATEST(oa_trace_stats.run_count, EXCLUDED.run_count),
error_count = GREATEST(oa_trace_stats.error_count, EXCLUDED.error_count),
trace_line_count = GREATEST(oa_trace_stats.trace_line_count, EXCLUDED.trace_line_count),
step_count = EXCLUDED.step_count,
llm_step_count = EXCLUDED.llm_step_count,
read_count = EXCLUDED.read_count,
edit_count = EXCLUDED.edit_count,
run_count = EXCLUDED.run_count,
error_count = EXCLUDED.error_count,
trace_line_count = EXCLUDED.trace_line_count,
output_max_seq = GREATEST(oa_trace_stats.output_max_seq, EXCLUDED.output_max_seq),
attempt_stats_json = EXCLUDED.attempt_stats_json,
last_event_sequence = GREATEST(COALESCE(oa_trace_stats.last_event_sequence, 0), EXCLUDED.last_event_sequence),
@@ -688,7 +688,7 @@ async function applyTraceStep(executor: SqlExecutor, event: OaEventRecord, scope
error_count: string | number;
}[]>`
SELECT
COUNT(*) AS step_count,
COUNT(*) FILTER (WHERE kind <> 'system') AS step_count,
COUNT(*) FILTER (WHERE kind = 'read') AS read_count,
COUNT(*) FILTER (WHERE kind = 'edit') AS edit_count,
COUNT(*) FILTER (WHERE kind = 'run') AS run_count,
@@ -718,13 +718,13 @@ async function applyTraceStep(executor: SqlExecutor, event: OaEventRecord, scope
subject_kind = EXCLUDED.subject_kind,
subject_id = EXCLUDED.subject_id,
stats_revision = oa_trace_stats.stats_revision + 1,
step_count = GREATEST(oa_trace_stats.step_count, EXCLUDED.step_count),
llm_step_count = GREATEST(oa_trace_stats.llm_step_count, EXCLUDED.llm_step_count),
read_count = GREATEST(oa_trace_stats.read_count, EXCLUDED.read_count),
edit_count = GREATEST(oa_trace_stats.edit_count, EXCLUDED.edit_count),
run_count = GREATEST(oa_trace_stats.run_count, EXCLUDED.run_count),
error_count = GREATEST(oa_trace_stats.error_count, EXCLUDED.error_count),
trace_line_count = GREATEST(oa_trace_stats.trace_line_count, EXCLUDED.trace_line_count),
step_count = EXCLUDED.step_count,
llm_step_count = EXCLUDED.llm_step_count,
read_count = EXCLUDED.read_count,
edit_count = EXCLUDED.edit_count,
run_count = EXCLUDED.run_count,
error_count = EXCLUDED.error_count,
trace_line_count = EXCLUDED.trace_line_count,
output_max_seq = GREATEST(oa_trace_stats.output_max_seq, EXCLUDED.output_max_seq),
attempt_stats_json = oa_trace_stats.attempt_stats_json,
last_event_sequence = GREATEST(COALESCE(oa_trace_stats.last_event_sequence, 0), EXCLUDED.last_event_sequence),