feat: add unified OA event flow

This commit is contained in:
Codex
2026-05-14 09:41:55 +00:00
parent b36d7f94d7
commit 9f483b002c
37 changed files with 3417 additions and 502 deletions
+23 -7
View File
@@ -2775,6 +2775,7 @@ async function getMicroserviceAvailabilitySummary(): Promise<Record<string, Json
function microserviceCacheTtlMs(serviceId: string, targetPath: string): number {
if (targetPath === "/health") return 0;
if (targetPath.endsWith("/stream")) return 0;
if (serviceId === "pipeline" && targetPath === "/api/snapshot") return 6_000;
if (serviceId === "pipeline" && targetPath.startsWith("/api/oa-event-flow/")) return 20_000;
if (serviceId === "pipeline" && targetPath.startsWith("/api/model-quota/")) return 60_000;
@@ -2888,22 +2889,36 @@ async function directMicroserviceResponse(
proxyOptions: { query: string; jsonArrayLimits: Record<string, JsonValue> },
requestHeaders: Record<string, JsonValue>,
bodyText: string,
abortSignal?: AbortSignal,
): Promise<Response> {
const baseUrl = new URL(service.backend.nodeBaseUrl);
const upstreamUrl = new URL(targetPath, baseUrl);
upstreamUrl.search = proxyOptions.query;
const headers = headersFromMicroserviceRequest(requestHeaders);
const streamExpected = method === "GET" && (targetPath.endsWith("/stream") || String(headers.get("accept") || "").toLowerCase().includes("text/event-stream"));
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), Math.max(1000, service.backend.timeoutMs));
const timer = streamExpected ? null : setTimeout(() => controller.abort(), Math.max(1000, service.backend.timeoutMs));
try {
const upstream = await fetch(upstreamUrl, {
method,
headers,
body: method === "GET" || method === "HEAD" ? undefined : bodyText,
signal: controller.signal,
signal: streamExpected ? abortSignal : controller.signal,
});
const rawBodyText = await upstream.text();
const upstreamContentType = upstream.headers.get("content-type") ?? "text/plain; charset=utf-8";
if (upstreamContentType.toLowerCase().includes("text/event-stream")) {
const responseHeaders: Record<string, string> = {
"content-type": upstreamContentType,
"cache-control": upstream.headers.get("cache-control") || "no-store, no-transform",
"connection": "keep-alive",
"x-unidesk-proxy-mode": "direct",
"x-unidesk-response-truncated": "false",
};
const buffering = upstream.headers.get("x-accel-buffering");
if (buffering !== null) responseHeaders["x-accel-buffering"] = buffering;
return new Response(upstream.body, { status: upstream.status, headers: responseHeaders });
}
const rawBodyText = await upstream.text();
const limitedBodyText = applyJsonArrayLimits(rawBodyText, upstreamContentType, proxyOptions.jsonArrayLimits);
const bounded = boundedMicroserviceBodyText(limitedBodyText, upstreamContentType, {
serviceId: service.id,
@@ -2922,7 +2937,7 @@ async function directMicroserviceResponse(
} catch (error) {
return jsonResponse({ ok: false, error: "direct microservice proxy failed", serviceId: service.id, detail: errorToJson(error) }, 502);
} finally {
clearTimeout(timer);
if (timer !== null) clearTimeout(timer);
}
}
@@ -2933,9 +2948,10 @@ async function fetchMicroserviceUpstreamResponse(
proxyOptions: { query: string; jsonArrayLimits: Record<string, JsonValue> },
requestHeaders: Record<string, JsonValue>,
bodyText: string,
abortSignal?: AbortSignal,
): Promise<Response> {
if (canDirectProxyMicroservice(service)) {
return directMicroserviceResponse(service, method, targetPath, proxyOptions, requestHeaders, bodyText);
return directMicroserviceResponse(service, method, targetPath, proxyOptions, requestHeaders, bodyText, abortSignal);
}
if (!(await providerSupports(service.providerId, "microservice.http"))) {
return jsonResponse({ ok: false, error: `provider does not declare microservice.http capability: ${service.providerId}` }, 409);
@@ -3028,8 +3044,8 @@ async function microserviceRoute(req: Request, url: URL): Promise<Response> {
return stale;
}
}
const response = await fetchMicroserviceUpstreamResponse(service, method, targetPath, proxyOptions, requestHeaders, bodyText);
if (method === "GET" || method === "HEAD") rememberMicroserviceCache(cacheKey, cacheTtlMs, await cacheableResponseSnapshot(response));
const response = await fetchMicroserviceUpstreamResponse(service, method, targetPath, proxyOptions, requestHeaders, bodyText, req.signal);
if ((method === "GET" || method === "HEAD") && cacheTtlMs > 0) rememberMicroserviceCache(cacheKey, cacheTtlMs, await cacheableResponseSnapshot(response));
return response;
}
File diff suppressed because one or more lines are too long
+196 -5
View File
@@ -1422,7 +1422,7 @@ input:focus, select:focus, textarea:focus { border-color: var(--accent-2); }
.result-card dd { margin: 0; }
.result-grid { grid-template-columns: repeat(auto-fill, minmax(280px, 1fr)); }
.microservice-page, .findjob-page, .pipeline-page, .met-page, .code-queue-page, .baidu-netdisk-page, .filebrowser-page {
.microservice-page, .findjob-page, .pipeline-page, .met-page, .code-queue-page, .baidu-netdisk-page, .filebrowser-page, .oa-event-flow-page {
display: grid;
gap: 10px;
}
@@ -1464,6 +1464,130 @@ input:focus, select:focus, textarea:focus { border-color: var(--accent-2); }
gap: 10px;
align-items: start;
}
.oa-flow-hero {
display: grid;
grid-template-columns: minmax(320px, 1fr) minmax(260px, 0.45fr);
gap: 10px;
align-items: stretch;
}
.oa-flow-signal {
position: relative;
display: grid;
gap: 4px;
min-width: 0;
padding: 10px;
overflow: hidden;
border: 1px solid rgba(78, 183, 168, 0.28);
background:
linear-gradient(120deg, rgba(78, 183, 168, 0.14), rgba(215, 161, 58, 0.08)),
repeating-linear-gradient(90deg, rgba(255,255,255,0.04) 0, rgba(255,255,255,0.04) 1px, transparent 1px, transparent 14px),
var(--panel-3);
}
.oa-flow-signal::after {
content: "";
position: absolute;
inset: auto -20% 0 -20%;
height: 2px;
background: linear-gradient(90deg, transparent, var(--accent-2), var(--accent), transparent);
animation: oa-signal-scan 2.8s linear infinite;
}
.oa-flow-signal span {
color: var(--accent);
font-size: 10px;
letter-spacing: 0.18em;
text-transform: uppercase;
}
.oa-flow-signal strong {
min-width: 0;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
font-size: 18px;
}
.oa-flow-signal code { color: var(--muted); }
@keyframes oa-signal-scan {
from { transform: translateX(-40%); opacity: 0.35; }
50% { opacity: 1; }
to { transform: translateX(40%); opacity: 0.35; }
}
.oa-flow-metrics {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(150px, 1fr));
gap: 8px;
}
.oa-filter-bar {
display: grid;
grid-template-columns: minmax(280px, 0.8fr) auto minmax(180px, 0.45fr);
gap: 10px;
align-items: end;
}
.oa-filter-bar label {
display: grid;
gap: 5px;
color: var(--muted);
font-size: 10px;
letter-spacing: 0.16em;
text-transform: uppercase;
}
.oa-filter-bar input {
width: 100%;
min-height: 34px;
padding: 7px 9px;
border: 1px solid var(--line);
color: var(--text);
background: var(--panel-3);
}
.oa-filter-presets {
display: flex;
flex-wrap: wrap;
gap: 6px;
}
.oa-filter-bar > code {
align-self: center;
min-width: 0;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
color: var(--accent-2);
}
.oa-type-strip {
display: flex;
flex-wrap: wrap;
gap: 6px;
margin-top: 10px;
}
.oa-flow-grid {
display: grid;
grid-template-columns: 1fr;
gap: 10px;
}
.oa-flow-wide { min-width: 0; }
.oa-event-table, .oa-stats-table {
min-width: 1120px;
font-size: 12px;
}
.oa-event-table td,
.oa-stats-table td {
vertical-align: top;
}
.oa-tag-rail {
display: flex;
flex-wrap: wrap;
gap: 4px;
max-width: 360px;
}
.oa-tag-rail code {
padding: 2px 5px;
border: 1px solid rgba(78, 183, 168, 0.22);
color: var(--accent-2);
background: rgba(78, 183, 168, 0.07);
}
.oa-payload-preview {
display: block;
max-width: 360px;
color: var(--muted);
overflow-wrap: anywhere;
}
.findjob-grid .panel:nth-child(3) { grid-column: 1 / -1; }
.claudeqq-page .findjob-grid .panel:nth-child(n+3) { grid-column: 1 / -1; }
.filebrowser-hero {
@@ -2356,7 +2480,7 @@ input:focus, select:focus, textarea:focus { border-color: var(--accent-2); }
min-width: 0;
align-items: center;
}
.codex-submit-queue-row select {
.codex-submit-queue-select {
grid-area: queue-select;
}
.codex-rename-queue-btn,
@@ -2376,6 +2500,49 @@ input:focus, select:focus, textarea:focus { border-color: var(--accent-2); }
.codex-merge-queue-btn {
grid-area: merge;
}
.codex-merge-dialog {
width: min(480px, 96vw);
overflow: hidden;
border-color: rgba(78, 183, 168, 0.34);
border-radius: 18px;
background:
radial-gradient(circle at 20% 0%, rgba(78, 183, 168, 0.16), transparent 34%),
#0a1015;
}
.codex-merge-dialog-body {
display: grid;
gap: 12px;
padding: 14px;
}
.codex-merge-dialog-body label {
display: grid;
gap: 6px;
color: rgba(226, 244, 244, 0.82);
font-size: 12px;
font-weight: 800;
letter-spacing: 0.04em;
text-transform: uppercase;
}
.codex-merge-dialog-body select {
width: 100%;
min-height: 40px;
box-sizing: border-box;
}
.codex-merge-dialog-target,
.codex-merge-dialog-note {
margin: 0;
color: rgba(217, 238, 238, 0.76);
line-height: 1.55;
}
.codex-merge-dialog-target code {
color: #eafffb;
}
.codex-merge-dialog-actions {
display: flex;
justify-content: flex-end;
gap: 8px;
flex-wrap: wrap;
}
.codex-reference-field {
display: grid;
gap: 5px;
@@ -5090,6 +5257,29 @@ input:focus, select:focus, textarea:focus { border-color: var(--accent-2); }
padding: 10px;
border-bottom: 1px solid var(--line);
}
.unidesk-dialog {
overflow: hidden;
}
.unidesk-dialog-body {
display: grid;
gap: 10px;
max-height: calc(86vh - 118px);
overflow: auto;
padding: 10px;
}
.unidesk-dialog-actions {
display: flex;
justify-content: flex-end;
gap: 8px;
flex-wrap: wrap;
padding: 10px;
border-top: 1px solid var(--line);
background: rgba(255,255,255,0.02);
}
.unidesk-dialog-body.codex-merge-dialog-body {
gap: 12px;
padding: 14px;
}
.raw-json {
max-height: calc(86vh - 58px);
overflow: auto;
@@ -5168,11 +5358,11 @@ input:focus, select:focus, textarea:focus { border-color: var(--accent-2); }
}
@media (max-width: 1120px) {
.metric-grid, .policy-grid, .security-board, .docker-metrics, .monitor-chart-grid, .monitor-summary-grid, .performance-metric-stack, .codex-load-test-grid, .baidu-doc-grid, .filebrowser-target-grid { grid-template-columns: repeat(2, minmax(0, 1fr)); }
.metric-grid, .policy-grid, .security-board, .docker-metrics, .monitor-chart-grid, .monitor-summary-grid, .performance-metric-stack, .codex-load-test-grid, .baidu-doc-grid, .filebrowser-target-grid, .oa-flow-metrics { grid-template-columns: repeat(2, minmax(0, 1fr)); }
.pipeline-oa-guarantees { grid-template-columns: repeat(2, minmax(0, 1fr)); }
.dispatch-form, .schedule-form { grid-template-columns: 1fr 1fr; }
.dispatch-actions { align-items: center; }
.page-grid, .docker-layout, .monitor-layout, .performance-top-grid, .performance-grid, .findjob-grid, .findjob-hero, .pipeline-grid, .pipeline-hero, .met-grid, .met-form-grid, .code-queue-layout, .code-queue-hero, .codex-detail-grid, .project-manager-hero, .project-manager-layout, .baidu-netdisk-grid, .baidu-netdisk-hero, .baidu-transfer-forms, .filebrowser-hero { grid-template-columns: 1fr; }
.page-grid, .docker-layout, .monitor-layout, .performance-top-grid, .performance-grid, .findjob-grid, .findjob-hero, .pipeline-grid, .pipeline-hero, .met-grid, .met-form-grid, .code-queue-layout, .code-queue-hero, .codex-detail-grid, .project-manager-hero, .project-manager-layout, .baidu-netdisk-grid, .baidu-netdisk-hero, .baidu-transfer-forms, .filebrowser-hero, .oa-flow-hero, .oa-filter-bar { grid-template-columns: 1fr; }
.codex-session-shell { grid-template-columns: minmax(260px, 0.42fr) minmax(0, 1fr); position: relative; }
.codex-session-shell.queue-collapsed { grid-template-columns: minmax(0, 1fr); }
.codex-session-sidebar { border-right: 1px solid var(--line); border-bottom: 0; }
@@ -5198,7 +5388,8 @@ input:focus, select:focus, textarea:focus { border-color: var(--accent-2); }
.pipeline-kv-grid,
.pipeline-field-list,
.performance-metric-stack,
.codex-load-test-grid {
.codex-load-test-grid,
.oa-flow-metrics {
grid-template-columns: 1fr;
}
.performance-hero { flex-direction: column; }
+3
View File
@@ -8,6 +8,7 @@ import { FileBrowserPage } from "./filebrowser";
import { FindJobPage } from "./findjob";
import { MetNonlinearPage } from "./met-nonlinear";
import { canonicalizeKnownRoute, createRouteRegistry, DEFAULT_ACTIVE_TABS, MODULES, pathForTarget, resolveRouteTarget } from "./navigation";
import { OaEventFlowPage } from "./oa-event-flow";
import { PipelinePage } from "./pipeline";
import { ProjectManagerPage } from "./project-manager";
import { TodoNotePage } from "./todo-note";
@@ -1638,6 +1639,7 @@ function MicroserviceCatalogPage({ microservices, onRaw, onNavigate }: AnyRecord
service.id === "met-nonlinear" ? h("button", { type: "button", className: "ghost-btn", onClick: () => onNavigate("apps", "met-nonlinear"), "data-testid": "open-met-nonlinear-button" }, "打开") : null,
service.id === "claudeqq" ? h("button", { type: "button", className: "ghost-btn", onClick: () => onNavigate("apps", "claudeqq"), "data-testid": "open-claudeqq-button" }, "打开") : null,
service.id === "baidu-netdisk" ? h("button", { type: "button", className: "ghost-btn", onClick: () => onNavigate("apps", "baidu-netdisk"), "data-testid": "open-baidu-netdisk-button" }, "打开") : null,
service.id === "oa-event-flow" ? h("button", { type: "button", className: "ghost-btn", onClick: () => onNavigate("apps", "oa-event-flow"), "data-testid": "open-oa-event-flow-button" }, "打开") : null,
service.id === "code-queue" ? h("button", { type: "button", className: "ghost-btn", onClick: () => onNavigate("apps", "code-queue"), "data-testid": "open-code-queue-button" }, "打开") : null,
service.id === "project-manager" ? h("button", { type: "button", className: "ghost-btn", onClick: () => onNavigate("apps", "project-manager"), "data-testid": "open-project-manager-button" }, "打开") : null,
h(RawButton, { title: `用户服务 ${service.id}`, data: service, onOpen: onRaw }),
@@ -2143,6 +2145,7 @@ function WorkArea({ activeModule, activeTab, data, session, refresh, onRaw, onNa
if (activeModule === "apps" && activeTab === "claudeqq") return h(ClaudeQqPage, { microservices: data.microservices, onRaw, apiBaseUrl: cfg.apiBaseUrl });
if (activeModule === "apps" && activeTab === "baidu-netdisk") return h(BaiduNetdiskPage, { microservices: data.microservices, onRaw, apiBaseUrl: cfg.apiBaseUrl });
if (activeModule === "apps" && activeTab === "filebrowser") return h(FileBrowserPage, { microservices: data.microservices, onRaw, apiBaseUrl: cfg.apiBaseUrl });
if (activeModule === "apps" && activeTab === "oa-event-flow") return h(OaEventFlowPage, { microservices: data.microservices, onRaw, apiBaseUrl: cfg.apiBaseUrl });
if (activeModule === "apps" && activeTab === "code-queue") return h(CodeQueuePage, { microservices: data.microservices, onRaw, apiBaseUrl: cfg.apiBaseUrl, initialTasksData: initialCodeQueueOverview });
if (activeModule === "apps" && activeTab === "project-manager") return h(ProjectManagerPage, { microservices: data.microservices, onRaw, apiBaseUrl: cfg.apiBaseUrl });
if (activeModule === "config" && activeTab === "topology") return h(TopologyPage, { data });
-1
View File
@@ -292,7 +292,6 @@ export function ClaudeQqPage({ microservices, onRaw, apiBaseUrl = "/api" }: AnyR
),
),
h(UniDeskErrorBanner, { error: state.error, wide: true }),
actionMessage ? h("div", { className: "form-success wide" }, actionMessage) : null,
),
h("div", { className: "metric-grid" },
h(MetricCard, { label: "Health", value: health.ok || health.status === "ok" ? "OK" : "--", hint: "D601 /health", tone: health.ok || health.status === "ok" ? "ok" : "warn" }),
+177 -51
View File
@@ -6,6 +6,7 @@ import { TraceView, codexTracePort } from "./trace";
import { errorMessage, requestJson as requestUniDeskJson } from "./unidesk-error";
import { UniDeskErrorBanner } from "./unidesk-error-banner";
import { useNotification } from "./notification-context";
import { UniDeskDialog } from "./dialog";
type AnyRecord = Record<string, any>;
@@ -150,6 +151,10 @@ function codexApi(apiBaseUrl: string, path: string): string {
return `${apiBaseUrl}/code-queue-direct${path}`;
}
function oaCodeQueueEventsApi(apiBaseUrl: string): string {
return `${apiBaseUrl}/microservices/oa-event-flow/proxy/api/events/stream?tags=${encodeURIComponent("service:code-queue")}`;
}
function codexNoCacheOptions(): AnyRecord {
return { headers: { "cache-control": "no-cache", "x-unidesk-no-cache": "1" } };
}
@@ -572,17 +577,43 @@ function nonNegativeCount(value: any): number {
return Number.isFinite(numeric) && numeric >= 0 ? Math.floor(numeric) : 0;
}
function canonicalTaskStepCount(task: any): number {
const summary = taskTraceSummary(task);
if (summary !== null) return traceSummaryStepCount(summary);
const value = Number(task?.stepCount ?? task?.llmStepCount ?? 0);
return Number.isFinite(value) && value >= 0 ? Math.floor(value) : 0;
function statCount(value: any): number | null {
const numeric = Number(value);
return Number.isFinite(numeric) && numeric >= 0 ? Math.floor(numeric) : null;
}
function traceSummaryStepCount(summary: AnyRecord | null): number {
const execution = objectRecord(summary?.execution) || {};
const value = Number(summary?.stepCount ?? summary?.llmStepCount ?? execution.stepCount ?? execution.llmStepCount ?? execution.toolCallCount ?? 0);
return nonNegativeCount(value);
function taskOaTraceStats(task: any): AnyRecord | null {
const direct = objectRecord(task?.traceStats);
if (direct && (task?.statsSource === "oa-event-flow" || direct.source === "oa-event-flow")) return direct;
const summary = taskTraceSummary(task);
const summaryStats = objectRecord(summary?.traceStats);
if (summaryStats && (summary?.statsSource === "oa-event-flow" || summaryStats.source === "oa-event-flow")) return summaryStats;
return null;
}
function attemptOaTraceStats(task: any, attempt: any): AnyRecord | null {
const direct = objectRecord(attempt?.traceStats) || objectRecord(attempt?.execution?.traceStats);
const source = String(attempt?.statsSource || attempt?.execution?.statsSource || "");
if (direct && (source === "oa-event-flow" || direct.source === "oa-event-flow")) return direct;
return null;
}
function traceStatCount(task: any, key: string): number | null {
return statCount(taskOaTraceStats(task)?.[key]);
}
function canonicalTaskStepCount(task: any): number | null {
const direct = traceStatCount(task, "stepCount");
if (direct !== null) return direct;
const summary = taskTraceSummary(task);
if (summary !== null) return traceSummaryStepCount(summary);
return null;
}
function traceSummaryStepCount(summary: AnyRecord | null): number | null {
const stats = objectRecord(summary?.traceStats);
if (!stats || (summary?.statsSource !== "oa-event-flow" && stats.source !== "oa-event-flow")) return null;
return statCount(stats.stepCount);
}
function traceSummaryIsCurrent(task: any): boolean {
@@ -853,8 +884,7 @@ function coalesceFileChangeTraceSteps(steps: any[]): any[] {
}
function canonicalExecutionSummary(execution: AnyRecord): AnyRecord {
const stepCount = nonNegativeCount(execution?.stepCount ?? execution?.llmStepCount ?? execution?.toolCallCount);
return { ...execution, stepCount, llmStepCount: stepCount };
return { ...execution };
}
function taskTraceStepsLoaded(task: any, attemptIndex: any = null): boolean {
@@ -1209,7 +1239,7 @@ function providerDefaultWorkdir(queue: any, providerId: string): string {
return id === mainProvider ? String(queue?.defaultWorkdir || "/root/unidesk") : String(queue?.remoteDefaultWorkdir || "/home/ubuntu");
}
function taskStepCount(task: any): number {
function taskStepCount(task: any): number | null {
return canonicalTaskStepCount(task);
}
@@ -1220,6 +1250,8 @@ function TaskCard({ task, selected, onSelect, onCopy, onReference, onMarkRead, c
const updatedAt = latestTimestampValue(task?.updatedAt, taskTraceSummary(task)?.updatedAt);
const recentUpdateLabel = `最近更新: ${fmtRelativeAge(updatedAt)}`;
const stepCount = taskStepCount(task);
const stepLabel = stepCount === null ? "--" : String(stepCount);
const stepTitle = stepCount === null ? "STEP 统计中心同步中" : "STEP 来自 OA Event Flow 统计中心";
return h("article", {
role: "button",
tabIndex: 0,
@@ -1282,7 +1314,7 @@ function TaskCard({ task, selected, onSelect, onCopy, onReference, onMarkRead, c
h("span", null, taskDurationLabel(task)),
),
h("div", { className: "codex-task-meta codex-task-update-meta" },
h("span", { className: "codex-task-recent-update codex-task-step-count", title: "STEP 按 read/edit/run 工具动作统计", "data-testid": `codex-task-step-count-${taskId || "unknown"}` }, `STEP ${stepCount}`),
h("span", { className: "codex-task-recent-update codex-task-step-count", title: stepTitle, "data-testid": `codex-task-step-count-${taskId || "unknown"}` }, `STEP ${stepLabel}`),
h("span", { className: "codex-task-recent-update", title: updatedAt ? `更新时间: ${fmtDate(updatedAt)}` : recentUpdateLabel, "data-testid": `codex-task-recent-update-${taskId || "unknown"}` }, recentUpdateLabel),
h("span", null, fmtDate(updatedAt || task?.updatedAt)),
),
@@ -1537,12 +1569,17 @@ function ProgressivePromptBlock({ task, loading, onLoadPromptPart, testId = "cod
function ProgressiveExecutionSummary({ task, attempt, attemptIndex, loading, onLoadSteps, onLoadStep, testId = "codex-execution-summary" }: AnyRecord) {
const steps = coalesceFileChangeTraceSteps(taskTraceSteps(task, attemptIndex));
const execution = canonicalExecutionSummary(attemptExecutionSummary(task, attempt));
const summary = taskTraceSummary(task);
const stats = attempt ? attemptOaTraceStats(task, attempt) : taskOaTraceStats(task);
const stepDetails = taskTraceStepDetails(task);
const stepsLoaded = taskTraceStepsLoaded(task, attemptIndex);
const errorCount = Number(attempt?.errorCount ?? summary?.errorCount ?? traceStepErrorCount(steps));
const toolCount = nonNegativeCount(execution.toolCallCount);
const stepCount = nonNegativeCount(execution.stepCount ?? execution.llmStepCount ?? execution.toolCallCount);
const errorCount = statCount(stats?.errorCount);
const toolCount = statCount(stats?.stepCount ?? stats?.llmStepCount);
const readCount = statCount(stats?.readCount);
const editCount = statCount(stats?.editCount);
const runCount = statCount(stats?.runCount);
const stepCount = statCount(stats?.stepCount ?? stats?.llmStepCount);
const toolsLabel = toolCount === null ? "--" : String(toolCount);
const stepLabel = stepCount === null ? "--" : String(stepCount);
const editedFiles = Array.isArray(execution.editedFiles) ? execution.editedFiles : [];
const commands = Array.isArray(execution.commands) ? execution.commands : [];
const synthetic = isSyntheticAttemptSegment(attempt, attemptIndex);
@@ -1565,14 +1602,14 @@ function ProgressiveExecutionSummary({ task, attempt, attemptIndex, loading, onL
h("strong", null, `执行过程摘要${labelSuffix}`),
running ? h("span", { className: "codex-summary-running-pill", "data-testid": `${testId}-running` }, "执行中") : null,
h("code", { title: updatedAt ? `最近更新: ${fmtDate(updatedAt)}` : recentUpdateLabel },
`${fmtDuration(execution.durationMs ?? execution.totalElapsedMs)} / ${toolCount} tools / ${recentUpdateLabel}`),
`${fmtDuration(execution.durationMs ?? execution.totalElapsedMs)} / ${toolsLabel} tools / ${recentUpdateLabel}`),
),
h("div", { className: "codex-execution-digest" },
h("span", null, `read ${Number(execution.readCount || 0)}`),
h("span", null, `edit ${Number(execution.editCount || 0)}`),
h("span", null, `run ${Number(execution.runCount || 0)}`),
h("span", null, `STEP ${stepCount}`),
errorCount > 0 ? h("span", { className: "codex-execution-error-pill", "data-testid": `${testId}-error-count` }, `Error ${errorCount}`) : null,
h("span", { title: "来自 OA Event Flow 统计中心" }, `read ${readCount === null ? "--" : readCount}`),
h("span", { title: "来自 OA Event Flow 统计中心" }, `edit ${editCount === null ? "--" : editCount}`),
h("span", { title: "来自 OA Event Flow 统计中心" }, `run ${runCount === null ? "--" : runCount}`),
h("span", { title: "来自 OA Event Flow 统计中心" }, `STEP ${stepLabel}`),
errorCount !== null && errorCount > 0 ? h("span", { className: "codex-execution-error-pill", "data-testid": `${testId}-error-count` }, `Error ${errorCount}`) : null,
),
),
h("div", { className: "codex-execution-digest expanded" },
@@ -1849,6 +1886,8 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
const [referenceTaskId, setReferenceTaskId] = useState("");
const [queueId, setQueueId] = useState("default");
const [selectedQueueId, setSelectedQueueId] = useState(allQueuesId);
const [mergeDialogOpen, setMergeDialogOpen] = useState(false);
const [mergeSourceQueueId, setMergeSourceQueueId] = useState("");
const [providerId, setProviderId] = useState("main-server");
const [model, setModel] = useState("gpt-5.5");
const [cwd, setCwd] = useState("/root/unidesk");
@@ -1888,6 +1927,8 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
const statistics = taskStatistics(tasksData, queue);
const pagination = taskPagination(tasksData);
const queueRows = knownQueueRows(queue, queueId);
const mergeTargetQueueId = String(queueId || "default").trim() || "default";
const mergeSourceQueueRows = queueRows.filter((row: any) => String(row?.id || "") !== mergeTargetQueueId);
const viewQueueRow = selectedQueueRow(queueRows, selectedQueueId);
const totalTaskCount = Number((isAllQueues(selectedQueueId) ? queue?.total : viewQueueRow?.total) ?? pagination.total ?? tasks.length);
const globalActiveIds = activeTaskIds(queue);
@@ -2138,6 +2179,8 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
attempts: Array.isArray(summary.attempts) ? summary.attempts : [],
stepCount: summary.stepCount,
llmStepCount: summary.llmStepCount,
traceStats: summary.traceStats,
statsSource: summary.statsSource,
timing: summary.timing,
_traceSummary: summary,
_traceSummaryLoaded: true,
@@ -2758,25 +2801,36 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
}, "修改 Codex queue 名称失败");
}
async function mergeQueue(): Promise<void> {
const targetQueueId = String(queueId || "default").trim() || "default";
const candidates = queueRows.filter((row: any) => String(row?.id || "") !== targetQueueId);
if (candidates.length === 0) {
function openMergeQueueDialog(): void {
if (mergeSourceQueueRows.length === 0) {
const msg = "没有可合并的其他 queue;请先创建或选择另一个 queue。";
setNotice(msg);
return;
}
const proposed = typeof window === "undefined"
? ""
: window.prompt(`将源 queue 合并到当前 queue=${targetQueueId};源 queue 记录会保留且不会删除。请输入源 queue ID:`, String(candidates[0]?.id || ""));
const existingSource = mergeSourceQueueRows.some((row: any) => String(row?.id || "") === mergeSourceQueueId)
? mergeSourceQueueId
: "";
setMergeSourceQueueId(existingSource);
setMergeDialogOpen(true);
}
async function mergeQueue(): Promise<void> {
const targetQueueId = mergeTargetQueueId;
if (mergeSourceQueueRows.length === 0) {
const msg = "没有可合并的其他 queue;请先创建或选择另一个 queue。";
setNotice(msg);
return;
}
const proposed = mergeSourceQueueId || "";
const sourceQueueId = String(proposed || "").trim();
if (!sourceQueueId) return;
if (!sourceQueueId) {
setNotice("请先选择要合并的源 queue。");
return;
}
if (sourceQueueId === targetQueueId) {
setError("源 queue 和目标 queue 不能相同。");
return;
}
const confirmed = typeof window === "undefined" ? true : window.confirm(`确认把 queue=${sourceQueueId} 的全部任务合并到 queue=${targetQueueId}?源 queue 不会删除;任务会按原 queueEnteredAt/createdAt 时间顺序进入目标 queue。`);
if (!confirmed) return;
await guarded(async () => {
const result = await requestJson(codexApi(apiBaseUrl, `/api/queues/${encodeURIComponent(targetQueueId)}/merge`), { method: "POST", body: { sourceQueueId } });
if (result?.summary) {
@@ -2786,9 +2840,11 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
setSelectedQueueId(targetQueueId);
setTasksData(null);
const count = Number(result?.mergedTaskCount || 0);
const msg = `已将 queue=${sourceQueueId} 合并到 ${targetQueueId},移动 ${count} 个任务;源 queue 已保留`;
const msg = `已将 queue=${sourceQueueId} 合并到 ${targetQueueId},移动 ${count} 个任务;源 queue 已自动删除`;
setNotice(msg);
addNotification("success", msg);
setMergeDialogOpen(false);
setMergeSourceQueueId("");
await load(selectedIdRef.current, true, targetQueueId);
}, "合并 Codex queue 失败");
}
@@ -2995,11 +3051,12 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
function taskEventNeedsOverviewRefresh(event: AnyRecord, taskId: string, previousTask: any): boolean {
const type = String(event?.type || "");
if (type === "queue-updated") return true;
if (type === "trace-stats-updated" || type === "trace-step-created") return false;
if (taskId.length === 0) return true;
if (!previousTask) return type !== "task-step";
if (!previousTask) return true;
if (event?.queueId && String(event.queueId) !== taskQueueLabel(previousTask)) return true;
if (event?.status && String(event.status) !== String(previousTask?.status || "")) return true;
return type !== "task-step" && String(event?.reason || "") !== "output";
return String(event?.reason || "") !== "output";
}
function scheduleEventDrivenOverviewRefresh(): void {
@@ -3026,30 +3083,72 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
}, 250);
}
function normalizeOaCodeQueueEvent(event: AnyRecord): AnyRecord {
const payload = objectRecord(event?.payload) || event;
const stats = objectRecord(payload?.stats);
const type = String(event?.type || payload?.type || "");
const subjectKind = String(payload?.subjectKind || stats?.subjectKind || "");
const scopeId = String(payload?.scopeId || stats?.scopeId || "");
const taskId = String(payload?.taskId || (subjectKind === "task" ? payload?.subjectId : "") || stats?.taskId || event?.taskId || "");
const attemptValue = payload?.attemptIndex ?? stats?.attemptIndex;
const attemptIndex = attemptValue === null || attemptValue === undefined || attemptValue === "" ? null : statCount(attemptValue);
const traceStats = stats || objectRecord(payload?.traceStats);
const stepCount = traceStats?.stepCount ?? traceStats?.llmStepCount;
const outputMaxSeq = traceStats?.outputMaxSeq;
return {
...payload,
type,
eventId: event?.eventId || payload?.eventId,
sequence: event?.sequence ?? payload?.sequence,
taskId,
subjectKind,
scopeId,
attemptIndex,
stepCount,
outputMaxSeq,
updatedAt: traceStats?.updatedAt || payload?.updatedAt || event?.createdAt,
traceStats,
statsSource: traceStats ? "oa-event-flow" : payload?.statsSource,
};
}
function applyCodeQueueEvent(event: AnyRecord): void {
const taskId = String(event?.taskId || "");
const stepCount = nonNegativeCount(event?.stepCount);
const normalizedEvent = normalizeOaCodeQueueEvent(event);
const taskId = String(normalizedEvent?.taskId || "");
const scopeId = String(normalizedEvent?.scopeId || normalizedEvent?.traceStats?.scopeId || "");
const isAttemptStats = String(normalizedEvent?.subjectKind || normalizedEvent?.traceStats?.subjectKind || "") === "task-attempt"
|| Number(normalizedEvent?.attemptIndex) > 0
|| /:attempt:\d+$/u.test(scopeId);
const stepCount = nonNegativeCount(normalizedEvent?.stepCount);
const patch: AnyRecord = {};
const previousTask = taskId.length > 0
? taskRows(tasksDataRef.current).find((task: any) => String(task?.id || "") === taskId)
: null;
if (taskId.length > 0) {
if (event?.status) patch.status = String(event.status);
if (event?.updatedAt) patch.updatedAt = String(event.updatedAt);
if (event?.queueId) patch.queueId = String(event.queueId);
if (Number.isFinite(Number(event?.stepCount))) {
if (normalizedEvent?.status) patch.status = String(normalizedEvent.status);
if (normalizedEvent?.updatedAt) patch.updatedAt = String(normalizedEvent.updatedAt);
if (normalizedEvent?.queueId) patch.queueId = String(normalizedEvent.queueId);
if (!isAttemptStats && Number.isFinite(Number(normalizedEvent?.stepCount))) {
patch.stepCount = stepCount;
patch.llmStepCount = stepCount;
}
if (!isAttemptStats && Number.isFinite(Number(normalizedEvent?.outputMaxSeq))) patch.outputMaxSeq = nonNegativeCount(normalizedEvent.outputMaxSeq);
if (!isAttemptStats && objectRecord(normalizedEvent?.traceStats)) {
patch.traceStats = normalizedEvent.traceStats;
patch.statsSource = "oa-event-flow";
}
if (Object.keys(patch).length > 0 && (sessionCacheRef.current.has(taskId) || selectedIdRef.current === taskId)) {
publishCachedTask(taskId, patch, detailLoadTokenRef.current);
}
if (Object.keys(patch).length > 0) patchSidebarTaskFromEvent(taskId, patch);
if (selectedIdRef.current === taskId && (event?.type === "task-step" || event?.type === "task-updated")) {
scheduleSelectedTraceEventRefresh(taskId, event?.type === "task-step");
if (selectedIdRef.current === taskId && (normalizedEvent?.type === "trace-step-created" || normalizedEvent?.type === "task-updated" || normalizedEvent?.type === "trace-stats-updated")) {
const previousStepCount = canonicalTaskStepCount(previousTask);
const shouldRefreshSteps = normalizedEvent?.type === "trace-step-created"
|| (normalizedEvent?.type === "trace-stats-updated" && !isAttemptStats && Number.isFinite(Number(normalizedEvent?.stepCount)) && (previousStepCount === null || stepCount > previousStepCount));
scheduleSelectedTraceEventRefresh(taskId, shouldRefreshSteps);
}
}
if (taskEventNeedsOverviewRefresh(event, taskId, previousTask)) scheduleEventDrivenOverviewRefresh();
if (taskEventNeedsOverviewRefresh(normalizedEvent, taskId, previousTask)) scheduleEventDrivenOverviewRefresh();
}
useEffect(() => {
@@ -3062,7 +3161,7 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
useEffect(() => {
if (!service || typeof EventSource === "undefined") return undefined;
const source = new EventSource(codexApi(apiBaseUrl, "/api/events"), { withCredentials: true });
const source = new EventSource(oaCodeQueueEventsApi(apiBaseUrl), { withCredentials: true });
const onEvent = (message: MessageEvent): void => {
try {
applyCodeQueueEvent(JSON.parse(String(message.data || "{}")));
@@ -3073,9 +3172,10 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
const onVisible = (): void => {
if (isDocumentVisible()) scheduleEventDrivenOverviewRefresh();
};
source.addEventListener("task-step", onEvent);
source.addEventListener("trace-step-created", onEvent);
source.addEventListener("task-updated", onEvent);
source.addEventListener("queue-updated", onEvent);
source.addEventListener("trace-stats-updated", onEvent);
document.addEventListener("visibilitychange", onVisible);
return () => {
source.close();
@@ -3101,7 +3201,7 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
if (autoTraceLoadKeysRef.current.has(key)) return;
autoTraceLoadKeysRef.current.add(key);
void ensureTraceSummary(taskId, true).catch((err) => setError(errorText(err, "自动加载 Trace Summary 失败")));
}, [service?.id, selectedTask?.id, selectedTask?.updatedAt, selectedTask?.stepCount, selectedTask?.llmStepCount, selectedTask?._traceSummaryUpdatedAt, selectedTask?._traceSummaryLoaded, selectedDetailLoading]);
}, [service?.id, selectedTask?.id, selectedTask?.updatedAt, selectedTask?.traceStats?.statsRevision, selectedTask?._traceSummaryUpdatedAt, selectedTask?._traceSummaryLoaded, selectedDetailLoading]);
const taskListContent = sidebarTasks.length === 0 ? h(EmptyState, {
title: searchActive ? (searchLoading ? "搜索中" : "没有匹配任务") : "队列为空",
@@ -3261,6 +3361,32 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
const loadDetailMs = Number(loadStats?.detailMs);
const loadTranscriptRows = Number(loadStats?.transcriptRows);
const loadState = loadStats?.phase === "complete" ? "complete" : String(loadStats?.phase || "idle");
const mergeTargetQueueRow = selectedQueueRow(queueRows, mergeTargetQueueId) || { id: mergeTargetQueueId, name: mergeTargetQueueId };
const mergeDialog = mergeDialogOpen ? h(UniDeskDialog, {
title: "合并 queue",
titleId: "codex-merge-dialog-title",
className: "codex-merge-dialog",
backdropClassName: "codex-merge-dialog-backdrop",
bodyClassName: "codex-merge-dialog-body",
actionsClassName: "codex-merge-dialog-actions",
testId: "codex-merge-queue-dialog",
closeTestId: "codex-merge-queue-close",
disableClose: busy,
onClose: () => setMergeDialogOpen(false),
actions: [
h("button", { key: "cancel", type: "button", className: "ghost-btn", onClick: () => setMergeDialogOpen(false), disabled: busy, "data-testid": "codex-merge-queue-cancel" }, "取消"),
h("button", { key: "confirm", type: "button", className: "primary-btn", onClick: () => void mergeQueue(), disabled: busy || submitting || !mergeSourceQueueId, "data-testid": "codex-merge-queue-confirm" }, busy ? "合并中..." : "确认合并"),
],
},
h("p", { className: "codex-merge-dialog-target" }, "目标 queue", h("code", null, queueOptionLabel(mergeTargetQueueRow))),
h("label", null, "源 queue",
h("select", { value: mergeSourceQueueId, disabled: busy || submitting, onChange: (event: any) => setMergeSourceQueueId(String(event.target.value || "")), "data-testid": "codex-merge-source-queue-select" },
h("option", { value: "" }, "选择要合并进来的源 queue"),
mergeSourceQueueRows.map((row: any) => h("option", { key: String(row?.id || ""), value: String(row?.id || "") }, queueOptionLabel(row))),
),
),
h("p", { className: "codex-merge-dialog-note" }, "会把源 queue 的任务归属合并到目标 queue,并自动删除源 queue;目标 queue 会按原 queueEnteredAt/createdAt 时间顺序运行。"),
) : null;
return h("div", {
className: `code-queue-page ${standalone ? "codex-standalone-page" : ""}`,
@@ -3274,7 +3400,7 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
"data-load-partial": loadStats?.partial ? "true" : "false",
},
h(UniDeskErrorBanner, { error, wide: true }),
notice ? h("div", { className: "form-success wide", "data-testid": "codex-create-success" }, notice) : null,
mergeDialog,
h("div", { className: "codex-session-stage codex-session-stage-top" },
sessionPanel,
),
@@ -3292,11 +3418,11 @@ export function CodeQueuePage({ microservices, onRaw, apiBaseUrl = "/api", initi
h("div", { className: "codex-form-grid" },
h("label", { className: "codex-submit-queue-field" }, "Queue",
h("div", { className: "codex-submit-queue-row" },
h("select", { value: queueId, disabled: submitting, onChange: (event: any) => setQueueId(String(event.target.value || "default")), "data-testid": "code-queue-id-select" },
h("select", { className: "codex-submit-queue-select", value: queueId, disabled: submitting, onChange: (event: any) => setQueueId(String(event.target.value || "default")), "data-testid": "code-queue-id-select" },
queueRows.map((row: any) => h("option", { key: String(row?.id || ""), value: String(row?.id || "") }, queueOptionLabel(row))),
),
h("button", { type: "button", className: "ghost-btn codex-rename-queue-btn", onClick: () => void renameQueue(), disabled: busy || submitting || !queueId, title: "修改当前 queue 的显示名称,ID 不变", "data-testid": "codex-rename-queue-button" }, "改名"),
h("button", { type: "button", className: "ghost-btn codex-merge-queue-btn", onClick: () => void mergeQueue(), disabled: busy || submitting || queueRows.length < 2, title: "把其他 queue 合并到当前 queue;源 queue 记录会保留,不会删除", "data-testid": "codex-merge-queue-button" }, "合并 queue"),
h("button", { type: "button", className: "ghost-btn codex-merge-queue-btn", onClick: () => openMergeQueueDialog(), disabled: busy || submitting || mergeSourceQueueRows.length === 0, title: "打开独立窗口选择源 queue 合并到当前 queue合并后自动删除源 queue", "data-testid": "codex-merge-queue-button" }, "合并 queue"),
h("button", { type: "button", className: "ghost-btn codex-create-queue-btn", onClick: () => void createQueue(), disabled: busy || submitting, "data-testid": "codex-create-queue-button" }, "创建 queue"),
),
),
+70
View File
@@ -0,0 +1,70 @@
import React, { type ReactNode } from "react";
const h = React.createElement;
function joinClassNames(...values: Array<string | undefined | null | false>): string {
return values.filter(Boolean).join(" ");
}
export interface UniDeskDialogProps {
title: ReactNode;
children?: ReactNode;
actions?: ReactNode;
onClose: () => void;
className?: string;
backdropClassName?: string;
bodyClassName?: string;
actionsClassName?: string;
titleId?: string;
testId?: string;
closeTestId?: string;
closeLabel?: string;
disableClose?: boolean;
closeOnBackdrop?: boolean;
}
export function UniDeskDialog({
title,
children,
actions,
onClose,
className,
backdropClassName,
bodyClassName,
actionsClassName,
titleId,
testId,
closeTestId,
closeLabel = "关闭",
disableClose = false,
closeOnBackdrop = true,
}: UniDeskDialogProps) {
const sectionProps: Record<string, unknown> = {
className: joinClassNames("raw-dialog", "unidesk-dialog", className),
role: "dialog",
"aria-modal": "true",
};
if (titleId) {
sectionProps["aria-labelledby"] = titleId;
} else if (typeof title === "string") {
sectionProps["aria-label"] = title;
}
if (testId) sectionProps["data-testid"] = testId;
return h("div", {
className: joinClassNames("modal-backdrop", "unidesk-dialog-backdrop", backdropClassName),
role: "presentation",
onClick: (event: React.MouseEvent<HTMLElement>) => {
if (closeOnBackdrop && !disableClose && event.target === event.currentTarget) onClose();
},
},
h("section", sectionProps,
h("div", { className: "raw-dialog-head unidesk-dialog-head" },
h("strong", titleId ? { id: titleId } : null, title),
h("button", { type: "button", className: "ghost-btn", onClick: onClose, disabled: disableClose, "data-testid": closeTestId }, closeLabel),
),
h("div", { className: joinClassNames("unidesk-dialog-body", bodyClassName) }, children),
actions ? h("div", { className: joinClassNames("unidesk-dialog-actions", actionsClassName) }, actions) : null,
),
);
}
+10 -11
View File
@@ -707,10 +707,9 @@ async function proxyCodeQueueDirect(req: Request, url: URL): Promise<Response> {
}
const overviewCacheKey = `${suffix}${url.search}`;
const canUseOverviewCache = req.method === "GET" && suffix === "/api/tasks/overview";
const isEventStreamRequest = req.method === "GET" && suffix === "/api/events";
const cacheControl = req.headers.get("cache-control") || "";
const bypassOverviewCache = /\bno-cache\b|\bno-store\b/iu.test(cacheControl) || req.headers.get("x-unidesk-no-cache") === "1";
const expectsJsonResponse = !isEventStreamRequest && (suffix === "/health" || suffix.startsWith("/api/"));
const expectsJsonResponse = suffix === "/health" || suffix.startsWith("/api/");
if (req.method !== "GET" && req.method !== "HEAD") invalidateCodeQueueOverviewCache();
if (canUseOverviewCache && !bypassOverviewCache) {
const cached = cachedCodeQueueOverview(overviewCacheKey);
@@ -725,7 +724,7 @@ async function proxyCodeQueueDirect(req: Request, url: URL): Promise<Response> {
headers.delete("connection");
headers.delete("content-length");
headers.delete("cookie");
const init: RequestInit = { method: req.method, headers, redirect: "manual" };
const init: RequestInit = { method: req.method, headers, redirect: "manual", signal: req.signal };
if (req.method !== "GET" && req.method !== "HEAD") {
init.body = await req.arrayBuffer();
}
@@ -736,13 +735,6 @@ async function proxyCodeQueueDirect(req: Request, url: URL): Promise<Response> {
const responseHeaders = new Headers();
const upstreamContentType = upstream.headers.get("content-type");
if (upstreamContentType !== null) responseHeaders.set("content-type", upstreamContentType);
if (isEventStreamRequest) {
responseHeaders.set("cache-control", upstream.headers.get("cache-control") || "no-store, no-transform");
responseHeaders.set("connection", "keep-alive");
const buffering = upstream.headers.get("x-accel-buffering");
if (buffering !== null) responseHeaders.set("x-accel-buffering", buffering);
return new Response(upstream.body, { status: upstream.status, headers: responseHeaders });
}
const upstreamBody = await upstream.arrayBuffer();
const isJsonResponse = (upstreamContentType ?? "").toLowerCase().includes("json");
let parsedJson: unknown = null;
@@ -821,7 +813,7 @@ async function proxyApi(req: Request, url: URL): Promise<Response> {
headers.delete("connection");
headers.delete("content-length");
headers.delete("cookie");
const init: RequestInit = { method: req.method, headers, redirect: "manual" };
const init: RequestInit = { method: req.method, headers, redirect: "manual", signal: req.signal };
if (req.method !== "GET" && req.method !== "HEAD") {
init.body = await req.arrayBuffer();
}
@@ -839,6 +831,13 @@ async function proxyApi(req: Request, url: URL): Promise<Response> {
const responseHeaders = new Headers();
const upstreamContentType = upstream.headers.get("content-type");
if (upstreamContentType !== null) responseHeaders.set("content-type", upstreamContentType);
if ((upstreamContentType ?? "").toLowerCase().includes("text/event-stream")) {
responseHeaders.set("cache-control", upstream.headers.get("cache-control") || "no-store, no-transform");
responseHeaders.set("connection", "keep-alive");
const buffering = upstream.headers.get("x-accel-buffering");
if (buffering !== null) responseHeaders.set("x-accel-buffering", buffering);
return new Response(upstream.body, { status: upstream.status, headers: responseHeaders });
}
return new Response(await upstream.arrayBuffer(), { status: upstream.status, headers: responseHeaders });
}
@@ -69,6 +69,7 @@ export const MODULES: UniDeskModuleDefinition[] = [
{ id: "claudeqq", label: "ClaudeQQ" },
{ id: "baidu-netdisk", label: "Baidu Netdisk" },
{ id: "filebrowser", label: "File Browser" },
{ id: "oa-event-flow", label: "OA Event Flow" },
{ id: "code-queue", label: "Code Queue" },
{ id: "project-manager", label: "Project Manager" },
] },
@@ -0,0 +1,332 @@
import React from "react";
import { fmtClock, fmtDate } from "./time";
import { LoadingTitle } from "./loading-indicator";
import { errorMessage, requestJson } from "./unidesk-error";
import { UniDeskErrorBanner } from "./unidesk-error-banner";
type AnyRecord = Record<string, any>;
const h = React.createElement;
const { useEffect, useMemo } = React;
const useState: any = React.useState;
function StatusBadge({ status, children, title }: AnyRecord) {
const normalized = String(status || "unknown").toLowerCase();
return h("span", { className: `status-badge ${normalized}`, title }, children || status || "unknown");
}
function MetricCard({ label, value, hint, tone }: AnyRecord) {
return h("article", { className: `metric-card ${tone || ""}` },
h("div", { className: "metric-label" }, label),
h("div", { className: "metric-value" }, value),
h("div", { className: "metric-hint" }, hint),
);
}
function Panel({ title, eyebrow, actions, children, className, loading }: AnyRecord) {
return h("section", { className: `panel ${className || ""}` },
h("div", { className: "panel-head" },
h("div", null,
eyebrow ? h("p", { className: "panel-eyebrow" }, eyebrow) : null,
h(LoadingTitle, { title, loading }),
),
actions ? h("div", { className: "panel-actions" }, actions) : null,
),
h("div", { className: "panel-body" }, children),
);
}
function RawButton({ title, data, onOpen, testId }: AnyRecord) {
return h("button", {
type: "button",
className: "ghost-btn",
"data-testid": testId,
onClick: () => onOpen?.(title, data),
}, "查看原始JSON");
}
function EmptyState({ title, text }: AnyRecord) {
return h("div", { className: "empty-state" }, h("strong", null, title), h("span", null, text));
}
function objectRecord(value: any): AnyRecord | null {
return value && typeof value === "object" && !Array.isArray(value) ? value : null;
}
function asArray(value: any): any[] {
return Array.isArray(value) ? value : [];
}
function fmtCount(value: any): string {
const number = Number(value);
return Number.isFinite(number) ? number.toLocaleString("zh-CN") : "--";
}
function compactText(value: any, max = 140): string {
if (value === null || value === undefined) return "--";
const text = typeof value === "string" ? value : JSON.stringify(value);
const normalized = String(text || "").replace(/\s+/gu, " ").trim();
return normalized.length > max ? `${normalized.slice(0, max - 1)}...` : normalized || "--";
}
function eventTags(event: any): string[] {
return asArray(event?.tags).map((tag) => String(tag || "").trim()).filter(Boolean);
}
function statNumber(value: any): number {
const number = Number(value);
return Number.isFinite(number) && number >= 0 ? Math.floor(number) : 0;
}
function microserviceRuntime(service: any): AnyRecord {
return service?.runtime && typeof service.runtime === "object" && !Array.isArray(service.runtime) ? service.runtime : {};
}
function microserviceBackend(service: any): AnyRecord {
return service?.backend && typeof service.backend === "object" && !Array.isArray(service.backend) ? service.backend : {};
}
function normalizeTagFilter(value: string): string {
return String(value || "")
.split(/[\s,]+/u)
.map((tag) => tag.trim())
.filter(Boolean)
.join(",");
}
function oaProxy(apiBaseUrl: string, path: string): string {
return `${apiBaseUrl}/microservices/oa-event-flow/proxy${path}`;
}
function eventTypeTone(type: string): string {
if (type.includes("error") || type.includes("failed")) return "failed";
if (type.includes("stats")) return "ok";
if (type.includes("step") || type.includes("updated")) return "running";
return "queued";
}
function statScopeLabel(stat: any): string {
const kind = String(stat?.subjectKind || "trace");
const id = String(stat?.subjectId || stat?.scopeId || "");
return id ? `${kind}:${id}` : String(stat?.scopeId || "--");
}
function EventTagRail({ tags }: AnyRecord) {
const visible = eventTags({ tags }).slice(0, 6);
return h("div", { className: "oa-tag-rail" },
visible.length === 0 ? h("span", { className: "muted" }, "--") : visible.map((tag) => h("code", { key: tag }, tag)),
);
}
function EventTable({ events, onRaw }: AnyRecord) {
const rows = [...asArray(events)].reverse();
return rows.length === 0
? h(EmptyState, { title: "事件表暂无记录", text: "等待 Code Queue 或 Pipeline 按 tag 发布 OA 事件" })
: h("div", { className: "table-wrap oa-event-table-wrap" }, h("table", { className: "oa-event-table", "data-testid": "oa-event-flow-event-table" },
h("thead", null, h("tr", null,
h("th", null, "Seq"),
h("th", null, "Type"),
h("th", null, "Source"),
h("th", null, "Aggregate"),
h("th", null, "Tags"),
h("th", null, "Payload"),
h("th", null, "Created"),
h("th", null, "Raw"),
)),
h("tbody", null, rows.map((event: any) => {
const type = String(event?.type || "event");
const aggregate = `${String(event?.aggregateType || "--")}:${String(event?.aggregateId || "--")}`;
return h("tr", { key: event?.eventId || event?.sequence },
h("td", null, h("code", null, fmtCount(event?.sequence))),
h("td", null, h(StatusBadge, { status: eventTypeTone(type) }, type)),
h("td", null, h("strong", null, event?.sourceId || "--"), h("code", null, event?.sourceKind || "--")),
h("td", null, h("code", null, aggregate)),
h("td", null, h(EventTagRail, { tags: event?.tags })),
h("td", null, h("span", { className: "oa-payload-preview" }, compactText(event?.payload, 180))),
h("td", null, fmtDate(event?.createdAt)),
h("td", null, h(RawButton, { title: `OA Event ${event?.sequence || ""}`, data: event, onOpen: onRaw, testId: `raw-oa-event-${event?.sequence || "unknown"}` })),
);
})),
));
}
function StatsTable({ stats, onRaw }: AnyRecord) {
const rows = asArray(stats);
return rows.length === 0
? h(EmptyState, { title: "统计中心暂无投影", text: "trace-stats-snapshot / trace-step-created 进入事件流后会更新这里" })
: h("div", { className: "table-wrap oa-stats-table-wrap" }, h("table", { className: "oa-stats-table", "data-testid": "oa-event-flow-stats" },
h("thead", null, h("tr", null,
h("th", null, "Scope"),
h("th", null, "Service"),
h("th", null, "STEP"),
h("th", null, "Read"),
h("th", null, "Edit"),
h("th", null, "Run"),
h("th", null, "Error"),
h("th", null, "Output Seq"),
h("th", null, "Revision"),
h("th", null, "Updated"),
h("th", null, "Raw"),
)),
h("tbody", null, rows.map((stat: any) => h("tr", { key: stat?.scopeId || `${stat?.serviceId}-${stat?.subjectId}` },
h("td", null, h("strong", null, statScopeLabel(stat)), h("code", null, stat?.scopeId || "--")),
h("td", null, h(StatusBadge, { status: String(stat?.serviceId || "unknown") === "code-queue" ? "running" : "queued" }, stat?.serviceId || "--")),
h("td", null, h("strong", null, fmtCount(statNumber(stat?.stepCount ?? stat?.llmStepCount)))),
h("td", null, fmtCount(statNumber(stat?.readCount))),
h("td", null, fmtCount(statNumber(stat?.editCount))),
h("td", null, fmtCount(statNumber(stat?.runCount))),
h("td", null, fmtCount(statNumber(stat?.errorCount))),
h("td", null, h("code", null, fmtCount(statNumber(stat?.outputMaxSeq)))),
h("td", null, fmtCount(statNumber(stat?.statsRevision))),
h("td", null, fmtDate(stat?.updatedAt)),
h("td", null, h(RawButton, { title: `OA Trace Stats ${stat?.scopeId || ""}`, data: stat, onOpen: onRaw, testId: `raw-oa-stats-${String(stat?.scopeId || "unknown").replace(/[^a-zA-Z0-9_-]/gu, "_")}` })),
))),
));
}
export function OaEventFlowPage({ microservices, onRaw, apiBaseUrl = "/api" }: AnyRecord) {
const service = microservices.find((item: any) => item.id === "oa-event-flow") || null;
const [tagFilter, setTagFilter] = useState("service:code-queue");
const [state, setState] = useState({ loading: false, error: "", health: null, diagnostics: null, events: [], stats: [], refreshedAt: null });
const [streamState, setStreamState] = useState({ status: "idle", message: "未连接", lastEventAt: "" });
const normalizedTags = useMemo(() => normalizeTagFilter(tagFilter), [tagFilter]);
async function load(): Promise<void> {
if (!service) return;
setState((previous: any) => ({ ...previous, loading: true, error: "" }));
try {
const tagQuery = normalizedTags ? `tags=${encodeURIComponent(normalizedTags)}&` : "";
const [health, diagnostics, events, stats] = await Promise.all([
requestJson(`${apiBaseUrl}/microservices/oa-event-flow/health`, { failureFields: [] }),
requestJson(oaProxy(apiBaseUrl, "/api/diagnostics")),
requestJson(oaProxy(apiBaseUrl, `/api/events?${tagQuery}limit=100`)),
requestJson(oaProxy(apiBaseUrl, `/api/stats/trace?${tagQuery}limit=100`)),
]);
setState({
loading: false,
error: "",
health,
diagnostics,
events: asArray(events?.events),
stats: asArray(stats?.stats),
refreshedAt: new Date(),
});
} catch (err) {
setState((previous: any) => ({ ...previous, loading: false, error: errorMessage(err, "OA Event Flow 加载失败") }));
}
}
useEffect(() => {
void load();
}, [service?.id, service?.runtime?.providerStatus, normalizedTags]);
useEffect(() => {
if (!service || typeof EventSource === "undefined") return undefined;
const tagQuery = normalizedTags ? `?tags=${encodeURIComponent(normalizedTags)}` : "";
const source = new EventSource(`${oaProxy(apiBaseUrl, "/api/events/stream")}${tagQuery}`, { withCredentials: true });
setStreamState({ status: "running", message: "SSE connecting", lastEventAt: "" });
const onHello = (message: MessageEvent): void => {
setStreamState({ status: "online", message: compactText(message.data, 120), lastEventAt: new Date().toISOString() });
};
const onMessage = (message: MessageEvent): void => {
try {
const event = JSON.parse(String(message.data || "{}"));
setStreamState({ status: "online", message: String(event?.type || message.type || "event"), lastEventAt: new Date().toISOString() });
setState((previous: any) => {
const events = [...asArray(previous.events).filter((row: any) => String(row?.eventId || "") !== String(event?.eventId || "")), event]
.sort((left: any, right: any) => Number(left?.sequence || 0) - Number(right?.sequence || 0))
.slice(-100);
const stats = event?.type === "trace-stats-updated" && objectRecord(event?.payload?.stats)
? [event.payload.stats, ...asArray(previous.stats).filter((row: any) => String(row?.scopeId || "") !== String(event.payload.stats.scopeId || ""))].slice(0, 100)
: previous.stats;
return { ...previous, events, stats };
});
} catch (err) {
setStreamState({ status: "warn", message: errorMessage(err, "SSE 事件解析失败"), lastEventAt: new Date().toISOString() });
}
};
const onError = (): void => {
setStreamState((previous: any) => ({ ...previous, status: "warn", message: "SSE reconnecting" }));
};
source.addEventListener("hello", onHello);
source.addEventListener("task-updated", onMessage);
source.addEventListener("queue-updated", onMessage);
source.addEventListener("trace-step-created", onMessage);
source.addEventListener("trace-stats-snapshot", onMessage);
source.addEventListener("trace-stats-updated", onMessage);
source.addEventListener("trace-error", onMessage);
source.onerror = onError;
return () => source.close();
}, [service?.id, apiBaseUrl, normalizedTags]);
if (!service) return h(EmptyState, { title: "OA Event Flow 未登记", text: "请在 config.json 的 microservices 中登记 id=oa-event-flow" });
const runtime = microserviceRuntime(service);
const backend = microserviceBackend(service);
const diagnostics = state.diagnostics || {};
const health = state.health || {};
const eventCount = diagnostics.eventCount ?? health.eventCount;
const statsCount = diagnostics.traceStatsCount ?? health.traceStatsCount;
const latestSequence = diagnostics.latestSequence ?? health.latestSequence;
const pipelineBridge = diagnostics.pipelineBridge || health.pipelineBridge || {};
const eventTypes = asArray(diagnostics.eventTypes).slice(0, 8);
return h("div", { className: "oa-event-flow-page", "data-testid": "oa-event-flow-page" },
h(Panel, {
title: "OA Event Flow 控制台",
eyebrow: "Unified OA Event Bus + Stats Projection",
loading: state.loading,
actions: h("div", { className: "panel-actions" },
h("button", { type: "button", className: "ghost-btn", onClick: load, disabled: state.loading, "data-testid": "oa-event-flow-refresh" }, state.loading ? "刷新中" : "刷新"),
h(RawButton, { title: "OA Event Flow Service", data: service, onOpen: onRaw, testId: "raw-oa-event-flow-service" }),
),
},
h("div", { className: "oa-flow-hero" },
h("div", null,
h("div", { className: "node-version-line" },
h(StatusBadge, { status: health?.ok || runtime.providerStatus === "online" ? "online" : "warn" }, health?.ok ? "HEALTH OK" : runtime.providerStatus || "unknown"),
h(StatusBadge, { status: streamState.status }, streamState.status.toUpperCase()),
h("span", null, backend.public ? "公网暴露" : "仅 UniDesk frontend 代理访问"),
),
h("p", { className: "muted paragraph" }, "独立事件流微服务统一承载 Code Queue 与 Pipeline 的事件发布、tag 订阅、事件表审计和 Trace/STEP 统计投影。"),
),
h("div", { className: "oa-flow-signal" },
h("span", null, "stream"),
h("strong", null, streamState.message || "--"),
h("code", null, streamState.lastEventAt ? fmtClock(new Date(streamState.lastEventAt)) : "waiting"),
),
),
h(UniDeskErrorBanner, { error: state.error, wide: true }),
),
h("div", { className: "oa-flow-metrics" },
h(MetricCard, { label: "事件总量", value: fmtCount(eventCount), hint: `latest seq ${fmtCount(latestSequence)}`, tone: "ok" }),
h(MetricCard, { label: "Trace Stats", value: fmtCount(statsCount), hint: "oa_trace_stats 投影" }),
h(MetricCard, { label: "SSE Clients", value: fmtCount(health?.sseClientCount ?? asArray(diagnostics.sseClients).length), hint: streamState.message || "tag subscription" }),
h(MetricCard, { label: "Pipeline Bridge", value: pipelineBridge?.enabled ? fmtCount(pipelineBridge?.insertedCount) : "OFF", hint: pipelineBridge?.lastError || pipelineBridge?.lastFinishedAt || `${pipelineBridge?.mode || "snapshot"} service:pipeline` }),
h(MetricCard, { label: "DB", value: health?.databaseReady || diagnostics.databaseReady ? "READY" : "WAIT", hint: health?.databaseLastError || diagnostics.databaseLastError || "PostgreSQL persisted" }),
),
h(Panel, { title: "标签订阅", eyebrow: state.refreshedAt ? `Updated ${fmtClock(state.refreshedAt)}` : "Tag Pub/Sub" },
h("div", { className: "oa-filter-bar" },
h("label", null, h("span", null, "tags"), h("input", { value: tagFilter, onChange: (event: any) => setTagFilter(event.target.value), placeholder: "service:code-queue, trace", "data-testid": "oa-event-flow-tag-filter" })),
h("div", { className: "oa-filter-presets" },
h("button", { type: "button", className: "ghost-btn", onClick: () => setTagFilter("service:code-queue") }, "Code Queue"),
h("button", { type: "button", className: "ghost-btn", onClick: () => setTagFilter("service:pipeline") }, "Pipeline"),
h("button", { type: "button", className: "ghost-btn", onClick: () => setTagFilter("trace") }, "Trace"),
h("button", { type: "button", className: "ghost-btn", onClick: () => setTagFilter("") }, "All"),
),
h("code", null, normalizedTags || "all events"),
),
h("div", { className: "oa-type-strip" },
eventTypes.length === 0 ? h("span", { className: "muted" }, "等待事件类型统计") : eventTypes.map((row: any) => h("span", { key: row.type, className: "data-chip" }, `${row.type} ${fmtCount(row.count)}`)),
),
),
h("div", { className: "oa-flow-grid" },
h(Panel, { title: "事件表", eyebrow: "oa_events persisted log", className: "oa-flow-wide", loading: state.loading,
actions: h(RawButton, { title: "OA Event Query", data: { events: state.events, diagnostics }, onOpen: onRaw, testId: "raw-oa-events" }),
}, h(EventTable, { events: state.events, onRaw })),
h(Panel, { title: "统计中心", eyebrow: "oa_trace_stats read model", className: "oa-flow-wide", loading: state.loading,
actions: h(RawButton, { title: "OA Trace Stats", data: state.stats, onOpen: onRaw, testId: "raw-oa-trace-stats" }),
}, h(StatsTable, { stats: state.stats, onRaw })),
),
);
}
+23 -71
View File
@@ -847,7 +847,7 @@ function PipelineProcedureAttemptList({ procedure, matchedStepKey = "", matchedA
),
h(PipelineKvGrid, { items: [
{ label: "messages", value: messages.messageCount ?? "--" },
{ label: "steps", value: messages.stepCount ?? steps.length },
{ label: "steps", value: messages.stepCount ?? "--" },
{ label: "tools", value: messages.toolCallCount ?? "--" },
{ label: "updated", value: fmtDate(messages.updatedAt) },
{ label: "sessions", value: sessionIds.join(", ") || "--" },
@@ -886,8 +886,7 @@ function pipelineGanttDetailPayload(baseDetails: any, nodeDetails: any, runId: s
return {
...base,
...nodeDetails,
// Keep run-level evidence unless the node endpoint has a more focused copy.
controlCommands: asArray(nodeDetails.controlCommands).length > 0 ? nodeDetails.controlCommands : base.controlCommands,
// Keep normalized OA control events unless the node endpoint has a focused copy.
controlEvents: asArray(nodeDetails.controlEvents).length > 0 ? nodeDetails.controlEvents : base.controlEvents,
procedureRuns: nodeProcedures.length > 0 ? nodeProcedures : base.procedureRuns,
};
@@ -2342,12 +2341,12 @@ function pipelineGanttAddObservationArrows(markers: AnyRecord[], arrows: AnyReco
return { markers, arrows: nextArrows };
}
function pipelinePromptMarkerTone(record: any, fallbackKind = ""): string {
const kind = eventKind(record) || fallbackKind;
function pipelinePromptMarkerTone(record: any): string {
const kind = eventKind(record);
const promptEvent = String(record?.promptEvent || "");
if (kind === "initial-prompt-delivered") return "initial";
if (promptEvent === "node-finished" || promptEvent === "node-long-running-observation" || promptEvent.startsWith("monitor-")) return "monitor";
if (kind === "monitor-prompt-delivered" || String(record?.sourceKind || "").toLowerCase() === "monitor" || fallbackKind === "monitor-prompt-queued") return "monitor";
if (kind === "monitor-prompt-delivered" || String(record?.sourceKind || "").toLowerCase() === "monitor") return "monitor";
return "append";
}
@@ -2355,16 +2354,16 @@ function pipelineRecordTags(record: any): string[] {
return asArray(record?.tags || record?.raw?.tags).map((tag) => String(tag || "")).filter(Boolean);
}
function pipelinePromptMarkerLabel(record: any, fallbackKind = ""): string {
const kind = eventKind(record) || fallbackKind;
function pipelinePromptMarkerLabel(record: any): string {
const kind = eventKind(record);
const promptEvent = String(record?.promptEvent || "");
if (kind === "initial-prompt-delivered") return "初始 prompt";
if (promptEvent === "node-long-running-observation") return "长任务观察";
if (promptEvent === "node-finished") return pipelineRecordTags(record).includes("monitor.audit") ? "节点完成 / OA 审核" : "节点完成";
if (promptEvent === "monitor-interval") return "旧版轮询";
if (promptEvent === "monitor-interval") return "Monitor observation";
if (promptEvent === "monitor-start") return "Monitor start";
if (promptEvent === "monitor-stop") return "Monitor stop";
if (kind === "monitor-prompt-delivered" || fallbackKind === "monitor-prompt-queued") return "Monitor prompt";
if (kind === "monitor-prompt-delivered") return "Monitor prompt";
if (kind === "append-prompt-queued") return "追加 prompt 已排队";
return "追加 prompt";
}
@@ -2382,7 +2381,7 @@ function controlRecordGroupKey(record: any, index: number): string {
if (commandId) return `command:${commandId}`;
const timestamp = eventTimestampIso(record) || firstIso(record?.createdAt, record?.timestamp) || `index-${index}`;
return [
"fallback",
"control-event",
timestamp,
String(record?.sourceKind || ""),
String(record?.sourceNodeId || ""),
@@ -2481,8 +2480,6 @@ function pipelineRunGanttSignals(details: any, activeRun: any): AnyRecord {
if (!nodeId) continue;
for (const attempt of asArray(procedure?.attempts)) {
const attemptId = attemptLabel(attempt);
const deliveredIds = new Set<string>();
const deliveredFallbackKeys = new Set<string>();
for (const record of structuredEventRecords(attempt?.controlEventRecords)) {
const kind = eventKind(record);
if (!["initial-prompt-delivered", "append-prompt-delivered", "monitor-prompt-delivered"].includes(kind)) continue;
@@ -2490,8 +2487,6 @@ function pipelineRunGanttSignals(details: any, activeRun: any): AnyRecord {
const ms = timeMs(timestampIso);
if (ms === null) continue;
const eventId = String(record?.eventId || "");
if (eventId) deliveredIds.add(eventId);
deliveredFallbackKeys.add(`${kind}:${timestampIso}:${String(record?.sourceKind || "")}:${String(record?.promptPreview || "")}`);
addMarker({
id: `prompt:${eventId || `${procedureRunId}:${attemptId}:${kind}:${ms}`}`,
runId,
@@ -2499,55 +2494,19 @@ function pipelineRunGanttSignals(details: any, activeRun: any): AnyRecord {
procedureRunId,
attempt: attemptId,
kind: "prompt",
tone: pipelinePromptMarkerTone(record, kind),
tone: pipelinePromptMarkerTone(record),
status: "delivered",
label: pipelinePromptMarkerLabel(record, kind),
label: pipelinePromptMarkerLabel(record),
ms,
timestampIso,
sourceKind: String(record?.sourceKind || ""),
sourceNodeId: String(record?.sourceNodeId || ""),
targetNodeId: nodeId,
action: "",
eventId,
commandId: String(record?.commandId || ""),
raw: record,
}, promptMarkers);
}
const fallbackSources = [
{ records: structuredEventRecords(attempt?.controlPromptRecords), fallbackKind: "append-prompt-queued" },
{ records: structuredEventRecords(attempt?.monitorPromptRecords), fallbackKind: "monitor-prompt-queued" },
];
for (const source of fallbackSources) {
for (const record of source.records) {
const timestampIso = eventTimestampIso(record);
const ms = timeMs(timestampIso);
if (ms === null) continue;
const eventId = String(record?.eventId || "");
if (eventId && deliveredIds.has(eventId)) continue;
const deliveredKind = source.fallbackKind === "monitor-prompt-queued" ? "monitor-prompt-delivered" : "append-prompt-delivered";
const fallbackKey = `${deliveredKind}:${timestampIso}:${String(record?.sourceKind || "")}:${String(record?.promptPreview || "")}`;
if (deliveredFallbackKeys.has(fallbackKey)) continue;
addMarker({
id: `prompt-fallback:${eventId || `${procedureRunId}:${attemptId}:${source.fallbackKind}:${ms}`}`,
runId,
nodeId,
procedureRunId,
attempt: attemptId,
kind: "prompt",
tone: pipelinePromptMarkerTone(record, source.fallbackKind),
status: "queued",
label: pipelinePromptMarkerLabel(record, source.fallbackKind),
ms,
timestampIso,
sourceKind: String(record?.sourceKind || ""),
sourceNodeId: String(record?.sourceNodeId || ""),
targetNodeId: nodeId,
action: "",
eventId,
commandId: String(record?.commandId || ""),
raw: record,
}, promptMarkers);
}
sourceKind: String(record?.sourceKind || ""),
sourceNodeId: String(record?.sourceNodeId || ""),
targetNodeId: nodeId,
action: "",
eventId,
commandId: String(record?.commandId || ""),
raw: record,
}, promptMarkers);
}
}
}
@@ -2555,22 +2514,15 @@ function pipelineRunGanttSignals(details: any, activeRun: any): AnyRecord {
const grouped = new Map<string, AnyRecord>();
structuredEventRecords(details?.controlEvents).forEach((record: AnyRecord, index: number) => {
const key = controlRecordGroupKey(record, index);
const group = grouped.get(key) || { key, events: [], commands: [] };
const group = grouped.get(key) || { key, events: [] };
group.events.push(record);
grouped.set(key, group);
});
asArray(details?.controlCommands).filter(isRecord).forEach((record: AnyRecord, index: number) => {
const key = controlRecordGroupKey(record, index);
const group = grouped.get(key) || { key, events: [], commands: [] };
group.commands.push(record);
grouped.set(key, group);
});
for (const group of grouped.values()) {
const events = asArray(group.events).slice().sort((left: any, right: any) => controlEventPriority(right) - controlEventPriority(left));
const commands = asArray(group.commands);
const queuedRecord = asArray(group.events).find((record: any) => eventKind(record) === "control-command-queued") || commands[0] || null;
const outcomeRecord = events[0] || commands[0] || queuedRecord;
const queuedRecord = asArray(group.events).find((record: any) => eventKind(record) === "control-command-queued") || null;
const outcomeRecord = events[0] || queuedRecord;
if (!queuedRecord && !outcomeRecord) continue;
const sourceNodeId = String(queuedRecord?.sourceNodeId || outcomeRecord?.sourceNodeId || "");
const sourceKind = String(queuedRecord?.sourceKind || outcomeRecord?.sourceKind || "");
@@ -43,7 +43,13 @@ function ctx(): DevContainerContext {
return context;
}
async function startDevContainerPlan(plan: DevContainerPlan, options: { forceRecreate: boolean; verifyPing: boolean; prepareRuntime: boolean }): Promise<{
async function startDevContainerPlan(plan: DevContainerPlan, options: {
forceRecreate: boolean;
verifyPing: boolean;
prepareRuntime: boolean;
onCommandStart?: (command: { name: string; providerId: string; timeoutMs: number }) => void;
onCommandComplete?: (command: DevContainerCommandLog) => void;
}): Promise<{
ok: boolean;
providerId: string;
plan: DevContainerPlan;
@@ -52,8 +58,10 @@ async function startDevContainerPlan(plan: DevContainerPlan, options: { forceRec
}> {
const commands: DevContainerCommandLog[] = [];
const run = async (targetProviderId: string, script: string, timeoutMs: number, name: string): Promise<DevContainerCommandLog> => {
options.onCommandStart?.({ name, providerId: targetProviderId, timeoutMs });
const command = await ctx().runCodeQueueSsh(targetProviderId, script, timeoutMs, name);
commands.push(command);
options.onCommandComplete?.(command);
ctx().throwIfCommandFailed(command);
return command;
};
@@ -68,7 +76,7 @@ async function startDevContainerPlan(plan: DevContainerPlan, options: { forceRec
await run("main-server", ctx().masterProxyPrepareScript(plan), 30_000, "master-proxy-prepare");
await run(plan.providerId, ctx().containerTunnelStartScript(plan), 45_000, "remote-tunnel-start");
await run("main-server", ctx().masterProxyFinishScript(plan), 30_000, "master-proxy-finish");
if (options.prepareRuntime) await run(plan.providerId, ctx().remoteCodexRuntimePrepareScript(plan), 600_000, "remote-codex-runtime-prepare");
if (options.prepareRuntime) await run(plan.providerId, ctx().remoteCodexRuntimePrepareScript(plan), 1_800_000, "remote-codex-runtime-prepare");
const verification: Record<string, JsonValue> = {};
if (options.verifyPing) {
const before = await run("main-server", ctx().masterProxyEvidenceScript(plan), 15_000, "master-proxy-evidence-before-ping");
@@ -92,7 +100,18 @@ export async function ensureTaskExecutionContainer(task: QueueTask): Promise<voi
if (existing !== undefined) return existing;
const promise = (async () => {
ctx().appendOutput(task, "system", `ensuring provider=${plan.providerId} container=${plan.containerName} workdir=${task.cwd}\n`, "provider/container");
const result = await startDevContainerPlan(plan, { forceRecreate: false, verifyPing: false, prepareRuntime: true });
const result = await startDevContainerPlan(plan, {
forceRecreate: false,
verifyPing: false,
prepareRuntime: true,
onCommandStart: (command) => {
ctx().appendOutput(task, "system", `provider prepare start name=${command.name} target=${command.providerId} timeoutMs=${command.timeoutMs}\n`, "provider/container");
},
onCommandComplete: (command) => {
const status = command.exitCode === 0 ? "ok" : `failed exit=${command.exitCode ?? "null"} signal=${command.signal ?? "null"}`;
ctx().appendOutput(task, command.exitCode === 0 ? "system" : "error", `provider prepare ${status} name=${command.name} target=${command.providerId} durationMs=${command.durationMs}\n`, "provider/container");
},
});
ctx().appendOutput(task, "system", `provider container ready provider=${plan.providerId} container=${plan.containerName} commands=${result.commands.length}\n`, "provider/container");
ctx().logger("info", "task_provider_container_ready", {
taskId: task.id,
@@ -109,6 +109,18 @@ import {
transcriptChunkResponse,
} from "./queue-api";
import { ReferenceTaskLookupError, configureReferences, injectReferencedTaskContext, taskReferenceIds } from "./references";
import {
applyOaTraceStatsToTaskJson,
configureOaEvents,
publishCodeQueueQueueUpdated,
publishCodeQueueTaskUpdated,
publishCodeQueueTraceStatsSnapshot,
publishCodeQueueTraceStep,
outputTraceKind,
readOaTraceStatsForTask,
readOaTraceStatsForTaskAttempts,
readOaTraceStatsForTasks,
} from "./oa-events";
import { configureSelfTests, runJudgeInfraSelfTest, runQueueOrderingSelfTest, runReferenceInjectionSelfTest, runTracePortSelfTest } from "./self-tests";
import {
configureTaskView,
@@ -151,31 +163,8 @@ const queueIdPattern = /^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$/u;
const queueNameMaxLength = 80;
const config = readConfig();
type CodeQueueEventType = "hello" | "task-updated" | "task-step" | "queue-updated";
interface CodeQueueEvent {
sequence: number;
type: CodeQueueEventType;
at: string;
reason: string;
taskId?: string;
queueId?: string;
status?: TaskStatus;
updatedAt?: string;
stepCount?: number;
outputMaxSeq?: number;
}
interface CodeQueueEventClient {
enqueue: (chunk: string) => void;
close: () => void;
}
const codeQueueEventClients = new Set<CodeQueueEventClient>();
const codeQueueEventEncoder = new TextEncoder();
const taskEventSnapshots = new Map<string, { stepCount: number; outputMaxSeq: number; status: TaskStatus; updatedAt: string }>();
let codeQueueEventSequence = 0;
const logger = createLogger("code-queue", config.logFile);
configureOaEvents({ baseUrl: config.oaEventFlowBaseUrl, logger, nowIso });
const state = emptyState();
let processing = false;
const processingQueues = new Set<string>();
@@ -330,6 +319,7 @@ function readConfig(): RuntimeConfig {
turnNoActivityTimeoutMs: Math.max(60_000, Math.min(30 * 60_000, envNumber("CODEX_TURN_NO_ACTIVITY_TIMEOUT_MS", 6 * 60_000))),
databaseUrl: envRequiredString("DATABASE_URL"),
databaseFlushIntervalMs: Math.max(100, Math.min(10_000, envNumber("CODE_QUEUE_DATABASE_FLUSH_INTERVAL_MS", 1000))),
oaEventFlowBaseUrl: envString("OA_EVENT_FLOW_BASE_URL", "http://oa-event-flow:4255").replace(/\/+$/u, ""),
notifyClaudeQqEnabled: envBool("CODE_QUEUE_NOTIFY_CLAUDEQQ_ENABLED", false),
notifyClaudeQqBaseUrl: envString("CODE_QUEUE_NOTIFY_CLAUDEQQ_BASE_URL", "http://backend-core:8080/api/microservices/claudeqq/proxy").replace(/\/+$/u, ""),
notifyClaudeQqTargetType: notifyTargetTypeRaw === "group" ? "group" : "private",
@@ -610,142 +600,10 @@ function nowIso(): string {
return new Date().toISOString();
}
function sseChunk(event: CodeQueueEvent): string {
return [
`id: ${event.sequence}`,
`event: ${event.type}`,
`data: ${JSON.stringify(event)}`,
"",
"",
].join("\n");
}
function emitCodeQueueEvent(event: Omit<CodeQueueEvent, "sequence" | "at"> & { at?: string }): void {
const payload: CodeQueueEvent = {
...event,
sequence: ++codeQueueEventSequence,
at: event.at || nowIso(),
};
if (codeQueueEventClients.size === 0) return;
const chunk = sseChunk(payload);
for (const client of Array.from(codeQueueEventClients)) {
try {
client.enqueue(chunk);
} catch {
client.close();
}
}
}
function taskOutputMaxSeq(task: QueueTask): number {
return taskViewOutputMaxSeq(task);
}
function publishTaskEvent(task: QueueTask, reason: string, options: { onlyStepChange?: boolean; stepChanged?: boolean } = {}): void {
// Hot path: never build full transcripts or read output archives here.
const stepCount = taskListStepCount(task);
const outputMaxSeq = taskOutputMaxSeq(task);
const previous = taskEventSnapshots.get(task.id);
const stepChanged = options.stepChanged ?? (previous === undefined ? stepCount > 0 : stepCount > previous.stepCount);
if (options.onlyStepChange === true && !stepChanged) {
taskEventSnapshots.set(task.id, { stepCount, outputMaxSeq, status: task.status, updatedAt: task.updatedAt });
return;
}
const type: CodeQueueEventType = stepChanged ? "task-step" : "task-updated";
taskEventSnapshots.set(task.id, { stepCount, outputMaxSeq, status: task.status, updatedAt: task.updatedAt });
emitCodeQueueEvent({
type,
reason,
taskId: task.id,
queueId: queueIdOf(task),
status: task.status,
updatedAt: task.updatedAt,
stepCount,
outputMaxSeq,
});
}
function publishQueueEvent(reason: string, queueId = ""): void {
emitCodeQueueEvent({ type: "queue-updated", reason, queueId: queueId || undefined });
}
function outputCanChangeStepCount(output: LiveOutput): boolean {
return output.channel === "command" || output.channel === "diff" || output.channel === "tool";
}
function outputStartsTraceStep(output: LiveOutput): boolean {
if (output.channel === "diff" || output.channel === "tool") return true;
if (output.channel !== "command") return false;
const method = String(output.method || "");
return method === "item/started" || (method !== "item/commandExecution/outputDelta" && method !== "item/completed");
}
function recordTaskOutputMetrics(task: QueueTask, output: LiveOutput, op: "set" | "append"): boolean {
task.outputMaxSeq = Math.max(taskOutputMaxSeq(task), Number.isFinite(Number(output.seq)) ? Number(output.seq) : 0);
if (op === "append" || !outputStartsTraceStep(output)) return false;
const storedStepCount = Number(task.stepCount ?? task.llmStepCount);
const nextStepCount = Number.isFinite(storedStepCount) && storedStepCount >= 0
? Math.floor(storedStepCount) + 1
: task.output.reduce((count, item) => count + (outputStartsTraceStep(item) ? 1 : 0), 0);
task.stepCount = nextStepCount;
task.llmStepCount = nextStepCount;
return true;
}
function codeQueueEventsResponse(): Response {
let heartbeat: ReturnType<typeof setInterval> | null = null;
let client: CodeQueueEventClient | null = null;
const stream = new ReadableStream<Uint8Array>({
start(controller) {
client = {
enqueue(chunk: string) {
controller.enqueue(codeQueueEventEncoder.encode(chunk));
},
close() {
if (heartbeat !== null) {
clearInterval(heartbeat);
heartbeat = null;
}
if (client !== null) codeQueueEventClients.delete(client);
client = null;
try {
controller.close();
} catch {
// The browser may have already closed the EventSource connection.
}
},
};
codeQueueEventClients.add(client);
controller.enqueue(codeQueueEventEncoder.encode("retry: 1000\n\n"));
client.enqueue(sseChunk({
sequence: codeQueueEventSequence,
type: "hello",
at: nowIso(),
reason: "connected",
}));
heartbeat = setInterval(() => {
try {
controller.enqueue(codeQueueEventEncoder.encode(`: heartbeat ${nowIso()}\n\n`));
} catch {
client?.close();
}
}, 25_000);
heartbeat.unref?.();
},
cancel() {
client?.close();
},
});
return new Response(stream, {
headers: {
"content-type": "text/event-stream; charset=utf-8",
"cache-control": "no-store, no-transform",
"connection": "keep-alive",
"x-accel-buffering": "no",
},
});
}
function resolveReasoningEffort(model: string, explicit?: string | null): string | null {
const requested = explicit?.trim();
if (requested) return requested;
@@ -869,7 +727,7 @@ function pruneTaskHotState(task: QueueTask): void {
}
function taskRetainedTraceStepCount(task: QueueTask): number {
return task.output.reduce((count, output) => count + (outputStartsTraceStep(output) ? 1 : 0), 0);
return task.output.reduce((count, output) => count + (outputStartsTraceStep(task, output) ? 1 : 0), 0);
}
function normalizeTaskMetric(value: unknown): number | null {
@@ -939,6 +797,104 @@ function redactDatabaseUrl(value: string): string {
}
}
function publishTaskOaEvent(task: QueueTask, reason: string, options: { onlyStepChange?: boolean; stepChanged?: boolean } = {}): void {
const stepCount = taskListStepCount(task);
const outputMaxSeq = taskOutputMaxSeq(task);
if (options.onlyStepChange === true && options.stepChanged !== true) return;
const queueId = queueIdOf(task);
publishCodeQueueTaskUpdated(task, queueId, reason, stepCount, outputMaxSeq);
publishCodeQueueTraceStatsSnapshot(task, queueId, reason, stepCount, outputMaxSeq);
}
function publishQueueEvent(reason: string, queueId = ""): void {
publishCodeQueueQueueUpdated(queueId, reason);
}
function isOpenCodeStepBoundaryMethod(method: string | undefined): boolean {
return method === "opencode/step-start" || method === "opencode/step-finish";
}
function outputCanChangeStepCount(output: LiveOutput): boolean {
if (output.channel === "user" && output.method === "enqueue") return false;
return !isOpenCodeStepBoundaryMethod(output.method);
}
function commandStartedBeforeIn(outputs: LiveOutput[], output: LiveOutput): boolean {
if (typeof output.itemId !== "string") return false;
return outputs.some((item) => item !== output && item.itemId === output.itemId && item.channel === "command" && item.method === "item/started");
}
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 === "user") return true;
if (output.channel !== "command") return true;
const method = String(output.method || "");
if (method === "item/started") return true;
if (method === "item/commandExecution/outputDelta") return false;
if (method === "item/completed") return !commandStartedBeforeIn(outputs, output);
return true;
}
function outputStartsTraceStep(task: QueueTask, output: LiveOutput): boolean {
return outputStartsTraceStepInHistory(task.output, output);
}
function traceStatsFromOutputs(outputs: LiveOutput[]): { stepCount: number; readCount: number; editCount: number; runCount: number; errorCount: number } {
const visibleOutputs = outputs.filter((output) => outputStartsTraceStepInHistory(outputs, output));
const stats = { stepCount: visibleOutputs.length, readCount: 0, editCount: 0, runCount: 0, errorCount: 0 };
for (const output of visibleOutputs) {
const kind = outputTraceKind(output);
if (kind === "read") stats.readCount += 1;
if (kind === "edit") stats.editCount += 1;
if (kind === "run") stats.runCount += 1;
if (kind === "error") stats.errorCount += 1;
}
return stats;
}
function attemptIndexFromOutput(output: LiveOutput): number | null {
const match = String(output.text || "").match(/\battempt\s+(\d+)\s*\/\s*\d+\b/iu);
const value = Number(match?.[1] ?? NaN);
return Number.isInteger(value) && value > 0 ? value : null;
}
function outputAttemptIndexMap(outputs: LiveOutput[]): Map<number, number> {
const result = new Map<number, number>();
let currentAttempt = 0;
for (const output of outputs.slice().sort((left, right) => Number(left.seq) - Number(right.seq))) {
const startedAttempt = attemptIndexFromOutput(output);
if (startedAttempt !== null) currentAttempt = startedAttempt;
if (currentAttempt > 0 && outputStartsTraceStepInHistory(outputs, output)) result.set(output.seq, currentAttempt);
}
return result;
}
function traceAttemptIndexesForTask(task: QueueTask): number[] {
const indexes = new Set<number>();
for (const attempt of task.attempts) {
const index = Number(attempt.index);
if (Number.isInteger(index) && index > 0) indexes.add(index);
}
const currentAttempt = Number(task.currentAttempt || 0);
const maxAttempt = Math.max(currentAttempt, ...Array.from(indexes), 0);
for (let index = 1; index <= maxAttempt; index += 1) indexes.add(index);
return Array.from(indexes).sort((left, right) => left - right);
}
function recordTaskOutputMetrics(task: QueueTask, output: LiveOutput, op: "set" | "append"): boolean {
task.outputMaxSeq = Math.max(taskOutputMaxSeq(task), Number.isFinite(Number(output.seq)) ? Number(output.seq) : 0);
if (op === "append" || !outputStartsTraceStep(task, output)) return false;
const storedStepCount = Number(task.stepCount ?? task.llmStepCount);
const nextStepCount = Number.isFinite(storedStepCount) && storedStepCount >= 0
? Math.floor(storedStepCount) + 1
: task.output.reduce((count, item) => count + (outputStartsTraceStep(task, item) ? 1 : 0), 0);
task.stepCount = nextStepCount;
task.llmStepCount = nextStepCount;
return true;
}
function errorToJson(error: unknown): JsonValue {
if (error instanceof Error) return { name: error.name, message: error.message, stack: error.stack ?? null };
return String(error);
@@ -957,7 +913,7 @@ function markTaskDirty(taskId: string): void {
function persistTaskState(task: QueueTask): void {
markTaskDirty(task.id);
persistState(false);
publishTaskEvent(task, "persist");
publishTaskOaEvent(task, "persist");
}
function markQueueDirty(queueId: string): void {
@@ -1780,7 +1736,7 @@ function addEvent(task: QueueTask, event: CodexEventSummary): void {
task.events.push(event);
if (config.maxInMemoryEventRecords > 0 && task.events.length > config.maxInMemoryEventRecords) task.events.splice(0, task.events.length - config.maxInMemoryEventRecords);
markTaskDirty(task.id);
publishTaskEvent(task, "agent-event", { onlyStepChange: true });
publishTaskOaEvent(task, "agent-event", { onlyStepChange: true });
}
function taskReferencesEqual(left: string[], right: string[]): boolean {
@@ -1970,8 +1926,9 @@ configureTaskOutput({
onOutputAppended: (task, output, op) => {
const archiveOp = op === "append" ? "append" : "set";
const stepChanged = recordTaskOutputMetrics(task, output, archiveOp);
if (stepChanged) publishCodeQueueTraceStep(task, queueIdOf(task), output, taskOutputMaxSeq(task));
if (archiveOp === "append" && !outputCanChangeStepCount(output)) return;
publishTaskEvent(task, "output", { onlyStepChange: archiveOp === "append", stepChanged });
publishTaskOaEvent(task, "output", { onlyStepChange: archiveOp === "append", stepChanged });
},
schedulePersistState,
});
@@ -2050,6 +2007,7 @@ configureQueueApi({
safeQueueName,
sql,
taskQueueEnteredAt,
traceStatsForTasks: readOaTraceStatsForTasks,
tasks: () => state.tasks,
truthyParam,
});
@@ -2928,7 +2886,7 @@ async function createTasks(req: Request): Promise<Response> {
for (const task of tasks) appendOutput(task, "user", `${task.prompt}\n`, "enqueue");
state.tasks.push(...tasks);
if (tasks.length > 0) armIdleNotification();
for (const task of tasks) publishTaskEvent(task, "enqueue");
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))) });
@@ -3057,7 +3015,7 @@ async function markTaskRead(task: QueueTask): Promise<Response> {
task.readAt = nowIso();
markTaskDirty(task.id);
persistState(false);
publishTaskEvent(task, "read");
publishTaskOaEvent(task, "read");
logger("info", "task_marked_read", { taskId: task.id, queueId: queueIdOf(task), status: task.status });
}
await flushDirtyTasksToDatabase(true);
@@ -3102,7 +3060,7 @@ async function markTaskReadById(taskId: string): Promise<Response> {
const hotTask = findTask(taskId);
if (hotTask !== null) {
hotTask.readAt = readAt;
publishTaskEvent(hotTask, "read");
publishTaskOaEvent(hotTask, "read");
}
logger("info", "task_marked_read", { taskId, queueId: safeQueueId(row.queue_id), status: row.status });
}
@@ -3148,7 +3106,7 @@ async function markTerminalTasksRead(url: URL): Promise<Response> {
for (const task of state.tasks) {
if (!ids.has(task.id)) continue;
task.readAt = readAt;
publishTaskEvent(task, "read-all");
publishTaskOaEvent(task, "read-all");
}
if (ids.size > 0) {
logger("info", "terminal_tasks_marked_read", { count: ids.size, queueId });
@@ -3161,7 +3119,7 @@ async function markTerminalTasksRead(url: URL): Promise<Response> {
if (!terminalTaskUnread(task)) continue;
task.readAt = readAt;
markTaskDirty(task.id);
publishTaskEvent(task, "read-all");
publishTaskOaEvent(task, "read-all");
count += 1;
}
if (count > 0) {
@@ -3269,6 +3227,40 @@ async function mergeDatabaseQueueTasks(sourceQueueIds: string[], targetQueueId:
return rows.map((row) => row.id);
}
async function deleteDatabaseQueues(queueIds: string[]): Promise<string[]> {
if (!databaseReady || queueIds.length === 0) return [];
const rows = await sql<Array<{ id: string }>>`
DELETE FROM unidesk_code_queue_queues
WHERE id IN ${sql(queueIds)}
RETURNING id
`;
return rows.map((row) => row.id);
}
function queueSnapshot(queueId: string, timestamp: string): QueueRecord {
const queue = knownQueue(queueId);
return queue === null
? { id: queueId, name: queueId, createdAt: timestamp, updatedAt: timestamp }
: { ...queue };
}
function deleteQueuesFromState(queueIds: string[]): QueueRecord[] {
const queueIdSet = new Set(queueIds);
const deletedQueues: QueueRecord[] = [];
const retainedQueues: QueueRecord[] = [];
for (const queue of state.queues) {
if (queueIdSet.has(queue.id)) {
deletedQueues.push({ ...queue });
dirtyDatabaseQueueIds.delete(queue.id);
} else {
retainedQueues.push(queue);
}
}
state.queues.splice(0, state.queues.length, ...retainedQueues);
for (const queueId of queueIds) dirtyDatabaseQueueIds.delete(queueId);
return deletedQueues;
}
async function mergeQueues(targetQueueIdValue: string | null, req: Request): Promise<Response> {
const body = await readJson(req);
const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
@@ -3289,13 +3281,9 @@ async function mergeQueues(targetQueueIdValue: string | null, req: Request): Pro
const mergedAt = nowIso();
const targetQueue = ensureQueue(targetQueueId);
const sourceQueues = sourceQueueIds.map((id) => ensureQueue(id));
const sourceQueues = sourceQueueIds.map((id) => queueSnapshot(id, mergedAt));
targetQueue.updatedAt = mergedAt;
markQueueDirty(targetQueue.id);
for (const queue of sourceQueues) {
queue.updatedAt = mergedAt;
markQueueDirty(queue.id);
}
const sourceSet = new Set(sourceQueueIds);
const hotMovedTasks: QueueTask[] = [];
@@ -3306,18 +3294,22 @@ async function mergeQueues(targetQueueIdValue: string | null, req: Request): Pro
task.queueId = targetQueueId;
hotMovedTasks.push(task);
markTaskDirty(task.id);
publishTaskEvent(task, "queue-merged");
publishTaskOaEvent(task, "queue-merged");
}
const databaseMovedTaskIds = await mergeDatabaseQueueTasks(sourceQueueIds, targetQueueId);
const deletedSourceQueues = deleteQueuesFromState(sourceQueueIds);
const databaseDeletedQueueIds = await deleteDatabaseQueues(sourceQueueIds);
persistState(false);
publishQueueEvent("queue-merged", targetQueueId);
for (const sourceQueueId of sourceQueueIds) publishQueueEvent("queue-merged", sourceQueueId);
for (const sourceQueueId of sourceQueueIds) publishQueueEvent("queue-deleted-after-merge", sourceQueueId);
if (hotMovedTasks.some((task) => task.status === "queued" || task.status === "retry_wait")) armIdleNotification();
logger("info", "queues_merged", {
targetQueueId,
sourceQueueIds,
deletedSourceQueueIds: deletedSourceQueues.map((queue) => queue.id),
hotMovedTaskCount: hotMovedTasks.length,
databaseMovedTaskCount: databaseReady ? databaseMovedTaskIds.length : null,
databaseDeletedQueueIds: databaseReady ? databaseDeletedQueueIds : null,
});
for (const id of mergeQueueIds) mergingQueues.delete(id);
scheduleQueue(targetQueueId);
@@ -3339,9 +3331,11 @@ async function mergeQueues(targetQueueIdValue: string | null, req: Request): Pro
mergedTaskCount: databaseReady ? databaseMovedTaskIds.length : hotMovedTasks.length,
movedTaskIds: orderedMovedTaskIds.slice(0, 500),
targetTaskOrder: targetTaskOrder.slice(0, 500),
order: "merged tasks keep their original queueEnteredAt/createdAt ordering; source queue records are retained and never deleted",
order: "merged tasks keep their original queueEnteredAt/createdAt ordering; source queue records are deleted after merge",
targetQueue,
sourceQueues,
deletedSourceQueues: deletedSourceQueues.length > 0 ? deletedSourceQueues : sourceQueues,
deletedSourceQueueIds: sourceQueueIds,
queues: perQueueSummaries(tasks),
summary: queueSummary(false, tasks),
}, 202);
@@ -3376,13 +3370,41 @@ async function moveTaskToQueue(task: QueueTask, req: Request): Promise<Response>
return jsonResponse({ ok: true, task: taskForResponse(task), queue: await queueSummaryForResponse() }, 202);
}
async function backfillOaTraceStats(url: URL): Promise<JsonValue> {
const limitRaw = Number(url.searchParams.get("limit") ?? 2000);
const limit = Number.isInteger(limitRaw) && limitRaw > 0 ? Math.min(20_000, limitRaw) : 2000;
const taskId = url.searchParams.get("taskId")?.trim() ?? "";
const tasks = taskId.length > 0
? [await findTaskForRead(taskId)].filter((task): task is QueueTask => task !== null)
: (await loadAllTasksForRead()).slice(-limit);
const includeSteps = url.searchParams.get("steps") !== "0";
let stepEventCount = 0;
for (const task of tasks) {
const queueId = queueIdOf(task);
const outputMaxSeq = taskOutputMaxSeq(task);
const output = taskFullOutput(task);
const traceStats = traceStatsFromOutputs(output);
const attemptBySeq = outputAttemptIndexMap(output);
if (includeSteps) {
for (const item of output) {
if (!outputStartsTraceStepInHistory(output, item)) continue;
publishCodeQueueTraceStep(task, queueId, item, outputMaxSeq, attemptBySeq.get(item.seq) ?? null);
stepEventCount += 1;
}
}
publishCodeQueueTraceStatsSnapshot(task, queueId, "backfill", traceStats.stepCount, outputMaxSeq, traceStats);
}
logger("info", "oa_trace_stats_backfill_enqueued", { taskCount: tasks.length, limit, taskId: taskId || null, includeSteps, stepEventCount });
return { ok: true, taskCount: tasks.length, limit, taskId: taskId || null, includeSteps, stepEventCount, eventFlowBaseUrl: config.oaEventFlowBaseUrl } as unknown as JsonValue;
}
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 === "/logs") return jsonResponse({ ok: true, logs: recentLogs.slice(-parseLimit(url)) });
if (url.pathname === "/api/events" && req.method === "GET") return codeQueueEventsResponse();
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() });
if (url.pathname === "/api/dev-containers" && req.method === "GET") {
const plan = buildDevContainerPlan(normalizeProviderId(config.devContainerDefaultProviderId) ?? "D601", {});
@@ -3413,6 +3435,7 @@ async function route(req: Request): Promise<Response> {
if (url.pathname === "/api/queue-order/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(runQueueOrderingSelfTest());
if (url.pathname === "/api/reference-injection/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(await runReferenceInjectionSelfTest());
if (url.pathname === "/api/trace-port/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(runTracePortSelfTest());
if (url.pathname === "/api/oa/backfill" && (req.method === "GET" || req.method === "POST")) return jsonResponse(await backfillOaTraceStats(url));
if (url.pathname === "/api/notifications/claudeqq" && req.method === "GET") {
await loadClaudeQqNotificationOutboxFromDatabase();
const limit = parseLimit(url);
@@ -3462,7 +3485,8 @@ async function route(req: Request): Promise<Response> {
.filter((task) => taskMatchesSearch(task, searchTerms));
const limit = parseLimit(url);
const page = taskPageRows(filteredTasks, url, limit);
const tasks = page.rows.map((task) => taskForListResponse(task, lite, allTasks));
const traceStats = await readOaTraceStatsForTasks(page.rows.map((task) => task.id));
const tasks = page.rows.map((task) => applyOaTraceStatsToTaskJson(taskForListResponse(task, lite, allTasks), traceStats.get(`task:${task.id}`) ?? null));
return jsonResponse({
ok: true,
queue: queueSummary(includeDevReady, allTasks),
@@ -3507,7 +3531,8 @@ async function route(req: Request): Promise<Response> {
if (traceSummaryMatch !== null && req.method === "GET") {
const task = await findTaskForRead(decodeURIComponent(traceSummaryMatch[1] ?? ""));
if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
return jsonResponse({ ok: true, summary: taskTraceSummaryResponse(task) });
const traceStats = await readOaTraceStatsForTaskAttempts(task.id, traceAttemptIndexesForTask(task));
return jsonResponse({ ok: true, summary: taskTraceSummaryResponse(task, traceStats.get(`task:${task.id}`) ?? null, traceStats) });
}
const traceStepsMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/trace-steps$/u);
if (traceStepsMatch !== null && req.method === "GET") {
@@ -3543,9 +3568,10 @@ async function route(req: Request): Promise<Response> {
if (action === "edit" && (req.method === "POST" || req.method === "PATCH")) return await editQueuedTaskPrompt(task, req);
if (action !== undefined) return jsonResponse({ ok: false, error: "not found" }, 404);
if (req.method === "GET") {
if (url.searchParams.get("meta") === "1") return jsonResponse({ ok: true, task: taskForMetaResponse(task) });
const traceStats = await readOaTraceStatsForTask(task.id);
if (url.searchParams.get("meta") === "1") return jsonResponse({ ok: true, task: applyOaTraceStatsToTaskJson(taskForMetaResponse(task), traceStats) });
const includeRaw = url.searchParams.get("raw") === "1" || url.searchParams.get("full") === "1";
return jsonResponse({ ok: true, task: taskForResponse(task, true, includeRaw) });
return jsonResponse({ ok: true, task: applyOaTraceStatsToTaskJson(taskForResponse(task, true, includeRaw), traceStats) });
}
if (req.method === "DELETE") return await interruptTask(task);
return jsonResponse({ ok: false, error: "method not allowed" }, 405);
@@ -3578,5 +3604,6 @@ const startupRecovered = queueActiveTasksForRestartRetry("Service restarted whil
if (startupRecovered > 0) logger("warn", "startup_requeued_active_tasks", { recovered: startupRecovered });
persistState();
serviceReady = true;
setTimeout(() => { void backfillOaTraceStats(new URL("http://code-queue.local/api/oa/backfill?limit=2000")).catch((error) => logger("warn", "oa_trace_stats_startup_backfill_failed", { error: errorToJson(error) })); }, 1000).unref?.();
scheduleQueue();
scheduleClaudeQqNotificationDrain(1000);
@@ -0,0 +1,370 @@
import { createHash } from "node:crypto";
import type { JsonValue, LiveOutput, QueueTask, TaskStatus } from "./types";
type JsonRecord = Record<string, JsonValue>;
interface OaEventContext {
baseUrl: string;
logger: (level: "debug" | "info" | "warn" | "error", message: string, data?: JsonValue) => void;
nowIso: () => string;
}
interface OaEventEnvelope {
eventId: string;
type: string;
createdAt: string;
sourceKind: string;
sourceId: string;
aggregateType: string;
aggregateId: string;
correlationId: string;
causationId?: string | null;
tags: string[];
payload: JsonRecord;
}
export interface OaTraceStats extends JsonRecord {
scopeId: string;
source: "oa-event-flow";
}
const postTimeoutMs = 2500;
let context: OaEventContext | null = null;
export function configureOaEvents(runtimeContext: OaEventContext): void {
const baseUrl = runtimeContext.baseUrl.replace(/\/+$/u, "");
if (baseUrl.length === 0) throw new Error("OA_EVENT_FLOW_BASE_URL is required");
context = { ...runtimeContext, baseUrl };
}
function ctx(): OaEventContext {
if (context === null) throw new Error("oa-events module is not configured");
return context;
}
function safePreview(value: string, max = 300): string {
const clean = String(value || "").replace(/\u001b\[[0-9;]*m/gu, "").trim();
return clean.length > max ? `${clean.slice(0, max)}...` : clean;
}
function hash(value: unknown): string {
return createHash("sha256").update(JSON.stringify(value)).digest("hex").slice(0, 20);
}
function addTag(tags: string[], value: string): void {
const tag = value.trim();
if (tag.length > 0 && !tags.includes(tag)) tags.push(tag);
}
function taskTags(task: QueueTask, queueId: string, extra: string[] = [], attemptIndexOverride: number | null = null): string[] {
const tags = ["service:code-queue", "trace", `task:${task.id}`, `queue:${queueId}`];
const attempt = Number(attemptIndexOverride ?? (task.currentAttempt || task.attempts.length || 0));
if (attempt > 0) addTag(tags, `attempt:${attempt}`);
for (const tag of extra) addTag(tags, tag);
return tags;
}
function postOaEvent(event: OaEventEnvelope): void {
const runtime = ctx();
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), postTimeoutMs);
void fetch(`${runtime.baseUrl}/api/events`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify(event),
signal: controller.signal,
}).then(async (response) => {
if (!response.ok) {
runtime.logger("warn", "oa_event_publish_failed", { eventId: event.eventId, type: event.type, status: response.status, body: safePreview(await response.text(), 1000) });
}
}).catch((error) => {
runtime.logger("warn", "oa_event_publish_error", { eventId: event.eventId, type: event.type, error: error instanceof Error ? error.message : String(error) });
}).finally(() => clearTimeout(timer));
}
function normalizeCommandText(text: string): string {
const trimmed = String(text || "").trim();
if (!trimmed.startsWith("{")) return trimmed;
try {
const parsed = JSON.parse(trimmed) as Record<string, unknown>;
const command = parsed.command ?? parsed.cmd ?? parsed.text ?? parsed.summary;
return typeof command === "string" ? command : trimmed;
} catch {
return trimmed;
}
}
function commandKind(command: string): "read" | "edit" | "run" {
if (/\b(apply_patch|git apply|cat >|tee .*<<|sed -i|python3? .*write_text|write|patch|edit|delete|create)\b/iu.test(command)) return "edit";
if (/\b(rg|grep|find|ls|cat|sed -n|tail|head|git status|git diff|ps|read|glob|search|view)\b/iu.test(command)) return "read";
return "run";
}
function openCodeToolRecord(output: LiveOutput): Record<string, unknown> | null {
if (!String(output.method || "").startsWith("opencode/")) return null;
try {
const parsed = JSON.parse(output.text) as unknown;
return typeof parsed === "object" && parsed !== null && !Array.isArray(parsed) ? parsed as Record<string, unknown> : null;
} catch {
return null;
}
}
function recordString(record: Record<string, unknown> | null, keys: string[]): string {
if (record === null) return "";
for (const key of keys) {
const value = record[key];
if (typeof value === "string") return value;
}
return "";
}
export function outputTraceKind(output: LiveOutput): "read" | "edit" | "run" | "error" | "message" | "system" {
if (output.channel === "diff") return "edit";
if (output.channel === "error") return "error";
if (output.channel === "assistant" || output.channel === "user" || output.channel === "reasoning") return "message";
if (output.channel === "tool") {
const record = openCodeToolRecord(output);
const part = record?.part && typeof record.part === "object" && !Array.isArray(record.part) ? record.part as Record<string, unknown> : null;
const state = part?.state && typeof part.state === "object" && !Array.isArray(part.state) ? part.state as Record<string, unknown> : null;
const input = state?.input && typeof state.input === "object" && !Array.isArray(state.input) ? state.input as Record<string, unknown> : null;
const tool = recordString(part, ["tool", "title"]) || recordString(record, ["type", "event", "name"]);
const command = recordString(input, ["command", "cmd", "script", "path", "pattern", "query"]);
return commandKind(`${tool} ${command}`);
}
if (output.channel === "command") return commandKind(normalizeCommandText(output.text));
return "system";
}
function outputTitle(output: LiveOutput, kind: string): string {
const method = String(output.method || "");
if (output.channel === "diff") return "Edited files";
if (output.channel === "error") return "Error";
if (output.channel === "assistant") return "Assistant message";
if (output.channel === "reasoning") return "Reasoning";
if (output.channel === "user") return method === "enqueue" ? "Submitted prompt" : "User prompt";
const command = normalizeCommandText(output.text).split(/\r?\n/u).find((line) => line.trim().length > 0)?.trim() ?? "";
if (command.length > 0) return safePreview(command, 160);
return kind === "read" ? "Read" : kind === "edit" ? "Edit" : kind === "run" ? "Run" : method || output.channel;
}
function outputSummaryLines(output: LiveOutput): string[] {
return String(output.text || "")
.replace(/\u001b\[[0-9;]*m/gu, "")
.split(/\r?\n/u)
.map((line) => line.trim())
.filter(Boolean)
.slice(0, 4)
.map((line) => safePreview(line, 220));
}
function outputAttemptIndex(output: LiveOutput): number | null {
const match = String(output.text || "").match(/\battempt\s+(\d+)\s*\/\s*\d+\b/iu);
const value = Number(match?.[1] ?? NaN);
return Number.isInteger(value) && value > 0 ? value : null;
}
function taskOutputAttemptIndex(task: QueueTask, output: LiveOutput, attemptIndexOverride: number | null = null): number | null {
if (attemptIndexOverride !== null && Number.isInteger(attemptIndexOverride) && attemptIndexOverride > 0) return attemptIndexOverride;
const fromOutput = outputAttemptIndex(output);
if (fromOutput !== null) return fromOutput;
const current = Number(task.currentAttempt || 0);
return Number.isInteger(current) && current > 0 ? current : null;
}
function taskStatusPayload(task: QueueTask, queueId: string, reason: string, stepCount: number, outputMaxSeq: number): JsonRecord {
return {
taskId: task.id,
queueId,
reason,
status: task.status,
updatedAt: task.updatedAt,
createdAt: task.createdAt,
startedAt: task.startedAt,
finishedAt: task.finishedAt,
readAt: task.readAt,
providerId: task.providerId,
model: task.model,
currentAttempt: task.currentAttempt,
stepCount,
outputMaxSeq,
};
}
export function publishCodeQueueTaskUpdated(task: QueueTask, queueId: string, reason: string, stepCount: number, outputMaxSeq: number): void {
postOaEvent({
eventId: `code-queue:task-updated:${task.id}:${reason}:${task.updatedAt}:${outputMaxSeq}:${hash({ status: task.status, queueId, readAt: task.readAt })}`,
type: "task-updated",
createdAt: ctx().nowIso(),
sourceKind: "service",
sourceId: "code-queue",
aggregateType: "task",
aggregateId: task.id,
correlationId: task.id,
tags: taskTags(task, queueId, ["task"]),
payload: taskStatusPayload(task, queueId, reason, stepCount, outputMaxSeq),
});
}
export function publishCodeQueueTraceStatsSnapshot(
task: QueueTask,
queueId: string,
reason: string,
stepCount: number,
outputMaxSeq: number,
statsOverride: Partial<{ llmStepCount: number; traceLineCount: number; readCount: number; editCount: number; runCount: number; errorCount: number }> = {},
): void {
postOaEvent({
eventId: `code-queue:trace-stats-snapshot:${task.id}:${task.updatedAt}:${stepCount}:${outputMaxSeq}`,
type: "trace-stats-snapshot",
createdAt: ctx().nowIso(),
sourceKind: "service",
sourceId: "code-queue",
aggregateType: "task",
aggregateId: task.id,
correlationId: task.id,
tags: taskTags(task, queueId, ["stats"]),
payload: {
...taskStatusPayload(task, queueId, reason, stepCount, outputMaxSeq),
scopeId: `task:${task.id}`,
stats: {
stepCount,
llmStepCount: statsOverride.llmStepCount ?? stepCount,
traceLineCount: statsOverride.traceLineCount ?? stepCount,
outputMaxSeq,
readCount: statsOverride.readCount ?? 0,
editCount: statsOverride.editCount ?? 0,
runCount: statsOverride.runCount ?? 0,
errorCount: statsOverride.errorCount ?? (task.lastError ? 1 : 0),
attempts: task.attempts.map((attempt) => ({ index: attempt.index, mode: attempt.mode, startedAt: attempt.startedAt, finishedAt: attempt.finishedAt, terminalStatus: attempt.terminalStatus })),
},
},
});
}
export function publishCodeQueueTraceStep(task: QueueTask, queueId: string, output: LiveOutput, outputMaxSeq: number, attemptIndexOverride: number | null = null): void {
const kind = outputTraceKind(output);
const attemptIndex = taskOutputAttemptIndex(task, output, attemptIndexOverride);
const attemptScopeId = attemptIndex === null ? null : taskAttemptScopeId(task.id, attemptIndex);
postOaEvent({
eventId: `code-queue:trace-step-created:${task.id}:${output.seq}`,
type: "trace-step-created",
createdAt: output.at || ctx().nowIso(),
sourceKind: "service",
sourceId: "code-queue",
aggregateType: "task",
aggregateId: task.id,
correlationId: task.id,
tags: taskTags(task, queueId, attemptScopeId === null ? [] : [attemptScopeId], attemptIndex),
payload: {
taskId: task.id,
queueId,
scopeId: `task:${task.id}`,
attemptIndex,
attemptScopeId,
seq: output.seq,
outputSeq: output.seq,
outputMaxSeq,
kind,
channel: output.channel,
method: output.method ?? "",
itemId: output.itemId ?? "",
title: outputTitle(output, kind),
status: task.status,
summaryLines: outputSummaryLines(output),
rawSeqs: [output.seq],
},
});
}
export function publishCodeQueueQueueUpdated(queueId: string, reason: string): void {
const at = ctx().nowIso();
postOaEvent({
eventId: `code-queue:queue-updated:${queueId || "all"}:${reason}:${at}`,
type: "queue-updated",
createdAt: at,
sourceKind: "service",
sourceId: "code-queue",
aggregateType: "queue",
aggregateId: queueId || "all",
correlationId: `queue:${queueId || "all"}`,
tags: ["service:code-queue", "queue", ...(queueId ? [`queue:${queueId}`] : [])],
payload: { queueId, reason, updatedAt: at },
});
}
export async function readOaTraceStatsForScopeIds(scopeIds: string[]): Promise<Map<string, OaTraceStats>> {
const result = new Map<string, OaTraceStats>();
const uniqueScopeIds = Array.from(new Set(scopeIds.map((id) => String(id || "").trim()).filter(Boolean)));
if (uniqueScopeIds.length === 0) return result;
const runtime = ctx();
const url = new URL(`${runtime.baseUrl}/api/stats/trace`);
url.searchParams.set("scopeIds", uniqueScopeIds.join(","));
url.searchParams.set("limit", String(Math.max(100, uniqueScopeIds.length)));
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), 2500);
try {
const response = await fetch(url, { signal: controller.signal });
if (!response.ok) throw new Error(`status=${response.status}`);
const body = await response.json() as Record<string, unknown>;
const stats = Array.isArray(body.stats) ? body.stats : [];
for (const item of stats) {
if (typeof item !== "object" || item === null || Array.isArray(item)) continue;
const record = item as OaTraceStats;
if (typeof record.scopeId !== "string") continue;
result.set(record.scopeId, record);
}
} catch (error) {
runtime.logger("warn", "oa_trace_stats_read_failed", { scopeCount: uniqueScopeIds.length, error: error instanceof Error ? error.message : String(error) });
} finally {
clearTimeout(timer);
}
return result;
}
export async function readOaTraceStatsForTasks(taskIds: string[]): Promise<Map<string, OaTraceStats>> {
const uniqueTaskIds = Array.from(new Set(taskIds.map((id) => String(id || "").trim()).filter(Boolean)));
return readOaTraceStatsForScopeIds(uniqueTaskIds.map((id) => taskScopeId(id)));
}
export async function readOaTraceStatsForTask(taskId: string): Promise<OaTraceStats | null> {
const map = await readOaTraceStatsForTasks([taskId]);
return map.get(taskScopeId(taskId)) ?? null;
}
export async function readOaTraceStatsForTaskAttempts(taskId: string, attemptIndexes: number[]): Promise<Map<string, OaTraceStats>> {
const uniqueAttempts = Array.from(new Set(attemptIndexes.filter((index) => Number.isInteger(index) && index > 0).map((index) => Math.floor(index))));
return readOaTraceStatsForScopeIds([taskScopeId(taskId), ...uniqueAttempts.map((index) => taskAttemptScopeId(taskId, index))]);
}
function statNumber(stats: OaTraceStats | null | undefined, key: string): number | null {
const value = Number(stats?.[key]);
return Number.isFinite(value) && value >= 0 ? Math.floor(value) : null;
}
export function applyOaTraceStatsToTaskJson(value: JsonValue, stats: OaTraceStats | null | undefined): JsonValue {
if (typeof value !== "object" || value === null || Array.isArray(value)) return value;
if (stats === null || stats === undefined) {
return { ...(value as JsonRecord), traceStats: null, statsSource: "unavailable", stepCount: null, llmStepCount: null } as unknown as JsonValue;
}
const stepCount = statNumber(stats, "stepCount");
const outputMaxSeq = statNumber(stats, "outputMaxSeq");
const patch: JsonRecord = {
traceStats: stats,
statsSource: "oa-event-flow",
};
if (stepCount !== null) {
patch.stepCount = stepCount;
patch.llmStepCount = stepCount;
}
if (outputMaxSeq !== null) patch.outputMaxSeq = outputMaxSeq;
return { ...(value as JsonRecord), ...patch };
}
export function taskScopeId(taskId: string): string {
return `task:${taskId}`;
}
export function taskAttemptScopeId(taskId: string, attemptIndex: number): string {
return `${taskScopeId(taskId)}:attempt:${Math.floor(attemptIndex)}`;
}
@@ -477,10 +477,10 @@ if command -v apt-get >/dev/null 2>&1; then
fi
fi
if ! command -v codex >/dev/null 2>&1; then
npm install -g @openai/codex@0.128.0
npm install -g --no-audit --no-fund --prefer-offline @openai/codex@0.128.0
fi
if ! command -v opencode >/dev/null 2>&1; then
npm install -g ${opencodeNpmPackage}
npm install -g --no-audit --no-fund --prefer-offline ${opencodeNpmPackage}
fi
mkdir -p "$WORKDIR" "$CODEX_HOME_DIR" "$OPENCODE_XDG_DIR/data" "$OPENCODE_XDG_DIR/config" "$OPENCODE_XDG_DIR/cache" "$OPENCODE_XDG_DIR/state"
echo "code_agent_runtime_ready codex=$(command -v codex) opencode=$(command -v opencode) cwd=$WORKDIR home=$CODEX_HOME_DIR opencodeXdg=$OPENCODE_XDG_DIR"
@@ -5,7 +5,8 @@ import { codeAgentPortForModel, codeAgentPortInfo, codeModelPorts as codeModelPo
import { claudeQqNotificationOutboxStats, notificationTargetConfigured, notificationTargetLabel } from "./notifications";
import { executionProviderOptions } from "./provider-runtime";
import { taskFullOutput } from "./task-output";
import { buildCompactTaskTranscript, buildTaskTranscript, cachedPreviewTranscript, fullTranscript, prefixPreview, safePreview, statsDaysFromUrl, taskForCompactMetaResponse, taskForMetaResponse, taskListStepCount, taskStatisticsSummary, taskTiming, timestampMs } from "./task-view";
import { applyOaTraceStatsToTaskJson, taskScopeId, type OaTraceStats } from "./oa-events";
import { buildCompactTaskTranscript, buildTaskTranscript, cachedPreviewTranscript, fullTranscript, prefixPreview, safePreview, statsDaysFromUrl, taskForCompactMetaResponse, taskForMetaResponse, taskStatisticsSummary, taskTiming, timestampMs } from "./task-view";
import { userPromptForDisplay } from "./prompts";
import type { ActiveRun, ActiveRunSlotWaiter } from "./code-agent/common";
import type { JsonValue, QueueRecord, QueuedStatusReason, QueueTask, RuntimeConfig, TaskStatus, TranscriptLine } from "./types";
@@ -43,6 +44,7 @@ export interface QueueApiContext {
safeQueueName: (value: unknown, queueId: string) => string;
sql: postgres.Sql;
taskQueueEnteredAt: (task: QueueTask) => string;
traceStatsForTasks: (taskIds: string[]) => Promise<Map<string, OaTraceStats>>;
tasks: () => QueueTask[];
truthyParam: (url: URL, name: string) => boolean;
}
@@ -140,10 +142,23 @@ function outputChunkResponse(task: QueueTask, url: URL): Response {
});
}
function taskIdsForStats(tasks: QueueTask[], selectedTask?: QueueTask | null): string[] {
const ids = tasks.map((task) => task.id);
if (selectedTask !== null && selectedTask !== undefined) ids.push(selectedTask.id);
return Array.from(new Set(ids.filter(Boolean)));
}
function statsForTask(stats: Map<string, OaTraceStats>, task: QueueTask): OaTraceStats | null {
return stats.get(taskScopeId(task.id)) ?? null;
}
function applyStats(value: JsonValue, stats: Map<string, OaTraceStats>, task: QueueTask): JsonValue {
return applyOaTraceStatsToTaskJson(value, statsForTask(stats, task));
}
function taskForListResponse(task: QueueTask, lite = false, queueTasks?: QueueTask[]): JsonValue {
const timing = taskTiming(task);
const displayPrompt = task.basePrompt || userPromptForDisplay(task.prompt);
const stepCount = taskListStepCount(task);
if (lite) {
return {
id: task.id,
@@ -157,8 +172,10 @@ function taskForListResponse(task: QueueTask, lite = false, queueTasks?: QueueTa
displayPromptChars: displayPrompt.length,
promptEditable: ctx().queuedTaskPromptEditable(task),
finalResponseChars: task.finalResponse.length,
stepCount,
llmStepCount: stepCount,
stepCount: null,
llmStepCount: null,
traceStats: null,
statsSource: "unavailable",
summaryOnly: true,
referenceTaskIds: task.referenceTaskIds,
referenceInjectionSummary: task.referenceInjection === null ? null : {
@@ -215,8 +232,10 @@ function taskForListResponse(task: QueueTask, lite = false, queueTasks?: QueueTa
displayPromptChars: displayPrompt.length,
promptEditable: ctx().queuedTaskPromptEditable(task),
finalResponseChars: task.finalResponse.length,
stepCount,
llmStepCount: stepCount,
stepCount: null,
llmStepCount: null,
traceStats: null,
statsSource: "unavailable",
summaryOnly: true,
referenceTaskIds: task.referenceTaskIds,
referenceInjection: task.referenceInjection,
@@ -1035,11 +1054,16 @@ async function databaseTasksOverviewResponse(url: URL): Promise<Response | null>
} as unknown as JsonValue;
}
const queueContextTasks = [...ctx().tasks(), ...rowsSource];
const traceStats = await ctx().traceStatsForTasks(taskIdsForStats(rowsSource, selectedTask));
if (selectedTask !== null && selected !== null && typeof selected === "object" && !Array.isArray(selected)) {
const selectedRecord = selected as Record<string, JsonValue>;
selectedRecord.task = applyStats(selectedRecord.task, traceStats, selectedTask);
}
return ctx().compactJsonResponse({
ok: true,
queue,
statistics,
tasks: rowsSource.map((task) => taskForListResponse(task, true, queueContextTasks)),
tasks: rowsSource.map((task) => applyStats(taskForListResponse(task, true, queueContextTasks), traceStats, task)),
selected,
pagination: {
limit,
@@ -1098,11 +1122,16 @@ async function tasksOverviewResponse(url: URL): Promise<Response> {
maxSeq,
} as unknown as JsonValue;
}
const traceStats = await ctx().traceStatsForTasks(taskIdsForStats(rowsSource, selectedTask));
if (selectedTask !== null && selected !== null && typeof selected === "object" && !Array.isArray(selected)) {
const selectedRecord = selected as Record<string, JsonValue>;
selectedRecord.task = applyStats(selectedRecord.task, traceStats, selectedTask);
}
return ctx().compactJsonResponse({
ok: true,
queue,
statistics: taskStatisticsSummary(filteredTasks, statsDaysFromUrl(url)),
tasks: rowsSource.map((task) => taskForListResponse(task, true, allTasks)),
tasks: rowsSource.map((task) => applyStats(taskForListResponse(task, true, allTasks), traceStats, task)),
selected,
pagination: {
limit,
@@ -1178,14 +1178,15 @@ function attemptForResponse(attempt: AttemptSummary, full = false): JsonValue {
function taskForResponse(task: QueueTask, full = false, includeRaw = full): JsonValue {
const displayPrompt = task.basePrompt || userPromptForDisplay(task.prompt);
const stepCount = taskLlmStepCount(task);
return {
...task,
agentPort: codeAgentPortForModel(task.model),
agentPortInfo: codeAgentPortInfo(codeAgentPortForModel(task.model)),
judgeFailRetryLimit,
stepCount,
llmStepCount: stepCount,
stepCount: null,
llmStepCount: null,
traceStats: null,
statsSource: "unavailable",
prompt: full ? task.prompt : safePreview(task.prompt, 2000),
basePrompt: full ? task.basePrompt : safePreview(task.basePrompt, 2000),
displayPrompt: full ? displayPrompt : safePreview(displayPrompt, 2000),
@@ -1226,8 +1227,8 @@ function taskStoredStepCount(task: QueueTask): number | null {
return nonNegativeInteger(task.stepCount) ?? nonNegativeInteger(task.llmStepCount);
}
// Used by overview/list responses and SSE patches. It must stay archive-free;
// trace/detail endpoints compute exact counts from the full transcript on demand.
// Used only as a producer-side seed for OA stats snapshots; frontend/API
// display authority is the oa_trace_stats projection applied at route edges.
function taskListStepCount(task: QueueTask): number {
return taskStoredStepCount(task) ?? retainedTaskStepCount(task);
}
@@ -1245,14 +1246,9 @@ function taskOutputMaxSeq(task: QueueTask): number {
return values.length > 0 ? Math.max(...values) : 0;
}
function taskLlmStepCount(task: QueueTask): number {
return taskStoredStepCount(task) ?? taskStepCountFromTranscript(task, cachedPreviewTranscript(task));
}
function taskForMetaResponse(task: QueueTask): JsonValue {
const lastOutputSeq = taskOutputMaxSeq(task);
const displayPrompt = task.basePrompt || userPromptForDisplay(task.prompt);
const stepCount = taskListStepCount(task);
return {
id: task.id,
queueId: ctx().queueIdOf(task),
@@ -1265,8 +1261,10 @@ function taskForMetaResponse(task: QueueTask): JsonValue {
displayPromptChars: displayPrompt.length,
promptEditable: ctx().queuedTaskPromptEditable(task),
finalResponseChars: task.finalResponse.length,
stepCount,
llmStepCount: stepCount,
stepCount: null,
llmStepCount: null,
traceStats: null,
statsSource: "unavailable",
summaryOnly: false,
referenceTaskIds: task.referenceTaskIds,
referenceInjection: task.referenceInjection,
@@ -1313,7 +1311,6 @@ function taskForMetaResponse(task: QueueTask): JsonValue {
function taskForCompactMetaResponse(task: QueueTask): JsonValue {
const displayPrompt = task.basePrompt || userPromptForDisplay(task.prompt);
const lastOutputSeq = taskOutputMaxSeq(task);
const stepCount = taskListStepCount(task);
return {
id: task.id,
queueId: ctx().queueIdOf(task),
@@ -1326,8 +1323,10 @@ function taskForCompactMetaResponse(task: QueueTask): JsonValue {
displayPromptChars: displayPrompt.length,
promptEditable: ctx().queuedTaskPromptEditable(task),
finalResponseChars: task.finalResponse.length,
stepCount,
llmStepCount: stepCount,
stepCount: null,
llmStepCount: null,
traceStats: null,
statsSource: "unavailable",
summaryOnly: true,
referenceTaskIds: task.referenceTaskIds,
referenceInjection: task.referenceInjection === null ? null : {
@@ -1517,20 +1516,10 @@ function toolStepCountsFromTranscript(lines: TranscriptLine[]): { toolCallCount:
};
}
function llmStepCountFromTranscript(lines: TranscriptLine[], _fallback = 0): number {
return toolStepCountsFromTranscript(lines).toolCallCount;
}
function realAttemptWindows(task: QueueTask, transcript: TranscriptLine[]): TraceAttemptWindow[] {
return traceAttemptWindows(task, transcript).filter((window) => window.synthetic !== true && window.index > 0);
}
function taskStepCountFromTranscript(task: QueueTask, transcript: TranscriptLine[]): number {
const windows = realAttemptWindows(task, transcript);
if (windows.length === 0) return llmStepCountFromTranscript(transcript);
return windows.reduce((sum, window) => sum + llmStepCountFromTranscript(executionLinesForAttempt(window.lines)), 0);
}
function taskExecutionTranscript(task: QueueTask, transcript: TranscriptLine[]): TranscriptLine[] {
const windows = realAttemptWindows(task, transcript);
if (windows.length === 0) return transcript;
@@ -1543,7 +1532,6 @@ function executionSummaryFromTranscript(
timing: Record<string, JsonValue>,
outputCount = taskFullOutput(task).length,
retainedOutputCount = task.output.length,
fallbackStepCount = 0,
): JsonValue {
const toolLines = transcript.filter(isToolActionLine);
const editedFiles = uniqueStrings(toolLines.flatMap((line) => line.kind === "edited" ? parseEditedFilesFromText(`${line.title}\n${line.bodyPreview ?? ""}`) : []), 40);
@@ -1571,7 +1559,7 @@ function executionSummaryFromTranscript(
}
function taskExecutionSummary(task: QueueTask, transcript = cachedPreviewTranscript(task)): JsonValue {
return executionSummaryFromTranscript(task, taskExecutionTranscript(task, transcript), taskTiming(task) as Record<string, JsonValue>, undefined, undefined, task.currentAttempt || task.attempts.length);
return executionSummaryFromTranscript(task, taskExecutionTranscript(task, transcript), taskTiming(task) as Record<string, JsonValue>);
}
function parseAttemptIndex(text: string): number | null {
@@ -1669,6 +1657,8 @@ interface TraceAttemptWindow {
label?: string;
}
type TraceStatsLookup = Map<string, JsonValue> | Record<string, JsonValue> | null | undefined;
function transcriptLineSeq(line: TranscriptLine | undefined): number | null {
const value = Number(line?.seq ?? NaN);
return Number.isFinite(value) ? value : null;
@@ -1806,7 +1796,7 @@ function executionLinesForAttempt(lines: TranscriptLine[]): TranscriptLine[] {
return lines.filter((line) => line.title !== "Submitted prompt" && line.title !== "Attempt started" && line.title !== "Judge result");
}
function taskTraceAttemptSummaries(task: QueueTask, transcript: TranscriptLine[]): JsonValue[] {
function taskTraceAttemptSummaries(task: QueueTask, transcript: TranscriptLine[], oaTraceStatsByScope: TraceStatsLookup = null): JsonValue[] {
const windows = traceAttemptWindows(task, transcript);
return windows.map((window) => {
const attempt = window.attempt;
@@ -1817,7 +1807,12 @@ function taskTraceAttemptSummaries(task: QueueTask, transcript: TranscriptLine[]
const finalResponse = synthetic ? "" : String(attempt?.finalResponse ?? attempt?.finalResponsePreview ?? (window.index === task.attempts.length ? task.finalResponse : ""));
const finalResponseChars = Number(attempt?.finalResponseChars ?? finalResponse.length);
const executionLines = executionLinesForAttempt(window.lines);
const errorCount = executionLines.filter((line) => line.kind === "error").length;
const attemptStats = synthetic || window.index <= 0 ? null : traceStatsForScope(oaTraceStatsByScope, taskAttemptScopeId(task.id, window.index));
const stepCount = traceStatsNumber(attemptStats, "stepCount");
const readCount = traceStatsNumber(attemptStats, "readCount");
const editCount = traceStatsNumber(attemptStats, "editCount");
const runCount = traceStatsNumber(attemptStats, "runCount");
const errorCount = traceStatsNumber(attemptStats, "errorCount");
const feedbackPrompt = synthetic ? null : attemptFeedbackPromptRecord(task, window.index, attempt, judge);
const inputPrompt = promptSnapshot(String(attempt?.inputPrompt ?? ""), 1200);
return {
@@ -1854,14 +1849,23 @@ function taskTraceAttemptSummaries(task: QueueTask, transcript: TranscriptLine[]
feedbackPromptSource: feedbackPrompt?.source ?? null,
feedbackPromptForAttempt: feedbackPrompt?.forAttempt ?? null,
feedbackPromptTruncated: feedbackPrompt?.truncated ?? false,
attemptScopeId: synthetic || window.index <= 0 ? null : taskAttemptScopeId(task.id, window.index),
stepCount,
readCount,
editCount,
runCount,
errorCount,
execution: executionSummaryFromTranscript(
task,
executionLines,
attemptTimingSummary(attempt, executionLines.length > 0 ? executionLines : window.lines),
window.lines.length,
window.lines.length,
synthetic || window.index <= 0 ? 0 : 1,
statsSource: attemptStats === null ? "unavailable" : "oa-event-flow",
traceStats: attemptStats,
execution: executionSummaryWithOaStats(
executionSummaryFromTranscript(
task,
executionLines,
attemptTimingSummary(attempt, executionLines.length > 0 ? executionLines : window.lines),
window.lines.length,
window.lines.length,
),
attemptStats,
),
};
}) as unknown as JsonValue[];
@@ -1905,15 +1909,82 @@ function taskTracePromptSummary(task: QueueTask): JsonValue {
} as unknown as JsonValue;
}
function taskTraceSummaryResponse(task: QueueTask): JsonValue {
function traceStatsRecord(value: JsonValue | null | undefined): Record<string, JsonValue> | null {
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, JsonValue> : null;
}
function traceStatsNumber(stats: Record<string, JsonValue> | null, key: string): number | null {
const value = Number(stats?.[key]);
return Number.isFinite(value) && value >= 0 ? Math.floor(value) : null;
}
function taskAttemptScopeId(taskId: string, attemptIndex: number): string {
return `task:${taskId}:attempt:${Math.floor(attemptIndex)}`;
}
function traceStatsForScope(lookup: TraceStatsLookup, scopeId: string): Record<string, JsonValue> | null {
if (lookup === null || lookup === undefined || scopeId.length === 0) return null;
const value = lookup instanceof Map ? lookup.get(scopeId) : lookup[scopeId];
return traceStatsRecord(value);
}
function executionSummaryWithOaStats(execution: JsonValue, stats: Record<string, JsonValue> | null): JsonValue {
if (typeof execution !== "object" || execution === null || Array.isArray(execution)) return execution;
const record = execution as Record<string, JsonValue>;
if (stats === null) {
const {
stepCount: _stepCount,
llmStepCount: _llmStepCount,
toolCallCount: _toolCallCount,
readCount: _readCount,
editCount: _editCount,
runCount: _runCount,
errorCount: _errorCount,
traceLineCount: _traceLineCount,
outputMaxSeq: _outputMaxSeq,
transcriptMaxSeq: _transcriptMaxSeq,
...rest
} = record;
return {
...rest,
statsSource: "unavailable",
traceStats: null,
statsUnavailable: true,
} as unknown as JsonValue;
}
const stepCount = traceStatsNumber(stats, "stepCount");
const llmStepCount = traceStatsNumber(stats, "llmStepCount") ?? stepCount;
const readCount = traceStatsNumber(stats, "readCount");
const editCount = traceStatsNumber(stats, "editCount");
const runCount = traceStatsNumber(stats, "runCount");
const errorCount = traceStatsNumber(stats, "errorCount");
const traceLineCount = traceStatsNumber(stats, "traceLineCount");
const outputMaxSeq = traceStatsNumber(stats, "outputMaxSeq");
const toolCallCount = readCount === null && editCount === null && runCount === null ? null : (readCount ?? 0) + (editCount ?? 0) + (runCount ?? 0);
return {
...record,
...(stepCount === null ? {} : { stepCount }),
...(llmStepCount === null ? {} : { llmStepCount }),
...(toolCallCount === null ? {} : { toolCallCount }),
...(readCount === null ? {} : { readCount }),
...(editCount === null ? {} : { editCount }),
...(runCount === null ? {} : { runCount }),
...(errorCount === null ? {} : { errorCount }),
...(traceLineCount === null ? {} : { traceLineCount }),
...(outputMaxSeq === null ? {} : { outputMaxSeq, transcriptMaxSeq: outputMaxSeq }),
statsSource: "oa-event-flow",
traceStats: stats,
} as unknown as JsonValue;
}
function taskTraceSummaryResponse(task: QueueTask, oaTraceStats: JsonValue | null = null, oaTraceStatsByScope: TraceStatsLookup = null): JsonValue {
const transcript = cachedPreviewTranscript(task);
const attempts = taskTraceAttemptSummaries(task, transcript);
const stepCount = taskStepCountFromTranscript(task, transcript);
const errorCount = attempts.reduce((sum: number, attempt: JsonValue) => {
if (typeof attempt !== "object" || attempt === null || Array.isArray(attempt)) return sum;
const value = Number((attempt as { errorCount?: unknown }).errorCount || 0);
return sum + (Number.isFinite(value) && value > 0 ? value : 0);
}, 0);
const attempts = taskTraceAttemptSummaries(task, transcript, oaTraceStatsByScope);
const stats = traceStatsRecord(oaTraceStats);
const stepCount = traceStatsNumber(stats, "stepCount");
const llmStepCount = traceStatsNumber(stats, "llmStepCount") ?? stepCount;
const errorCount = traceStatsNumber(stats, "errorCount");
const execution = executionSummaryWithOaStats(taskExecutionSummary(task, transcript), stats);
return {
id: task.id,
queueId: ctx().queueIdOf(task),
@@ -1931,10 +2002,12 @@ function taskTraceSummaryResponse(task: QueueTask): JsonValue {
currentAttempt: task.currentAttempt,
maxAttempts: task.maxAttempts,
stepCount,
llmStepCount: stepCount,
llmStepCount,
promptEditable: ctx().queuedTaskPromptEditable(task),
prompt: taskTracePromptSummary(task),
execution: taskExecutionSummary(task, transcript),
execution,
traceStats: stats,
statsSource: stats === null ? "unavailable" : "oa-event-flow",
finalResponse: task.finalResponse,
finalResponseChars: task.finalResponse.length,
lastJudge: task.lastJudge,
@@ -2132,7 +2205,6 @@ export {
statsDaysFromUrl,
taskExecutionSummary,
taskListStepCount,
taskLlmStepCount,
taskOutputMaxSeq,
taskForCompactMetaResponse,
taskForMetaResponse,
@@ -63,6 +63,7 @@ export interface RuntimeConfig {
turnNoActivityTimeoutMs: number;
databaseUrl: string;
databaseFlushIntervalMs: number;
oaEventFlowBaseUrl: string;
notifyClaudeQqEnabled: boolean;
notifyClaudeQqBaseUrl: string;
notifyClaudeQqTargetType: "private" | "group";
@@ -0,0 +1,11 @@
FROM oven/bun:1-alpine
WORKDIR /app/src/components/microservices/oa-event-flow
COPY src/components/microservices/oa-event-flow/package.json ./package.json
RUN bun install --production
COPY src/components/microservices/oa-event-flow/tsconfig.json ./tsconfig.json
COPY src/components/shared /app/src/components/shared
COPY src/components/microservices/oa-event-flow/src ./src
EXPOSE 4255
CMD ["bun", "run", "src/index.ts"]
@@ -0,0 +1,15 @@
{
"lockfileVersion": 1,
"configVersion": 1,
"workspaces": {
"": {
"name": "@unidesk/oa-event-flow",
"dependencies": {
"postgres": "latest",
},
},
},
"packages": {
"postgres": ["postgres@3.4.9", "", {}, "sha512-GD3qdB0x1z9xgFI6cdRD6xu2Sp2WCOEoe3mtnyB5Ee0XrrL5Pe+e4CCnJrRMnL1zYtRDZmQQVbvOttLnKDLnaw=="],
}
}
@@ -0,0 +1,12 @@
{
"name": "@unidesk/oa-event-flow",
"private": true,
"type": "module",
"scripts": {
"start": "bun run src/index.ts",
"check": "tsc -p tsconfig.json --noEmit"
},
"dependencies": {
"postgres": "latest"
}
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,18 @@
{
"compilerOptions": {
"composite": true,
"target": "ES2022",
"module": "ESNext",
"moduleResolution": "Bundler",
"types": ["bun", "node"],
"strict": true,
"noImplicitReturns": true,
"noFallthroughCasesInSwitch": true,
"declaration": true,
"emitDeclarationOnly": true,
"outDir": "dist",
"skipLibCheck": true
},
"include": ["src/**/*.ts"],
"references": [{ "path": "../../shared" }]
}