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

Merge pull request #1516 from pikasTech/fix/1515-agentrun-projection-sync

Browse Source 
fix: D601 v0.3 AgentRun turn 后台投影
This commit is contained in:
Lyon
2026-06-18 21:32:23 +08:00
committed by GitHub
parent 95931fecbf b469b3efa6
commit 522605ecdb
3 changed files with 105 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/code-agent-agentrun-adapter.ts
+1
View File
@@ -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,
internal/cloud/server-agent-chat.test.ts
+15 -22
View File
@@ -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<void>((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);
internal/cloud/server-code-agent-http.ts
+89
View File
@@ -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({