From cc82ac5c1a618bea4a5fb7e7110d3e285b769a3c Mon Sep 17 00:00:00 2001 From: Bo Wu Date: Sat, 5 Sep 2026 19:19:02 -0700 Subject: [PATCH] Fix result and trace publication --- control-server/src/server.mjs | 24 ++++++++++++-- control-server/src/session-runtime.mjs | 32 ++++++++++++++++++ control-server/test/session-runtime.test.mjs | 35 ++++++++++++++++++++ 3 files changed, 89 insertions(+), 2 deletions(-) diff --git a/control-server/src/server.mjs b/control-server/src/server.mjs index 51bcbbf..63b76d1 100644 --- a/control-server/src/server.mjs +++ b/control-server/src/server.mjs @@ -26,6 +26,7 @@ import { import { acceptsLiveInput, automaticResumeLimit, + boundedFinalMessage, completionExitDelayMs, controlMode, executionTerminalOutcome, @@ -40,6 +41,7 @@ import { shouldAutomaticallyResume, submitLocalFollowup, validResourceId, + waitForTraceFinalization, } from "./session-runtime.mjs"; const here = path.dirname(fileURLToPath(import.meta.url)); @@ -334,13 +336,29 @@ function traceReferences(id) { return references; } +function latestAgentTracesReady(id) { + const agents = path.join(traceRoot(id), "agents"); + try { + return fs.readdirSync(agents, { withFileTypes: true }) + .filter((item) => item.isDirectory()) + .every((entry) => { + const base = path.join(agents, entry.name); + let attempt = ""; + try { attempt = fs.readFileSync(path.join(base, "latest"), "utf8").trim(); } catch {} + return Boolean(attempt) && fs.existsSync(path.join(base, attempt, "events.jsonl")); + }); + } catch { + return true; + } +} + function writeTraceSummary(id, status) { const root = traceRoot(id); fs.mkdirSync(root, { recursive: true }); let result = ""; let fallback = ""; - try { result = conciseTail(fs.readFileSync(path.join(sessionStateDir(id), "orchestrator-result.md"), "utf8"), 80, 6000); } catch {} - try { fallback = conciseTail(fs.readFileSync(path.join(sessionStateDir(id), "orchestrator-last-message.txt"), "utf8"), 40, 6000); } catch {} + try { result = boundedFinalMessage(fs.readFileSync(path.join(sessionStateDir(id), "orchestrator-result.md"), "utf8"), 6000); } catch {} + try { fallback = boundedFinalMessage(fs.readFileSync(path.join(sessionStateDir(id), "orchestrator-last-message.txt"), "utf8"), 6000); } catch {} const finalMessage = selectFinalMessage(result, fallback); const completionRoute = workflowCompletionRoute(id); const terminalOutcome = executionTerminalOutcome({ @@ -893,6 +911,8 @@ async function retireSession(id, status, actor, terminalOutcome = status === "fa if (uidSandbox) runSessionControl(id, "stop"); else runTmux(id, ["kill-session", "-t", id]); } + await waitForTraceFinalization({ ready: () => latestAgentTracesReady(id) }); + checkpoint(id); const now = new Date().toISOString(); record.status = status; record.terminalOutcome = terminalOutcome; diff --git a/control-server/src/session-runtime.mjs b/control-server/src/session-runtime.mjs index cbd5e57..1fb09a8 100644 --- a/control-server/src/session-runtime.mjs +++ b/control-server/src/session-runtime.mjs @@ -76,6 +76,38 @@ export function selectFinalMessage(result, fallback) { return String(result || "").trim() || String(fallback || "").trim(); } +export function boundedFinalMessage(value, maximumBytes = 6000) { + const text = String(value || "").trim(); + if (Buffer.byteLength(text, "utf8") <= maximumBytes) return text; + const suffix = "…"; + const budget = Math.max(0, maximumBytes - Buffer.byteLength(suffix, "utf8")); + let result = ""; + let bytes = 0; + for (const character of text) { + const size = Buffer.byteLength(character, "utf8"); + if (bytes + size > budget) break; + result += character; + bytes += size; + } + return `${result.trimEnd()}${suffix}`; +} + +export async function waitForTraceFinalization({ + ready, + timeoutMs = 2000, + pollIntervalMs = 50, + now = Date.now, + sleep = (milliseconds) => new Promise((resolve) => setTimeout(resolve, milliseconds)), +}) { + const deadline = now() + timeoutMs; + while (!ready()) { + const remaining = deadline - now(); + if (remaining <= 0) return false; + await sleep(Math.min(pollIntervalMs, remaining)); + } + return true; +} + export function responseTypeForMessage(message, completionRoute = "") { if (!["direct-response", "observe", "request-review", "human-review"].includes(completionRoute)) return "assistant_message"; const text = String(message || "").trim(); diff --git a/control-server/test/session-runtime.test.mjs b/control-server/test/session-runtime.test.mjs index 50d7d33..b2a059a 100644 --- a/control-server/test/session-runtime.test.mjs +++ b/control-server/test/session-runtime.test.mjs @@ -4,6 +4,7 @@ import test from "node:test"; import { acceptsLiveInput, automaticResumeLimit, + boundedFinalMessage, completionExitDelayMs, controlMode, executionTerminalOutcome, @@ -19,6 +20,7 @@ import { shouldAutomaticallyResume, submitLocalFollowup, validResourceId, + waitForTraceFinalization, workerReportInterruptedEvent, workerReportPublicEvent, } from "../src/session-runtime.mjs"; @@ -125,6 +127,39 @@ test("completed session reports prefer the explicit bounded caller result", () = assert.equal(normalizeWorkerReport({ report: "bad", completionRoute: "human-review", terminalOutcome: "review_requested", message: "not a question" }), null); }); +test("bounded final messages preserve complete multi-line answers below the byte limit", () => { + const message = Array.from({ length: 85 }, (_, index) => `line ${index + 1}`).join("\n"); + assert.equal(Buffer.byteLength(message, "utf8") < 6000, true); + assert.equal(boundedFinalMessage(`\n${message}\n`, 6000), message); + const unicode = boundedFinalMessage("界".repeat(3000), 6000); + assert.equal(Buffer.byteLength(unicode, "utf8") <= 6000, true); + assert.equal(unicode.endsWith("…"), true); +}); + +test("trace finalization wait is bounded and observes late normalized events", async () => { + let clock = 0; + let checks = 0; + const ready = await waitForTraceFinalization({ + ready: () => ++checks >= 3, + timeoutMs: 100, + pollIntervalMs: 10, + now: () => clock, + sleep: async (milliseconds) => { clock += milliseconds; }, + }); + assert.equal(ready, true); + assert.equal(checks, 3); + + clock = 0; + assert.equal(await waitForTraceFinalization({ + ready: () => false, + timeoutMs: 25, + pollIntervalMs: 10, + now: () => clock, + sleep: async (milliseconds) => { clock += milliseconds; }, + }), false); + assert.equal(clock, 25); +}); + test("repair reports preserve only explicit source and reviewed-operation effects", () => { assert.deepEqual(normalizeWorkerReport({ report: "A reviewed restart is required.",