diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index 52095d89..966bb702 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -225,6 +225,7 @@ export function initialAgentRunChatResult({ params = {}, options = {}, traceId } status: "running", shortConnection: true, traceId, + projectId: firstNonEmpty(params.projectId) || null, conversationId: safeConversationId(params.conversationId) || null, sessionId: safeSessionId(params.sessionId) || agentRunSessionId(traceId), threadId: safeOpaqueId(params.threadId) || null, diff --git a/internal/cloud/server-agent-chat.test.ts b/internal/cloud/server-agent-chat.test.ts index 2960ab55..552d3ba8 100644 --- a/internal/cloud/server-agent-chat.test.ts +++ b/internal/cloud/server-agent-chat.test.ts @@ -565,9 +565,6 @@ test("cloud api /v1/agent/chat delegates v0.2 turns to AgentRun v0.1 over adapte const agentRunPort = agentRunServer.address().port; const traceStore = createCodeAgentTraceStore(); - let deferOwnerRecord = false; - let releaseOwnerRecord: (() => void) | null = null; - let deferredOwnerRecordStarted = false; const server = createCloudApiServer({ traceStore, env: { @@ -588,10 +585,6 @@ test("cloud api /v1/agent/chat delegates v0.2 turns to AgentRun v0.1 over adapte return { ok: true, actor: TEST_AGENT_ACTOR, session: TEST_AUTH_SESSION }; }, async recordAgentSessionOwner(input) { - if (deferOwnerRecord && input.traceId === "trc_server-test-agentrun-adapter" && input.status === "running") { - deferredOwnerRecordStarted = true; - await new Promise((resolve) => { releaseOwnerRecord = resolve; }); - } const existing = ownerSessions.get(input.sessionId ?? input.id); const record = testAgentSessionRecord({ ...existing, @@ -608,7 +601,7 @@ test("cloud api /v1/agent/chat delegates v0.2 turns to AgentRun v0.1 over adapte assert.equal(actor.id, "usr_agent_owner"); assert.equal(actor.role, "user"); assert.equal(conversationId, "cnv_server-test-agentrun"); - assert.equal(projectId, "pikasTech/HWLAB"); + assert.equal(projectId, "prj_hwpod_workbench"); return { conversationId, messages: conversationMessages }; } } @@ -621,7 +614,7 @@ test("cloud api /v1/agent/chat delegates v0.2 turns to AgentRun v0.1 over adapte ownerSessions.set("ses_server-test-agentrun", testAgentSessionRecord({ sessionId: "ses_server-test-agentrun", conversationId: "cnv_server-test-agentrun", - projectId: "pikasTech/HWLAB", + projectId: "prj_hwpod_workbench", status: "idle", threadId: "019e8078-db67-7750-a5d9-1a99f3abd445" })); @@ -646,6 +639,12 @@ test("cloud api /v1/agent/chat delegates v0.2 turns to AgentRun v0.1 over adapte assert.equal(accepted.userMessageId, "msg_server-test-agentrun-adapter_user"); assert.equal(accepted.assistantMessageId, "msg_server-test-agentrun-adapter_agent"); + for (let index = 0; index < 50 && !traceStore.snapshot(traceId).events.some((event) => event.label === "agentrun:result:completed"); index += 1) await delay(10); + const projectedTrace = traceStore.snapshot(traceId); + assert.equal(projectedTrace.status, "completed"); + assert.ok(projectedTrace.events.some((event) => event.label === "agentrun:assistant:message")); + assert.ok(projectedTrace.events.some((event) => event.label === "agentrun:result:completed")); + const acceptedTurn = await fetch(`http://127.0.0.1:${port}/v1/agent/turns/${encodeURIComponent(traceId)}`, { headers: { cookie: "hwlab_session=test-stub-session" } }); assert.equal(acceptedTurn.status, 200); const acceptedTurnPayload = await acceptedTurn.json(); @@ -759,7 +758,6 @@ test("cloud api /v1/agent/chat delegates v0.2 turns to AgentRun v0.1 over adapte } ); - deferOwnerRecord = true; const steerRequest = fetch(`http://127.0.0.1:${port}/v1/agent/chat/steer`, { method: "POST", headers: { "content-type": "application/json", "x-trace-id": traceId, cookie: "hwlab_session=test-stub-session" }, @@ -777,21 +775,15 @@ test("cloud api /v1/agent/chat delegates v0.2 turns to AgentRun v0.1 over adapte steerRequest, delay(100).then(() => null) ]); - assert.ok(steer, "steer short response must not wait for owner persistence"); - assert.equal(steer.status, 202); - for (let index = 0; index < 20 && !deferredOwnerRecordStarted; index += 1) await delay(10); - assert.equal(deferredOwnerRecordStarted, true); - releaseOwnerRecord?.(); - deferOwnerRecord = false; + assert.ok(steer, "terminal steer rejection must return as a short response"); + assert.equal(steer.status, 409); const steerBody = await steer.json(); - assert.equal(steerBody.accepted, true); + assert.equal(steerBody.accepted, false); + assert.equal(steerBody.error.code, "steer_trace_terminal"); assert.equal(steerBody.route, "/v1/agent/chat/steer"); assert.equal(steerBody.traceId, traceId); - assert.equal(steerBody.steerTraceId, "trc_steer_server_test"); assert.equal(steerBody.agentRun.runId, "run_hwlab_adapter"); - assert.equal(steerBody.agentRun.targetCommandId, "cmd_hwlab_adapter"); - assert.equal(steerBody.agentRun.steerCommandId, "cmd_hwlab_adapter_steer"); - assert.ok(steerBody.runnerTrace.events.some((event) => event.label === "agentrun:steer:command-created")); + assert.equal(calls.some((call) => call.method === "POST" && call.path === "/api/v1/runs/run_hwlab_adapter/commands" && call.body?.type === "steer"), false); const secondTraceId = "trc_server-test-agentrun-adapter-second"; const second = await fetch(`http://127.0.0.1:${port}/v1/agent/chat`, { @@ -830,7 +822,7 @@ test("cloud api /v1/agent/chat delegates v0.2 turns to AgentRun v0.1 over adapte assert.equal(secondPayload.runnerTrace.events.some((event) => String(event.text ?? event.message ?? "").includes("旧 command 尾部")), false); assert.equal(calls.filter((call) => call.method === "POST" && call.path === "/api/v1/runs").length, 1); assert.equal(calls.filter((call) => call.method === "POST" && call.path === "/api/v1/runs/run_hwlab_adapter/runner-jobs").length, 2); - assert.equal(calls.filter((call) => call.method === "POST" && call.path === "/api/v1/runs/run_hwlab_adapter/commands").length, 3); + assert.equal(calls.filter((call) => call.method === "POST" && call.path === "/api/v1/runs/run_hwlab_adapter/commands").length, 2); } finally { await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve()))); await new Promise((resolve, reject) => agentRunServer.close((error) => (error ? reject(error) : resolve()))); @@ -2985,6 +2977,7 @@ test("cloud api repairs historical same-session AgentRun trace after lastTraceId assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands`)); assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${firstCommandId}/result`)); assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${secondCommandId}/result`), false); + for (let index = 0; index < 20 && !ownerSessions.get("ses_issue955_historical")?.session?.traceResults?.[firstTraceId]; index += 1) await delay(10); assert.equal(ownerSessions.get("ses_issue955_historical").session.traceResults[firstTraceId].finalResponse.text, firstFinalText); const repairedSession = ownerSessions.get("ses_issue955_historical"); assert.equal(repairedSession.lastTraceId, secondTraceId); diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index 01a7b217..a9e49ff6 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -682,6 +682,7 @@ function submitCodeAgentChatTurn({ params, options, traceId }) { const owned = annotateOwner(withCodeAgentBillingReservation(payload, params), params); await recordCodeAgentSessionOwner({ payload: owned, params, options: executionOptions, status: "running" }); results.set(traceId, owned); + scheduleAgentRunProjectionSync({ traceId, params, options: executionOptions, traceStore, currentResult: owned }); } catch (error) { if (isCodeAgentResultCanceled(results.get(traceId))) return; const payload = annotateOwner({ @@ -787,6 +788,69 @@ function submitCodeAgentChatTurn({ params, options, traceId }) { }); } +function scheduleAgentRunProjectionSync({ traceId, params = {}, options = {}, traceStore = defaultCodeAgentTraceStore, currentResult = null } = {}) { + if (!safeTraceId(traceId) || !currentResult?.agentRun?.runId || !currentResult?.agentRun?.commandId) return null; + const env = options.env ?? process.env; + const timeoutMs = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_SYNC_TIMEOUT_MS, 90_000); + const pollIntervalMs = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_SYNC_POLL_INTERVAL_MS, 1_000); + const refreshOptions = codeAgentTurnStatusRefreshOptions(options); + const startedAt = Date.now(); + + setImmediate(() => { + void (async () => { + let result = currentResult; + let lastError = null; + while (Date.now() - startedAt <= timeoutMs) { + try { + const synced = await syncAgentRunChatResult({ traceId, currentResult: result, options: refreshOptions, traceStore }); + result = synced.result ?? result; + if (result && isTraceCommandTerminalStatus(result.status)) { + scheduleCodeAgentTerminalTurnStatusEffects({ payload: result, params, options, preserveLastTraceId: true }); + return; + } + lastError = null; + } catch (error) { + lastError = error; + } + const elapsedMs = Date.now() - startedAt; + if (elapsedMs >= timeoutMs) break; + await sleepAgentRunProjectionSync(Math.min(pollIntervalMs, Math.max(0, timeoutMs - elapsedMs))); + } + traceStore.append(traceId, { + type: "turn-status", + status: "degraded", + label: lastError ? "agentrun:projection:sync-failed" : "agentrun:projection:sync-timeout", + errorCode: lastError?.code ?? null, + message: lastError?.message ?? "AgentRun projection sync did not reach terminal within the configured budget.", + runId: result?.agentRun?.runId ?? currentResult.agentRun.runId, + commandId: result?.agentRun?.commandId ?? currentResult.agentRun.commandId, + timeoutMs, + waitingFor: "agentrun-result", + terminal: false, + valuesPrinted: false + }); + })().catch((error) => { + traceStore.append(traceId, { + type: "turn-status", + status: "degraded", + label: "agentrun:projection:sync-crashed", + errorCode: error?.code ?? "agentrun_projection_sync_crashed", + message: error?.message ?? "AgentRun projection sync crashed.", + runId: currentResult.agentRun.runId, + commandId: currentResult.agentRun.commandId, + waitingFor: "agentrun-result", + terminal: false, + valuesPrinted: false + }); + }); + }); + return { scheduled: true, traceId, timeoutMs, pollIntervalMs, valuesPrinted: false }; +} + +function sleepAgentRunProjectionSync(ms) { + return new Promise((resolve) => setTimeout(resolve, Math.max(0, Number(ms) || 0))); +} + function codeAgentChatShortConnectionRequested(request, params, options = {}) { const header = firstHeaderValue(request, "prefer"); const explicit = firstHeaderValue(request, "x-hwlab-short-connection"); @@ -1900,6 +1964,31 @@ export async function handleCodeAgentSteerHttp(request, response, options) { }); return; } + if (isTraceCommandTerminalStatus(currentResult.status)) { + sendJson(response, 409, { + ok: false, + accepted: false, + status: "blocked", + traceId, + route: "/v1/agent/chat/steer", + sessionId: safeSessionId(currentResult.sessionId ?? params.sessionId) || null, + threadId: safeOpaqueId(currentResult.threadId ?? params.threadId) || null, + agentRun: currentResult.agentRun, + runnerTrace: traceStore.snapshot(traceId), + error: { + code: "steer_trace_terminal", + layer: "api", + category: "not_in_flight", + retryable: false, + message: `Trace ${traceId} is already terminal and cannot accept a steer command.`, + traceId, + route: "/v1/agent/chat/steer", + toolName: "agentrun.command.steer" + }, + valuesPrinted: false + }); + return; + } try { const payload = await steerAgentRunChatTurn({