Merge pull request #1493 from pikasTech/fix/issue-1422-turn-session-cache
fix: cache Code Agent turn session lookup
This commit is contained in:
@@ -759,7 +759,7 @@ export async function steerAgentRunChatTurn({ traceId, currentResult = null, par
|
|||||||
export async function loadPersistedAgentRunResult(traceId, options = {}) {
|
export async function loadPersistedAgentRunResult(traceId, options = {}) {
|
||||||
const safeId = safeTraceId(traceId);
|
const safeId = safeTraceId(traceId);
|
||||||
if (!safeId || typeof options.accessController?.getAgentSessionByTraceId !== "function") return null;
|
if (!safeId || typeof options.accessController?.getAgentSessionByTraceId !== "function") return null;
|
||||||
const session = await options.accessController.getAgentSessionByTraceId(safeId);
|
const session = await loadAgentSessionByTraceId(safeId, options);
|
||||||
const agentRun = agentRunSeedFromSession(session, safeId);
|
const agentRun = agentRunSeedFromSession(session, safeId);
|
||||||
if (!agentRun?.runId) return null;
|
if (!agentRun?.runId) return null;
|
||||||
const backendProfile = firstNonEmpty(agentRun.backendProfile, "deepseek");
|
const backendProfile = firstNonEmpty(agentRun.backendProfile, "deepseek");
|
||||||
@@ -792,6 +792,19 @@ export async function loadPersistedAgentRunResult(traceId, options = {}) {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function loadAgentSessionByTraceId(traceId, options = {}) {
|
||||||
|
const safeId = safeTraceId(traceId);
|
||||||
|
if (!safeId || typeof options.accessController?.getAgentSessionByTraceId !== "function") return null;
|
||||||
|
const cache = options.codeAgentTraceSessionCache;
|
||||||
|
if (cache instanceof Map) {
|
||||||
|
if (cache.has(safeId)) return cache.get(safeId);
|
||||||
|
const session = await options.accessController.getAgentSessionByTraceId(safeId);
|
||||||
|
cache.set(safeId, session ?? null);
|
||||||
|
return session ?? null;
|
||||||
|
}
|
||||||
|
return await options.accessController.getAgentSessionByTraceId(safeId);
|
||||||
|
}
|
||||||
|
|
||||||
function agentRunSeedFromSession(session, traceId) {
|
function agentRunSeedFromSession(session, traceId) {
|
||||||
const snapshot = session?.session && typeof session.session === "object" ? session.session : null;
|
const snapshot = session?.session && typeof session.session === "object" ? session.session : null;
|
||||||
const traceResults = snapshot?.traceResults && typeof snapshot.traceResults === "object" ? snapshot.traceResults : null;
|
const traceResults = snapshot?.traceResults && typeof snapshot.traceResults === "object" ? snapshot.traceResults : null;
|
||||||
|
|||||||
@@ -2303,6 +2303,99 @@ test("cloud api turn status reuses AgentRun result-sync trace refresh (#1422)",
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("cloud api turn status reuses request session lookup for project check and persisted AgentRun result (#1422)", async () => {
|
||||||
|
const calls = [];
|
||||||
|
const traceId = "trc_issue1422_turn_status_session_cache";
|
||||||
|
const runId = "run_issue1422_turn_status_session_cache";
|
||||||
|
const commandId = "cmd_issue1422_turn_status_session_cache";
|
||||||
|
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_session_cache" })}\n`);
|
||||||
|
};
|
||||||
|
if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/events`) {
|
||||||
|
return send({ items: [
|
||||||
|
{ id: "evt_issue1422_session_cache_running", runId, seq: 1, type: "backend_status", payload: { phase: "runner-job-created", commandId }, createdAt: "2026-06-18T05:20:00.000Z" }
|
||||||
|
] });
|
||||||
|
}
|
||||||
|
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;
|
||||||
|
let sessionLookupCount = 0;
|
||||||
|
const session = testAgentSessionRecord({
|
||||||
|
sessionId: "ses_issue1422_turn_status_session_cache",
|
||||||
|
projectId: "prj_hwpod_workbench",
|
||||||
|
conversationId: "cnv_issue1422_turn_status_session_cache",
|
||||||
|
threadId: "thread-issue1422-turn-status-session-cache",
|
||||||
|
lastTraceId: traceId,
|
||||||
|
status: "running",
|
||||||
|
session: {
|
||||||
|
agentRun: {
|
||||||
|
adapter: "agentrun-v01",
|
||||||
|
managerUrl: `http://127.0.0.1:${agentRunPort}`,
|
||||||
|
runId,
|
||||||
|
commandId,
|
||||||
|
traceId,
|
||||||
|
providerTrace: { traceId, valuesPrinted: false },
|
||||||
|
lastSeq: 0,
|
||||||
|
status: "running",
|
||||||
|
commandState: "running",
|
||||||
|
backendProfile: "deepseek",
|
||||||
|
valuesPrinted: false
|
||||||
|
},
|
||||||
|
valuesRedacted: true
|
||||||
|
}
|
||||||
|
});
|
||||||
|
const server = createCloudApiServer({
|
||||||
|
traceStore: createCodeAgentTraceStore(),
|
||||||
|
codeAgentChatResults: new Map(),
|
||||||
|
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 };
|
||||||
|
},
|
||||||
|
async getAgentSessionByTraceId(requestTraceId) {
|
||||||
|
sessionLookupCount += 1;
|
||||||
|
return requestTraceId === traceId ? session : null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
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}?projectId=prj_hwpod_workbench`, {
|
||||||
|
headers: { cookie: "hwlab_session=test-stub-session" }
|
||||||
|
});
|
||||||
|
assert.equal(response.status, 200);
|
||||||
|
const body = await response.json();
|
||||||
|
assert.equal(body.status, "running");
|
||||||
|
assert.equal(body.sessionId, "ses_issue1422_turn_status_session_cache");
|
||||||
|
assert.equal(sessionLookupCount, 1);
|
||||||
|
assert.equal(calls.filter((call) => call.path === `/api/v1/runs/${runId}/events`).length, 1);
|
||||||
|
} 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 turn status skips AgentRun refresh for complete terminal evidence (#1422)", async () => {
|
test("cloud api turn status skips AgentRun refresh for complete terminal evidence (#1422)", async () => {
|
||||||
const calls = [];
|
const calls = [];
|
||||||
const traceId = "trc_issue1422_terminal_fast_path";
|
const traceId = "trc_issue1422_terminal_fast_path";
|
||||||
|
|||||||
@@ -1143,12 +1143,13 @@ export async function handleCodeAgentTurnHttp(request, response, url, options) {
|
|||||||
});
|
});
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const projectMismatch = await traceProjectMismatchSnapshot(traceId, url, options, "turn");
|
const requestOptions = codeAgentRequestScopedOptions(options);
|
||||||
|
const projectMismatch = await traceProjectMismatchSnapshot(traceId, url, requestOptions, "turn");
|
||||||
if (projectMismatch) {
|
if (projectMismatch) {
|
||||||
sendJson(response, projectMismatch.statusCode, projectMismatch.body);
|
sendJson(response, projectMismatch.statusCode, projectMismatch.body);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const resolved = await resolveCodeAgentTurnStatusSnapshot(traceId, options);
|
const resolved = await resolveCodeAgentTurnStatusSnapshot(traceId, requestOptions);
|
||||||
sendJson(response, resolved.statusCode, resolved.body);
|
sendJson(response, resolved.statusCode, resolved.body);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1237,6 +1238,13 @@ function codeAgentTurnStatusRefreshOptions(options = {}) {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function codeAgentRequestScopedOptions(options = {}) {
|
||||||
|
return {
|
||||||
|
...options,
|
||||||
|
codeAgentTraceSessionCache: options.codeAgentTraceSessionCache instanceof Map ? options.codeAgentTraceSessionCache : new Map()
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
function forbiddenTurnSnapshot(traceId) {
|
function forbiddenTurnSnapshot(traceId) {
|
||||||
return {
|
return {
|
||||||
statusCode: 403,
|
statusCode: 403,
|
||||||
@@ -1257,7 +1265,7 @@ function forbiddenTurnSnapshot(traceId) {
|
|||||||
async function traceProjectMismatchSnapshot(traceId, url, options, resourceKind) {
|
async function traceProjectMismatchSnapshot(traceId, url, options, resourceKind) {
|
||||||
const requestedProjectId = textValue(url.searchParams.get("projectId"));
|
const requestedProjectId = textValue(url.searchParams.get("projectId"));
|
||||||
if (!requestedProjectId || !options.accessController?.getAgentSessionByTraceId) return null;
|
if (!requestedProjectId || !options.accessController?.getAgentSessionByTraceId) return null;
|
||||||
const session = await options.accessController.getAgentSessionByTraceId(traceId);
|
const session = await getCodeAgentSessionByTraceId(traceId, options);
|
||||||
if (!session) return null;
|
if (!session) return null;
|
||||||
if (session.ownerUserId && options.actor?.role !== "admin" && session.ownerUserId !== options.actor?.id) return forbiddenTurnSnapshot(traceId);
|
if (session.ownerUserId && options.actor?.role !== "admin" && session.ownerUserId !== options.actor?.id) return forbiddenTurnSnapshot(traceId);
|
||||||
const actualProjectId = textValue(session.projectId);
|
const actualProjectId = textValue(session.projectId);
|
||||||
@@ -1278,6 +1286,19 @@ async function traceProjectMismatchSnapshot(traceId, url, options, resourceKind)
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function getCodeAgentSessionByTraceId(traceId, options = {}) {
|
||||||
|
const safeId = safeTraceId(traceId);
|
||||||
|
if (!safeId || typeof options.accessController?.getAgentSessionByTraceId !== "function") return null;
|
||||||
|
const cache = options.codeAgentTraceSessionCache;
|
||||||
|
if (cache instanceof Map) {
|
||||||
|
if (cache.has(safeId)) return cache.get(safeId);
|
||||||
|
const session = await options.accessController.getAgentSessionByTraceId(safeId);
|
||||||
|
cache.set(safeId, session ?? null);
|
||||||
|
return session ?? null;
|
||||||
|
}
|
||||||
|
return await options.accessController.getAgentSessionByTraceId(safeId);
|
||||||
|
}
|
||||||
|
|
||||||
function codeAgentTurnStatusPayload({ traceId, result, snapshot, resultPollError, refreshError, options }) {
|
function codeAgentTurnStatusPayload({ traceId, result, snapshot, resultPollError, refreshError, options }) {
|
||||||
const resultObject = result && typeof result === "object" ? result : null;
|
const resultObject = result && typeof result === "object" ? result : null;
|
||||||
const snapshotObject = snapshot && typeof snapshot === "object" ? snapshot : null;
|
const snapshotObject = snapshot && typeof snapshot === "object" ? snapshot : null;
|
||||||
@@ -2502,13 +2523,14 @@ export async function handleCodeAgentTraceHttp(request, response, url, options)
|
|||||||
});
|
});
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const projectMismatch = await traceProjectMismatchSnapshot(traceId, url, options, "trace");
|
const requestOptions = codeAgentRequestScopedOptions(options);
|
||||||
|
const projectMismatch = await traceProjectMismatchSnapshot(traceId, url, requestOptions, "trace");
|
||||||
if (projectMismatch) {
|
if (projectMismatch) {
|
||||||
sendJson(response, projectMismatch.statusCode, projectMismatch.body);
|
sendJson(response, projectMismatch.statusCode, projectMismatch.body);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const result = options.codeAgentChatResults?.get(traceId) ?? null;
|
const result = requestOptions.codeAgentChatResults?.get(traceId) ?? null;
|
||||||
if (result && !canAccessOwnedResult(result, options.actor)) {
|
if (result && !canAccessOwnedResult(result, requestOptions.actor)) {
|
||||||
sendJson(response, 403, {
|
sendJson(response, 403, {
|
||||||
error: {
|
error: {
|
||||||
code: "agent_session_owner_required",
|
code: "agent_session_owner_required",
|
||||||
@@ -2518,9 +2540,9 @@ export async function handleCodeAgentTraceHttp(request, response, url, options)
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
let agentRunResult = result?.agentRun ? result : await loadPersistedAgentRunResult(traceId, options);
|
let agentRunResult = result?.agentRun ? result : await loadPersistedAgentRunResult(traceId, requestOptions);
|
||||||
let refreshError = null;
|
let refreshError = null;
|
||||||
if (agentRunResult && !canAccessOwnedResult(agentRunResult, options.actor)) {
|
if (agentRunResult && !canAccessOwnedResult(agentRunResult, requestOptions.actor)) {
|
||||||
sendJson(response, 403, {
|
sendJson(response, 403, {
|
||||||
error: {
|
error: {
|
||||||
code: "agent_session_owner_required",
|
code: "agent_session_owner_required",
|
||||||
@@ -2531,16 +2553,16 @@ export async function handleCodeAgentTraceHttp(request, response, url, options)
|
|||||||
}
|
}
|
||||||
if (agentRunResult?.agentRun) {
|
if (agentRunResult?.agentRun) {
|
||||||
try {
|
try {
|
||||||
const snapshot = await refreshAgentRunTrace({ traceId, result: agentRunResult, options, traceStore });
|
const snapshot = await refreshAgentRunTrace({ traceId, result: agentRunResult, options: requestOptions, traceStore });
|
||||||
agentRunResult = options.codeAgentChatResults?.get?.(traceId) ?? agentRunResult;
|
agentRunResult = requestOptions.codeAgentChatResults?.get?.(traceId) ?? agentRunResult;
|
||||||
if (traceNeedsCommandResultSync(agentRunResult, snapshot)) {
|
if (traceNeedsCommandResultSync(agentRunResult, snapshot)) {
|
||||||
const synced = await syncAgentRunChatResult({ traceId, currentResult: agentRunResult, options, traceStore, appendResultEvent: false, refreshEvents: false, forceResultSync: true });
|
const synced = await syncAgentRunChatResult({ traceId, currentResult: agentRunResult, options: requestOptions, traceStore, appendResultEvent: false, refreshEvents: false, forceResultSync: true });
|
||||||
agentRunResult = synced.result ?? agentRunResult;
|
agentRunResult = synced.result ?? agentRunResult;
|
||||||
}
|
}
|
||||||
if (isTraceCommandTerminalStatus(agentRunResult?.status)) {
|
if (isTraceCommandTerminalStatus(agentRunResult?.status)) {
|
||||||
await finalizeCodeAgentBillingUsage({ payload: agentRunResult, params: agentRunResult, options });
|
await finalizeCodeAgentBillingUsage({ payload: agentRunResult, params: agentRunResult, options: requestOptions });
|
||||||
recordCodeAgentConversationFact(agentRunResult, options);
|
recordCodeAgentConversationFact(agentRunResult, requestOptions);
|
||||||
await recordCodeAgentSessionOwner({ payload: agentRunResult, params: agentRunResult, options, status: codeAgentOwnerStatusForResult(agentRunResult), preserveLastTraceId: true });
|
await recordCodeAgentSessionOwner({ payload: agentRunResult, params: agentRunResult, options: requestOptions, status: codeAgentOwnerStatusForResult(agentRunResult), preserveLastTraceId: true });
|
||||||
}
|
}
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
refreshError = error;
|
refreshError = error;
|
||||||
@@ -2548,7 +2570,7 @@ export async function handleCodeAgentTraceHttp(request, response, url, options)
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (streamSegment === "stream") {
|
if (streamSegment === "stream") {
|
||||||
sendTraceSse(response, traceStore, traceId, options);
|
sendTraceSse(response, traceStore, traceId, requestOptions);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user