Merge pull request #1471 from pikasTech/fix/1422-turn-status-refresh-budget
fix: avoid duplicate AgentRun turn status refresh
This commit is contained in:
@@ -440,7 +440,7 @@ function isAgentRunCommandAlreadyClaimed(error) {
|
|||||||
export async function syncAgentRunChatResult({ traceId, currentResult = null, options = {}, traceStore = defaultCodeAgentTraceStore, appendResultEvent = true, refreshEvents = true, forceResultSync = false }) {
|
export async function syncAgentRunChatResult({ traceId, currentResult = null, options = {}, traceStore = defaultCodeAgentTraceStore, appendResultEvent = true, refreshEvents = true, forceResultSync = false }) {
|
||||||
const initial = currentResult ?? await loadPersistedAgentRunResult(traceId, options);
|
const initial = currentResult ?? await loadPersistedAgentRunResult(traceId, options);
|
||||||
const mapped = initial ? await resolveAgentRunTraceCommandMapping({ traceId, mapped: initial, options }) : initial;
|
const mapped = initial ? await resolveAgentRunTraceCommandMapping({ traceId, mapped: initial, options }) : initial;
|
||||||
if (!mapped?.agentRun?.runId || !mapped?.agentRun?.commandId) return { result: currentResult, runnerTrace: traceStore.snapshot(traceId), found: Boolean(currentResult) };
|
if (!mapped?.agentRun?.runId || !mapped?.agentRun?.commandId) return { result: currentResult, runnerTrace: traceStore.snapshot(traceId), found: Boolean(currentResult), eventsRefreshed: false, resultSynced: false };
|
||||||
const env = options.env ?? process.env;
|
const env = options.env ?? process.env;
|
||||||
const fetchImpl = options.fetchImpl ?? globalThis.fetch;
|
const fetchImpl = options.fetchImpl ?? globalThis.fetch;
|
||||||
const managerUrl = resolveAgentRunManagerUrl(env, mapped.agentRun.managerUrl);
|
const managerUrl = resolveAgentRunManagerUrl(env, mapped.agentRun.managerUrl);
|
||||||
@@ -460,7 +460,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
|
|||||||
});
|
});
|
||||||
const payload = decorateAgentRunRunningResult({ base: { ...mapped, status: "running", agentRun: nextMapping, updatedAt: nowIso(options.now) }, mapping: nextMapping, traceStore, traceId });
|
const payload = decorateAgentRunRunningResult({ base: { ...mapped, status: "running", agentRun: nextMapping, updatedAt: nowIso(options.now) }, mapping: nextMapping, traceStore, traceId });
|
||||||
options.codeAgentChatResults?.set?.(traceId, payload);
|
options.codeAgentChatResults?.set?.(traceId, payload);
|
||||||
return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true };
|
return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: false };
|
||||||
}
|
}
|
||||||
const result = await agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(mapped.agentRun.runId)}/commands/${encodeURIComponent(mapped.agentRun.commandId)}/result`, {
|
const result = await agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(mapped.agentRun.runId)}/commands/${encodeURIComponent(mapped.agentRun.commandId)}/result`, {
|
||||||
method: "GET",
|
method: "GET",
|
||||||
@@ -481,7 +481,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
|
|||||||
const base = { ...mapped, agentRun: nextMapping, updatedAt: nowIso(options.now) };
|
const base = { ...mapped, agentRun: nextMapping, updatedAt: nowIso(options.now) };
|
||||||
const payload = agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId, appendResultEvent });
|
const payload = agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId, appendResultEvent });
|
||||||
options.codeAgentChatResults?.set?.(traceId, payload);
|
options.codeAgentChatResults?.set?.(traceId, payload);
|
||||||
return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true };
|
return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: true };
|
||||||
}
|
}
|
||||||
|
|
||||||
function agentRunCommandResultSyncRequired(mapped = {}, eventsResponse = null) {
|
function agentRunCommandResultSyncRequired(mapped = {}, eventsResponse = null) {
|
||||||
|
|||||||
@@ -2221,6 +2221,88 @@ test("cloud api running trace refreshes AgentRun events without waiting for comm
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("cloud api turn status reuses AgentRun result-sync trace refresh (#1422)", async () => {
|
||||||
|
const calls = [];
|
||||||
|
const traceId = "trc_issue1422_turn_status_fast_path";
|
||||||
|
const runId = "run_issue1422_turn_status_fast_path";
|
||||||
|
const commandId = "cmd_issue1422_turn_status_fast_path";
|
||||||
|
const events = [
|
||||||
|
{ id: "evt_issue1422_created", runId, seq: 1, type: "backend_status", payload: { phase: "runner-job-created", commandId }, createdAt: "2026-06-18T04:30:00.000Z" },
|
||||||
|
{ id: "evt_issue1422_message", runId, seq: 2, type: "assistant_message", payload: { commandId, itemId: "msg_issue1422_running", text: "正在处理请求。" }, createdAt: "2026-06-18T04:30:01.000Z" }
|
||||||
|
];
|
||||||
|
const agentRunServer = createHttpServer(async (request, response) => {
|
||||||
|
const url = new URL(request.url || "/", "http://127.0.0.1");
|
||||||
|
calls.push({ method: request.method, path: url.pathname, search: url.search });
|
||||||
|
const send = (data) => {
|
||||||
|
response.writeHead(200, { "content-type": "application/json" });
|
||||||
|
response.end(`${JSON.stringify({ ok: true, data, traceId: "trc_fake_issue1422_turn" })}\n`);
|
||||||
|
};
|
||||||
|
if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/events`) return send({ items: events });
|
||||||
|
if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/commands/${commandId}/result`) {
|
||||||
|
return send({ runId, commandId, status: "running", runStatus: "running", commandState: "running", terminalStatus: null, completed: false, reply: null, lastSeq: 2, eventCount: 2 });
|
||||||
|
}
|
||||||
|
response.writeHead(404, { "content-type": "application/json" });
|
||||||
|
response.end(`${JSON.stringify({ ok: false, message: `unexpected ${request.method} ${url.pathname}` })}\n`);
|
||||||
|
});
|
||||||
|
await new Promise((resolve) => agentRunServer.listen(0, "127.0.0.1", resolve));
|
||||||
|
|
||||||
|
const agentRunPort = agentRunServer.address().port;
|
||||||
|
const codeAgentChatResults = new Map([[traceId, {
|
||||||
|
accepted: true,
|
||||||
|
status: "running",
|
||||||
|
shortConnection: true,
|
||||||
|
traceId,
|
||||||
|
conversationId: "cnv_issue1422_turn_status",
|
||||||
|
sessionId: "ses_issue1422_turn_status",
|
||||||
|
threadId: "thread-issue1422-turn-status",
|
||||||
|
ownerUserId: TEST_AGENT_ACTOR.id,
|
||||||
|
ownerRole: TEST_AGENT_ACTOR.role,
|
||||||
|
agentRun: { adapter: "agentrun-v01", managerUrl: `http://127.0.0.1:${agentRunPort}`, runId, commandId, traceId, providerTrace: { traceId }, lastSeq: 0, status: "running", commandState: "running", valuesPrinted: false },
|
||||||
|
valuesPrinted: false
|
||||||
|
}]]);
|
||||||
|
const server = createCloudApiServer({
|
||||||
|
traceStore: createCodeAgentTraceStore(),
|
||||||
|
codeAgentChatResults,
|
||||||
|
env: {
|
||||||
|
HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01",
|
||||||
|
AGENTRUN_MGR_URL: `http://127.0.0.1:${agentRunPort}`,
|
||||||
|
HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1",
|
||||||
|
HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "D601",
|
||||||
|
HWLAB_ENVIRONMENT: "v02",
|
||||||
|
HWLAB_GITOPS_PROFILE: "v02"
|
||||||
|
},
|
||||||
|
accessController: {
|
||||||
|
required: false,
|
||||||
|
async authenticate() {
|
||||||
|
return { ok: true, actor: TEST_AGENT_ACTOR, session: TEST_AUTH_SESSION };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||||
|
|
||||||
|
try {
|
||||||
|
const { port } = server.address();
|
||||||
|
const response = await fetch(`http://127.0.0.1:${port}/v1/agent/turns/${traceId}`, {
|
||||||
|
headers: { cookie: "hwlab_session=test-stub-session" }
|
||||||
|
});
|
||||||
|
assert.equal(response.status, 200);
|
||||||
|
const body = await response.json();
|
||||||
|
assert.equal(body.status, "running");
|
||||||
|
assert.ok(body.eventCount >= 1);
|
||||||
|
assert.ok(Array.isArray(body.runnerTrace.events));
|
||||||
|
assert.ok(body.runnerTrace.events.length >= 1);
|
||||||
|
assert.equal(calls.filter((call) => call.path === `/api/v1/runs/${runId}/events`).length, 1);
|
||||||
|
assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${commandId}/result`), false);
|
||||||
|
} 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()));
|
||||||
|
});
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
test("cloud api repairs historical same-session AgentRun trace after lastTraceId advances (#955)", async () => {
|
test("cloud api repairs historical same-session AgentRun trace after lastTraceId advances (#955)", async () => {
|
||||||
const calls = [];
|
const calls = [];
|
||||||
const ownerSessions = new Map();
|
const ownerSessions = new Map();
|
||||||
|
|||||||
@@ -1214,10 +1214,12 @@ async function resolveCodeAgentTurnStatusSnapshot(traceId, options) {
|
|||||||
|
|
||||||
const adapterEnabled = codeAgentAgentRunAdapterEnabled(options.env ?? process.env);
|
const adapterEnabled = codeAgentAgentRunAdapterEnabled(options.env ?? process.env);
|
||||||
let resultPollError = null;
|
let resultPollError = null;
|
||||||
|
let turnRefreshSatisfiedByResultSync = false;
|
||||||
if (adapterEnabled && (result?.agentRun?.runId || !result)) {
|
if (adapterEnabled && (result?.agentRun?.runId || !result)) {
|
||||||
try {
|
try {
|
||||||
const synced = await syncAgentRunChatResult({ traceId, currentResult: result, options: refreshOptions, traceStore });
|
const synced = await syncAgentRunChatResult({ traceId, currentResult: result, options: refreshOptions, traceStore });
|
||||||
result = synced.result ?? result;
|
result = synced.result ?? result;
|
||||||
|
turnRefreshSatisfiedByResultSync = synced.eventsRefreshed === true || synced.resultSynced === true;
|
||||||
if (result && !canAccessOwnedResult(result, options.actor)) return forbiddenTurnSnapshot(traceId);
|
if (result && !canAccessOwnedResult(result, options.actor)) return forbiddenTurnSnapshot(traceId);
|
||||||
if (result && isTraceCommandTerminalStatus(result.status)) {
|
if (result && isTraceCommandTerminalStatus(result.status)) {
|
||||||
await finalizeCodeAgentBillingUsage({ payload: result, params: result, options });
|
await finalizeCodeAgentBillingUsage({ payload: result, params: result, options });
|
||||||
@@ -1254,7 +1256,7 @@ async function resolveCodeAgentTurnStatusSnapshot(traceId, options) {
|
|||||||
if (agentRunResult && !canAccessOwnedResult(agentRunResult, options.actor)) return forbiddenTurnSnapshot(traceId);
|
if (agentRunResult && !canAccessOwnedResult(agentRunResult, options.actor)) return forbiddenTurnSnapshot(traceId);
|
||||||
|
|
||||||
let refreshError = null;
|
let refreshError = null;
|
||||||
if (agentRunResult?.agentRun) {
|
if (agentRunResult?.agentRun && !turnRefreshSatisfiedByResultSync && !resultPollError) {
|
||||||
try {
|
try {
|
||||||
const refreshedTrace = await refreshAgentRunTrace({ traceId, result: agentRunResult, options: refreshOptions, traceStore });
|
const refreshedTrace = await refreshAgentRunTrace({ traceId, result: agentRunResult, options: refreshOptions, traceStore });
|
||||||
agentRunResult = options.codeAgentChatResults?.get?.(traceId) ?? agentRunResult;
|
agentRunResult = options.codeAgentChatResults?.get?.(traceId) ?? agentRunResult;
|
||||||
|
|||||||
Reference in New Issue
Block a user