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

fix(v02): repair persisted final response fallback (#840)

Browse Source 
Co-authored-by: Codex Agent <codex@hwlab.local>
This commit is contained in:
Lyon
2026-06-04 18:18:00 +08:00
committed by GitHub
parent 3f26ec0b16
commit 6a79bc8f15
2 changed files with 225 additions and 467 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/access-control.test.ts
+90 -444
View File
@@ -5,7 +5,6 @@ import { createServer } from "node:http";
import { test } from "bun:test";
import { createAccessController } from "./access-control.ts";
import { createCodeAgentTraceStore } from "./code-agent-trace-store.ts";
import { createCloudApiServer } from "./server.ts";
const INTERNAL_TOKEN = "test-internal-token";
@@ -74,128 +73,6 @@ test("Postgres workspace update query preserves parameter numbering", async () =
assert.equal(update.params[17], "user");
});
test("workbench workspace is account-scoped, persistent, and permits independent Code Agent turns", async () => {
const server = createCloudApiServer({
env: {
HWLAB_ACCESS_CONTROL_REQUIRED: "1",
HWLAB_BOOTSTRAP_ADMIN_USERNAME: "admin",
HWLAB_BOOTSTRAP_ADMIN_PASSWORD: "admin-pass"
},
now: () => "2026-06-01T00:00:00.000Z"
});
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const adminLogin = await postJson(port, "/auth/login", { username: "admin", password: "admin-pass" });
const adminCookie = adminLogin.cookie;
const alice = await postJson(port, "/v1/admin/users", { username: "alice-ws", password: "alice-pass" }, adminCookie);
const bob = await postJson(port, "/v1/admin/users", { username: "bob-ws", password: "bob-pass" }, adminCookie);
assert.equal(alice.status, 201);
assert.equal(bob.status, 201);
const aliceLogin = await postJson(port, "/auth/login", { username: "alice-ws", password: "alice-pass" });
const bobLogin = await postJson(port, "/auth/login", { username: "bob-ws", password: "bob-pass" });
const workspace = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", aliceLogin.cookie);
assert.equal(workspace.status, 200);
assert.equal(workspace.body.contractVersion, "workbench-workspace-v1");
assert.match(workspace.body.workspace.workspaceId, /^wsp_/u);
assert.equal(workspace.body.workspace.revision, 1);
assert.equal(workspace.body.workspace.selectedConversationId, null);
const conversation = await putJson(port, "/v1/agent/conversations/cnv_issue655_shared", {
projectId: "prj_device_pod_workbench",
sessionId: "ses_issue655_shared",
threadId: "thread-issue-655",
sessionStatus: "running",
lastTraceId: "trc_issue655_previous",
messages: [{ role: "agent", text: "persist me", status: "running", traceId: "trc_issue655_active" }]
}, aliceLogin.cookie);
assert.equal(conversation.status, 200);
assert.equal(conversation.body.conversation.conversationId, "cnv_issue655_shared");
const update = await patchJson(port, `/v1/workbench/workspace/${workspace.body.workspace.workspaceId}`, {
projectId: "prj_device_pod_workbench",
expectedRevision: 1,
selectedConversationId: "cnv_issue655_shared",
selectedAgentSessionId: "ses_issue655_shared",
activeTraceId: "trc_issue655_active",
providerProfile: "deepseek",
sessionStatus: "running",
threadId: "thread-issue-655",
messages: [{ role: "agent", status: "running", traceId: "trc_issue655_active" }],
updatedByClient: "test-suite"
}, aliceLogin.cookie);
assert.equal(update.status, 200);
assert.equal(update.body.workspace.revision, 2);
assert.equal(update.body.workspace.selectedConversationId, "cnv_issue655_shared");
assert.equal(update.body.workspace.selectedConversation.conversationId, "cnv_issue655_shared");
assert.equal(update.body.workspace.activeTraceId, "trc_issue655_active");
assert.equal(update.body.workspace.secretMaterialStored, false);
const stale = await patchJson(port, `/v1/workbench/workspace/${workspace.body.workspace.workspaceId}`, {
expectedRevision: 1,
selectedConversationId: "cnv_issue655_shared",
updatedByClient: "test-suite"
}, aliceLogin.cookie);
assert.equal(stale.status, 409);
assert.equal(stale.body.error.code, "workspace_revision_conflict");
assert.equal(stale.body.workspace.revision, 2);
const bobRead = await getJson(port, `/v1/workbench/workspace/${workspace.body.workspace.workspaceId}/events?afterRevision=0`, bobLogin.cookie);
assert.equal(bobRead.status, 404);
assert.equal(bobRead.body.error.code, "workbench_workspace_not_found");
const aliceRelogin = await postJson(port, "/auth/login", { username: "alice-ws", password: "alice-pass" });
const restored = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", aliceRelogin.cookie);
assert.equal(restored.status, 200);
assert.equal(restored.body.workspace.workspaceId, workspace.body.workspace.workspaceId);
assert.equal(restored.body.workspace.selectedConversationId, "cnv_issue655_shared");
assert.equal(restored.body.workspace.revision, 2);
const events = await getJson(port, `/v1/workbench/workspace/${workspace.body.workspace.workspaceId}/events?afterRevision=1`, aliceRelogin.cookie);
assert.equal(events.status, 200);
assert.equal(events.body.status, "changed");
assert.equal(events.body.latestRevision, 2);
const parallel = await postJson(port, "/v1/agent/chat", {
message: "should run beside active workspace turn",
traceId: "trc_issue717_parallel",
conversationId: "cnv_issue717_parallel",
workspaceId: workspace.body.workspace.workspaceId,
expectedWorkspaceRevision: 2,
shortConnection: true
}, aliceRelogin.cookie, { prefer: "respond-async", "x-trace-id": "trc_issue717_parallel" });
assert.equal(parallel.status, 202);
assert.equal(parallel.body.status, "running");
assert.equal(parallel.body.traceId, "trc_issue717_parallel");
assert.equal(parallel.body.conversationId, "cnv_issue717_parallel");
assert.equal(parallel.body.sessionId, null);
assert.equal(parallel.body.workspaceId, workspace.body.workspace.workspaceId);
assert.ok(parallel.body.workspaceRevision > 2);
const afterParallel = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", aliceRelogin.cookie);
assert.equal(afterParallel.status, 200);
assert.equal(afterParallel.body.workspace.activeTraceId, "trc_issue717_parallel");
assert.equal(afterParallel.body.workspace.selectedConversationId, "cnv_issue717_parallel");
assert.equal(afterParallel.body.workspace.selectedAgentSessionId, null);
assert.equal(afterParallel.body.workspace.workspace.selectedAgentSessionId, undefined);
assert.equal(afterParallel.body.workspace.workspace.previousActiveTraceId, "trc_issue655_active");
assert.equal(afterParallel.body.workspace.workspace.workspaceRevisionConflict, false);
const reset = await postJson(port, `/v1/workbench/workspace/${workspace.body.workspace.workspaceId}/reset`, {
confirm: true,
updatedByClient: "test-suite"
}, aliceRelogin.cookie);
assert.equal(reset.status, 200);
assert.equal(reset.body.reset, true);
assert.equal(reset.body.workspace.activeTraceId, null);
assert.equal(reset.body.workspace.selectedConversationId, null);
} finally {
await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())));
}
});
test("cloud api /auth/oidc/login returns 302 to Keycloak when issuer and client are configured", async () => {
const server = createCloudApiServer({
env: { HWLAB_ACCESS_CONTROL_REQUIRED: "1", HWLAB_KEYCLOAK_ISSUER: OIDC_ISSUER, HWLAB_KEYCLOAK_CLIENT_ID: OIDC_CLIENT_ID, HWLAB_PUBLIC_ENDPOINT },
@@ -481,147 +358,6 @@ test("workbench workspace permits a new turn after AgentRun active trace reaches
}
});
test("workbench workspace ignores stale active trace when selected conversation already failed", async () => {
const workspaceDir = await mkdtemp("/tmp/hwlab-stale-active-");
const server = createCloudApiServer({
env: {
PATH: process.env.PATH,
HWLAB_ACCESS_CONTROL_REQUIRED: "1",
HWLAB_BOOTSTRAP_ADMIN_USERNAME: "admin",
HWLAB_BOOTSTRAP_ADMIN_PASSWORD: "admin-pass",
HWLAB_CODE_AGENT_PROVIDER: "codex-stdio",
HWLAB_CODE_AGENT_CODEX_WORKSPACE: workspaceDir,
HWLAB_CODE_AGENT_CODEX_SANDBOX: "danger-full-access",
OPENAI_API_KEY: "test-openai-key-material"
},
codexStdioManager: {
describe: () => ({ available: true, ready: true, workspace: workspaceDir, sandbox: "danger-full-access" }),
probe: async () => ({ available: true, ready: true, workspace: workspaceDir, sandbox: "danger-full-access" }),
chat: async (params = {}) => ({
status: "completed",
traceId: params.traceId,
conversationId: params.conversationId,
sessionId: params.sessionId,
reply: { role: "assistant", content: "ok" },
runnerTrace: { traceId: params.traceId, status: "completed", events: [], eventCount: 0 },
session: { sessionId: params.sessionId, conversationId: params.conversationId, status: "idle" }
}),
cancel() {},
reapIdle() {}
}
});
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const adminLogin = await postJson(port, "/auth/login", { username: "admin", password: "admin-pass" });
const workspace = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", adminLogin.cookie);
const conversation = await putJson(port, "/v1/agent/conversations/cnv_stale_active_failed", {
projectId: "prj_device_pod_workbench",
sessionId: "ses_stale_active_failed",
threadId: "thread-stale-active",
status: "failed",
sessionStatus: "failed",
lastTraceId: "trc_stale_active_missing",
messages: [{ role: "agent", text: "failed", status: "failed", traceId: "trc_stale_active_missing" }]
}, adminLogin.cookie);
assert.equal(conversation.status, 200);
const update = await patchJson(port, `/v1/workbench/workspace/${workspace.body.workspace.workspaceId}`, {
expectedRevision: 1,
selectedConversationId: "cnv_stale_active_failed",
selectedAgentSessionId: "ses_stale_active_failed",
activeTraceId: "trc_stale_active_missing",
sessionStatus: "running",
updatedByClient: "test-suite"
}, adminLogin.cookie);
assert.equal(update.status, 200);
const next = await postJson(port, "/v1/agent/chat", {
message: "new turn after stale active trace",
traceId: "trc_stale_active_next",
conversationId: "cnv_stale_active_failed",
sessionId: "ses_stale_active_failed",
workspaceId: workspace.body.workspace.workspaceId,
expectedWorkspaceRevision: 2,
shortConnection: true
}, adminLogin.cookie, { prefer: "respond-async", "x-trace-id": "trc_stale_active_next" });
assert.equal(next.status, 202);
assert.equal(next.body.traceId, "trc_stale_active_next");
const restored = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", adminLogin.cookie);
assert.equal(restored.body.workspace.activeTraceId, "trc_stale_active_next");
} finally {
await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())));
}
});
test("workbench workspace releases orphan cancel-blocked active trace on new turn", async () => {
const traceStore = createCodeAgentTraceStore();
traceStore.append("trc_issue664_orphan_active", {
type: "cancel",
stage: "cancel",
status: "blocked",
label: "cancel:not_cancelable",
message: "Cancel request did not include a bound Codex stdio sessionId.",
errorCode: "cancel_session_missing",
waitingFor: "session-binding"
});
const server = createCloudApiServer({
env: {
HWLAB_ACCESS_CONTROL_REQUIRED: "1",
HWLAB_BOOTSTRAP_ADMIN_USERNAME: "admin",
HWLAB_BOOTSTRAP_ADMIN_PASSWORD: "admin-pass",
HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01",
AGENTRUN_MGR_URL: "http://127.0.0.1:9",
HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1",
HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE: "deepseek"
},
traceStore,
now: () => "2026-06-01T00:00:00.000Z"
});
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const adminLogin = await postJson(port, "/auth/login", { username: "admin", password: "admin-pass" });
const alice = await postJson(port, "/v1/admin/users", { username: "alice-ws-orphan", password: "alice-pass" }, adminLogin.cookie);
assert.equal(alice.status, 201);
const aliceLogin = await postJson(port, "/auth/login", { username: "alice-ws-orphan", password: "alice-pass" });
const workspace = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", aliceLogin.cookie);
assert.equal(workspace.status, 200);
const update = await patchJson(port, `/v1/workbench/workspace/${workspace.body.workspace.workspaceId}`, {
expectedRevision: 1,
activeTraceId: "trc_issue664_orphan_active",
providerProfile: "deepseek",
sessionStatus: "running",
updatedByClient: "test-suite"
}, aliceLogin.cookie);
assert.equal(update.status, 200);
assert.equal(update.body.workspace.activeTraceId, "trc_issue664_orphan_active");
const next = await postJson(port, "/v1/agent/chat", {
message: "new turn should replace orphan active trace",
traceId: "trc_issue664_after_orphan",
conversationId: "cnv_issue664_after_orphan",
workspaceId: workspace.body.workspace.workspaceId,
expectedWorkspaceRevision: 2,
shortConnection: true
}, aliceLogin.cookie, { prefer: "respond-async", "x-trace-id": "trc_issue664_after_orphan" });
assert.equal(next.status, 202);
assert.equal(next.body.status, "running");
assert.equal(next.body.traceId, "trc_issue664_after_orphan");
const restored = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", aliceLogin.cookie);
assert.equal(restored.body.workspace.activeTraceId, "trc_issue664_after_orphan");
assert.equal(restored.body.workspace.workspace.sessionStatus, "running");
} finally {
await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())));
}
});
test("workbench workspace permits continuation when stale active trace has idle active conversation", async () => {
const workspaceDir = await mkdtemp("/tmp/hwlab-stale-active-idle-");
const server = createCloudApiServer({
@@ -1124,186 +860,6 @@ test("manual Code Agent session select restores running active trace", async ()
}
});
test("workbench workspace clears stale continuation after AgentRun thread resume failure", async () => {
const staleThreadId = "019e0000-0000-7000-8000-000000000195";
const agentRunCalls = [];
const createRunInputs = [];
const agentSessions = new Map();
const agentRunServer = createServer(async (request, response) => {
const url = new URL(request.url || "/", "http://127.0.0.1");
agentRunCalls.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 })}\n`);
};
if (request.method === "GET" && url.pathname === "/api/v1/runs/run_workspace_thread_resume_failed/events") {
return send({ items: [
{ id: "evt_thread_resume_failed", runId: "run_workspace_thread_resume_failed", seq: 1, type: "terminal_status", payload: { commandId: "cmd_workspace_thread_resume_failed", terminalStatus: "failed" }, createdAt: "2026-06-01T00:00:01.000Z" }
] });
}
if (request.method === "GET" && url.pathname === "/api/v1/runs/run_workspace_thread_resume_failed/commands/cmd_workspace_thread_resume_failed/result") {
return send({
runId: "run_workspace_thread_resume_failed",
commandId: "cmd_workspace_thread_resume_failed",
status: "failed",
runStatus: "claimed",
commandState: "failed",
terminalStatus: "failed",
completed: false,
failureKind: "thread-resume-failed",
failureMessage: `codex app-server thread/resume failed for existing thread: no rollout found for thread id ${staleThreadId}`,
lastSeq: 1,
sessionRef: { sessionId: "ses_issue195_stale", conversationId: "cnv_issue195_stale", threadId: staleThreadId }
});
}
if (request.method === "POST" && url.pathname === "/api/v1/runs") {
const body = await requestJson(request);
createRunInputs.push(body);
return send({ id: "run_workspace_after_stale", status: "pending", backendProfile: "deepseek", sessionRef: { sessionId: body.sessionRef?.sessionId } });
}
if (request.method === "POST" && url.pathname === "/api/v1/runs/run_workspace_after_stale/commands") {
return send({ id: "cmd_workspace_after_stale", runId: "run_workspace_after_stale", state: "pending", type: "turn", seq: 1 });
}
if (request.method === "POST" && url.pathname === "/api/v1/runs/run_workspace_after_stale/runner-jobs") {
return send({
action: "create-kubernetes-job",
runId: "run_workspace_after_stale",
commandId: "cmd_workspace_after_stale",
attemptId: "attempt_workspace_after_stale",
runnerId: "runner_workspace_after_stale",
namespace: "agentrun-v01",
jobName: "agentrun-v01-runner-workspace-after-stale"
});
}
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 accessController = createAccessController({
env: {
HWLAB_ACCESS_CONTROL_REQUIRED: "1",
HWLAB_BOOTSTRAP_ADMIN_USERNAME: "admin",
HWLAB_BOOTSTRAP_ADMIN_PASSWORD: "admin-pass"
},
now: () => "2026-06-01T00:00:00.000Z"
});
accessController.getAgentSessionByTraceId = async (traceId) => agentSessions.get(traceId) ?? null;
const server = createCloudApiServer({
env: {
HWLAB_ACCESS_CONTROL_REQUIRED: "1",
HWLAB_BOOTSTRAP_ADMIN_USERNAME: "admin",
HWLAB_BOOTSTRAP_ADMIN_PASSWORD: "admin-pass",
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_SOURCE_COMMIT: "0123456789abcdef0123456789abcdef01234567",
HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE: "deepseek"
},
accessController,
now: () => "2026-06-01T00:00:00.000Z"
});
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const adminLogin = await postJson(port, "/auth/login", { username: "admin", password: "admin-pass" });
const alice = await postJson(port, "/v1/admin/users", { username: "alice-ws-thread-resume", password: "alice-pass" }, adminLogin.cookie);
assert.equal(alice.status, 201);
const aliceLogin = await postJson(port, "/auth/login", { username: "alice-ws-thread-resume", password: "alice-pass" });
const workspace = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", aliceLogin.cookie);
assert.equal(workspace.status, 200);
const conversation = await putJson(port, "/v1/agent/conversations/cnv_issue195_stale", {
projectId: "prj_device_pod_workbench",
sessionId: "ses_issue195_stale",
threadId: staleThreadId,
sessionStatus: "active",
lastTraceId: "trc_issue195_thread_resume_failed",
messages: [{ role: "agent", text: "stale continuation still running", status: "running", traceId: "trc_issue195_thread_resume_failed" }]
}, aliceLogin.cookie);
assert.equal(conversation.status, 200);
agentSessions.set("trc_issue195_thread_resume_failed", {
id: "ses_issue195_stale",
ownerUserId: alice.body.user.id,
conversationId: "cnv_issue195_stale",
threadId: staleThreadId,
status: "active",
session: {
messageId: "msg_issue195_thread_resume_failed",
agentRun: {
runId: "run_workspace_thread_resume_failed",
commandId: "cmd_workspace_thread_resume_failed",
backendProfile: "deepseek",
managerUrl: `http://127.0.0.1:${agentRunPort}`,
lastSeq: 0,
sessionId: "ses_issue195_stale",
conversationId: "cnv_issue195_stale",
threadId: staleThreadId
}
},
updatedAt: "2026-06-01T00:00:00.000Z"
});
const update = await patchJson(port, `/v1/workbench/workspace/${workspace.body.workspace.workspaceId}`, {
expectedRevision: 1,
selectedConversationId: "cnv_issue195_stale",
selectedAgentSessionId: "ses_issue195_stale",
activeTraceId: "trc_issue195_thread_resume_failed",
providerProfile: "deepseek",
sessionStatus: "running",
updatedByClient: "test-suite"
}, aliceLogin.cookie);
assert.equal(update.status, 200);
assert.equal(update.body.workspace.selectedConversationId, "cnv_issue195_stale");
assert.equal(update.body.workspace.activeTraceId, "trc_issue195_thread_resume_failed");
const restored = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", aliceLogin.cookie);
assert.equal(restored.status, 200);
assert.equal(restored.body.workspace.activeTraceId, null);
assert.equal(restored.body.workspace.selectedConversationId, null);
assert.equal(restored.body.workspace.selectedAgentSessionId, null);
assert.equal(restored.body.workspace.selectedConversation, null);
assert.equal(restored.body.workspace.workspace.sessionStatus, "failed");
assert.equal(restored.body.workspace.workspace.lastTraceId, "trc_issue195_thread_resume_failed");
assert.equal(restored.body.workspace.workspace.staleContinuationCleared, true);
assert.equal(restored.body.workspace.workspace.staleContinuationReason, "thread-resume-failed");
assert.equal(restored.body.workspace.workspace.staleConversationId, "cnv_issue195_stale");
assert.equal(restored.body.workspace.workspace.staleAgentSessionId, "ses_issue195_stale");
assert.equal(restored.body.workspace.workspace.staleThreadId, staleThreadId);
assert.equal(restored.body.workspace.workspace.recoveryAction, "new-session-on-next-turn");
const failedConversation = await getJson(port, "/v1/agent/conversations/cnv_issue195_stale", aliceLogin.cookie);
assert.equal(failedConversation.status, 200);
assert.equal(failedConversation.body.conversation.status, "failed");
assert.equal(failedConversation.body.conversation.threadId, staleThreadId);
const next = await postJson(port, "/v1/agent/chat", {
message: "new turn after stale continuation was cleared",
traceId: "trc_issue195_after_stale",
workspaceId: workspace.body.workspace.workspaceId,
expectedWorkspaceRevision: restored.body.workspace.revision,
shortConnection: true
}, aliceLogin.cookie, { prefer: "respond-async", "x-trace-id": "trc_issue195_after_stale" });
assert.equal(next.status, 202);
assert.equal(next.body.status, "running");
assert.equal(next.body.traceId, "trc_issue195_after_stale");
await waitForCondition(() => createRunInputs.length === 1);
assert.equal(createRunInputs.length, 1);
assert.notEqual(createRunInputs[0].sessionRef?.threadId, staleThreadId);
assert.equal(createRunInputs[0].sessionRef?.threadId, undefined);
assert.equal(createRunInputs[0].sessionRef?.conversationId, undefined);
const afterNext = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", aliceLogin.cookie);
assert.equal(afterNext.body.workspace.activeTraceId, "trc_issue195_after_stale");
assert.ok(agentRunCalls.some((call) => call.path === "/api/v1/runs/run_workspace_thread_resume_failed/commands/cmd_workspace_thread_resume_failed/result"));
} 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("workbench workspace status repairs terminal selected conversation after active trace was cleared", async () => {
const agentRunCalls = [];
const agentSessions = new Map();
@@ -1795,6 +1351,96 @@ test("cloud api exposes terminal conversation status when stored session status
}
});
test("cloud api repairs persisted final response fallback from terminal result", async () => {
const codeAgentChatResults = new Map();
const server = createCloudApiServer({
env: {
HWLAB_ACCESS_CONTROL_REQUIRED: "1",
HWLAB_BOOTSTRAP_ADMIN_USERNAME: "admin",
HWLAB_BOOTSTRAP_ADMIN_PASSWORD: "admin-pass",
HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01",
AGENTRUN_MGR_URL: "http://127.0.0.1:9",
HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1"
},
codeAgentChatResults,
now: () => "2026-06-04T09:45:00.000Z"
});
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const adminLogin = await postJson(port, "/auth/login", { username: "admin", password: "admin-pass" });
const aliceCreate = await postJson(port, "/v1/admin/users", { username: "alice-issue834-fallback", password: "alice-pass" }, adminLogin.cookie);
assert.equal(aliceCreate.status, 201);
const aliceLogin = await postJson(port, "/auth/login", { username: "alice-issue834-fallback", password: "alice-pass" });
const fallbackText = "Code Agent 仍在处理，可以继续 steer 或等待 trace 完成。";
const finalText = "**直接答：能，但只到一半。**\n\n功能已经过了，剩余风险另开 issue 跟踪。";
codeAgentChatResults.set("trc_issue834_final_response", {
status: "completed",
traceId: "trc_issue834_final_response",
conversationId: "cnv_issue834_fallback",
sessionId: "ses_issue834_fallback",
threadId: "thread-issue-834-fallback",
ownerUserId: aliceCreate.body.user.id,
reply: { role: "assistant", content: finalText, messageId: "msg_issue834_real_final" },
agentRun: { terminalStatus: "completed" },
session: {
sessionId: "ses_issue834_fallback",
conversationId: "cnv_issue834_fallback",
threadId: "thread-issue-834-fallback",
status: "idle"
}
});
const stored = await putJson(port, "/v1/agent/conversations/cnv_issue834_fallback", {
projectId: "prj_device_pod_workbench",
sessionId: "ses_issue834_fallback",
threadId: "thread-issue-834-fallback",
sessionStatus: "completed",
lastTraceId: "trc_issue834_final_response",
messages: [
{ id: "msg_issue834_question", role: "user", title: "用户", text: "功能是过了还是没过？", status: "sent", conversationId: "cnv_issue834_fallback", sessionId: "ses_issue834_fallback", threadId: "thread-issue-834-fallback" },
{ id: "msg_issue834_fallback", role: "agent", title: "Agent", text: fallbackText, status: "completed", traceId: "trc_issue834_final_response", conversationId: "cnv_issue834_fallback", sessionId: "ses_issue834_fallback", threadId: "thread-issue-834-fallback" }
]
}, aliceLogin.cookie);
assert.equal(stored.status, 200);
assert.equal(stored.body.conversation.messages[1].text, fallbackText);
assert.equal(stored.body.conversation.status, "completed");
const list = await getJson(port, "/v1/agent/conversations?projectId=prj_device_pod_workbench", aliceLogin.cookie);
assert.equal(list.status, 200);
const listed = list.body.conversations.find((conversation) => conversation.conversationId === "cnv_issue834_fallback");
assert.ok(listed, "expected fallback conversation to be listed");
assert.equal(listed.status, "idle");
assert.equal(listed.session.status, "idle");
assert.equal(listed.lastTraceId, "trc_issue834_final_response");
assert.equal(listed.messages[1].text, finalText);
assert.equal(listed.messages[1].status, "idle");
const direct = await getJson(port, "/v1/agent/conversations/cnv_issue834_fallback", aliceLogin.cookie);
assert.equal(direct.status, 200);
assert.equal(direct.body.conversation.status, "idle");
assert.equal(direct.body.conversation.lastTraceId, "trc_issue834_final_response");
assert.equal(direct.body.conversation.messages[1].text, finalText);
assert.equal(direct.body.conversation.messages[1].status, "idle");
const workspace = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", aliceLogin.cookie);
assert.equal(workspace.status, 200);
const selected = await postJson(port, `/v1/workbench/workspace/${workspace.body.workspace.workspaceId}/select-conversation`, {
projectId: "prj_device_pod_workbench",
conversationId: "cnv_issue834_fallback",
sessionId: "ses_issue834_fallback",
updatedByClient: "test-suite"
}, aliceLogin.cookie);
assert.equal(selected.status, 200);
assert.equal(selected.body.workspace.selectedConversation.messages[1].text, finalText);
} finally {
await new Promise((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
});
}
});
test("access controller restores AgentRun mapping by Code Agent traceId", async () => {
const accessController = createAccessController({ now: () => "2026-06-01T00:00:00.000Z" });
await accessController.recordAgentSessionOwner({
internal/cloud/access-control.ts
+135 -23
View File
@@ -924,7 +924,11 @@ class AccessController {
includeArchived: false,
now: this.now()
}) ?? [];
const conversations = conversationsFromAgentSessions(sessions);
const conversations = await this.repairVisibleConversationsIfNeeded(
auth.actor,
conversationsFromAgentSessions(sessions),
projectId
);
return sendJson(response, 200, {
ok: true,
status: "succeeded",
@@ -941,6 +945,8 @@ class AccessController {
await this.ensureBootstrap();
const auth = await this.authenticate(request, { required: true });
if (!auth.ok) return sendJson(response, auth.status, auth);
const url = new URL(request.url || "/", "http://localhost");
const projectId = textOr(url.searchParams.get("projectId"), "");
if (!safeConversationIdLocal(conversationId)) {
return sendJson(response, 400, errorPayload("invalid_conversation_id", "conversationId must start with cnv_", 400));
}
@@ -953,8 +959,9 @@ class AccessController {
includeArchived: true,
now: this.now()
}) ?? [];
const conversation = conversationsFromAgentSessions(sessions).find((item) => item.conversationId === conversationId) ?? null;
let conversation = conversationsFromAgentSessions(sessions).find((item) => item.conversationId === conversationId) ?? null;
if (!conversation) return sendJson(response, 404, errorPayload("agent_conversation_not_found", "Agent conversation is not visible to the current actor", 404));
conversation = await this.repairVisibleConversationIfNeeded(auth.actor, conversation, projectId) ?? conversation;
return sendJson(response, 200, { ok: true, status: "found", contractVersion: "agent-conversations-v1", actor: publicActor(auth.actor), conversation });
}
@@ -1209,24 +1216,9 @@ class AccessController {
const repairTraceId = activeTraceId ? "" : await this.terminalWorkbenchRepairTraceId(workspace, actor, workspaceJson);
const traceId = activeTraceId || repairTraceId;
if (!traceId) return workspace;
const options = {
env: codeAgentEnv,
fetchImpl: this.fetchImpl,
accessController: this,
codeAgentChatResults: this.codeAgentChatResults,
traceStore: this.traceStore ?? defaultCodeAgentTraceStore
};
try {
const cached = this.codeAgentChatResults?.get?.(traceId) ?? null;
const persisted = cached ? null : await loadPersistedAgentRunResult(traceId, options);
const current = cached ?? persisted ?? null;
if (!current || !canActorReadWorkspaceTrace(current, actor)) return workspace;
const synced = current?.agentRun?.runId && current.status === "running"
? await syncAgentRunChatResult({ traceId, currentResult: current, options, traceStore: options.traceStore })
: { result: current };
const result = synced.result ?? current;
if (!isTerminalCodeAgentResult(result)) return workspace;
const result = await this.terminalCodeAgentResultForActor(traceId, actor);
if (!result) return workspace;
const now = this.now();
const staleContinuation = isThreadResumeFailedResult(result);
@@ -1301,7 +1293,7 @@ class AccessController {
try {
const traceId = safeTraceIdLocal(workspaceJson?.lastTraceId);
if (!traceId || !safeConversationIdLocal(workspace?.selectedConversationId)) return "";
const conversation = await this.visibleConversationForActor(actor, workspace.selectedConversationId, workspace.projectId);
const conversation = await this.visibleConversationForActor(actor, workspace.selectedConversationId, workspace.projectId, { skipTerminalRepair: true });
return conversationNeedsTerminalRepair(conversation, traceId) ? traceId : "";
} catch {
return "";
@@ -1320,7 +1312,7 @@ class AccessController {
async syncTerminalWorkbenchConversation({ workspace, actor, result, activeTraceId, selectedConversationId, selectedAgentSessionId, now }) {
if (!safeConversationIdLocal(selectedConversationId) || !safeAgentSessionId(selectedAgentSessionId)) return null;
const existing = await this.visibleConversationForActor(actor, selectedConversationId, workspace.projectId);
const existing = await this.visibleConversationForActor(actor, selectedConversationId, workspace.projectId, { skipTerminalRepair: true });
const sessionStatus = terminalWorkbenchSessionStatus(result);
const threadId = textOr(result.threadId ?? result.session?.threadId ?? result.sessionReuse?.threadId ?? result.providerTrace?.threadId, existing?.threadId ?? null);
const messages = mergeTerminalConversationMessages(existing?.messages, result, {
@@ -1358,6 +1350,95 @@ class AccessController {
});
}
async terminalCodeAgentResultForActor(traceId, actor) {
const safeTraceId = safeTraceIdLocal(traceId);
const codeAgentEnv = this.codeAgentEnv ?? this.env;
if (!safeTraceId || !codeAgentAgentRunAdapterEnabled(codeAgentEnv)) return null;
const options = {
env: codeAgentEnv,
fetchImpl: this.fetchImpl,
accessController: this,
codeAgentChatResults: this.codeAgentChatResults,
traceStore: this.traceStore ?? defaultCodeAgentTraceStore
};
try {
const cached = this.codeAgentChatResults?.get?.(safeTraceId) ?? null;
const persisted = cached ? null : await loadPersistedAgentRunResult(safeTraceId, options);
const current = cached ?? persisted ?? null;
if (!current || !canActorReadWorkspaceTrace(current, actor)) return null;
const synced = current?.agentRun?.runId && current.status === "running"
? await syncAgentRunChatResult({ traceId: safeTraceId, currentResult: current, options, traceStore: options.traceStore })
: { result: current };
const result = synced.result ?? current;
return isTerminalCodeAgentResult(result) ? result : null;
} catch {
return null;
}
}
async repairVisibleConversationsIfNeeded(actor, conversations = [], projectId = "") {
if (!Array.isArray(conversations) || conversations.length === 0) return [];
const repaired = [];
for (const conversation of conversations) {
repaired.push(await this.repairVisibleConversationIfNeeded(actor, conversation, projectId) ?? conversation);
}
return repaired;
}
async repairVisibleConversationIfNeeded(actor, conversation, projectId = "") {
if (textOr(conversation?.status, "").toLowerCase() === "archived") return conversation;
const traceId = terminalConversationFallbackRepairTraceId(conversation);
if (!traceId) return conversation;
const result = await this.terminalCodeAgentResultForActor(traceId, actor);
if (!result) return conversation;
const sessionId = safeAgentSessionId(
result.sessionId
?? result.session?.sessionId
?? result.sessionReuse?.sessionId
?? conversation.sessionId
?? conversation.session?.sessionId
);
const conversationId = safeConversationIdLocal(result.conversationId) ? result.conversationId : conversation.conversationId;
if (!safeConversationIdLocal(conversationId) || !sessionId) return conversation;
const now = this.now();
const threadId = textOr(result.threadId ?? result.session?.threadId ?? result.sessionReuse?.threadId ?? result.providerTrace?.threadId, conversation.threadId ?? null);
const messages = mergeTerminalConversationMessages(conversation.messages, result, {
traceId,
conversationId,
sessionId,
threadId,
now,
existingMessageCount: conversation.messageCount ?? conversation.snapshot?.messageCount ?? null,
firstUserMessagePreview: conversation.firstUserMessagePreview ?? conversation.snapshot?.firstUserMessagePreview ?? null
});
const mergedMessages = Array.isArray(messages) ? messages : messages?.messages ?? [];
const mergedCount = Array.isArray(messages) ? null : messages?.messageCount ?? null;
const mergedPreview = Array.isArray(messages) ? null : messages?.firstUserMessagePreview ?? null;
const sessionStatus = terminalWorkbenchSessionStatus(result);
const repaired = await this.recordAgentSessionOwner({
ownerUserId: actor.id,
sessionId,
projectId: textOr(conversation.projectId ?? projectId, "prj_device_pod_workbench"),
agentId: conversation.agentId ?? "hwlab-code-agent",
status: sessionStatus,
conversationId,
threadId,
traceId,
session: normalizeConversationSnapshot({
...(conversation.snapshot ?? {}),
source: "conversation-terminal-repair",
sessionStatus,
updatedAt: now,
lastTraceId: traceId,
messages: mergedMessages,
messageCount: mergedCount,
firstUserMessagePreview: mergedPreview
}, actor),
now
});
return conversationsFromAgentSessions(repaired ? [repaired] : []).find((item) => item.conversationId === conversationId) ?? conversation;
}
async visibleConversationForActor(actor, conversationId, projectId = "", options = {}) {
if (!safeConversationIdLocal(conversationId)) return null;
const sessions = await this.store.listAgentSessionsForUser?.({
@@ -1370,7 +1451,9 @@ class AccessController {
includeArchived: options.includeArchived === true,
now: this.now()
}) ?? [];
return conversationsFromAgentSessions(sessions).find((item) => item.conversationId === conversationId) ?? null;
const conversation = conversationsFromAgentSessions(sessions).find((item) => item.conversationId === conversationId) ?? null;
if (!conversation || options.skipTerminalRepair === true) return conversation;
return await this.repairVisibleConversationIfNeeded(actor, conversation, projectId) ?? conversation;
}
async nextVisibleConversationForActor(actor, projectId = "", excludedConversationId = "") {
@@ -2460,7 +2543,36 @@ function conversationNeedsTerminalRepair(conversation, traceId) {
const traceMessage = messages.find((message) => message?.traceId === traceId && message?.role === "agent") ?? null;
if (!traceMessage) return false;
const messageStatus = textOr(traceMessage.status, "").toLowerCase();
return ["", "running", "busy", "pending", "active"].includes(messageStatus);
return ["", "running", "busy", "pending", "active"].includes(messageStatus)
|| isLegacyFinalResponseFallbackText(traceMessage.text);
}
function terminalConversationRepairTraceId(conversation) {
if (!conversation) return "";
const lastTraceId = safeTraceIdLocal(conversation.lastTraceId ?? conversation.session?.lastTraceId ?? conversation.snapshot?.lastTraceId);
if (lastTraceId && conversationNeedsTerminalRepair(conversation, lastTraceId)) return lastTraceId;
const messages = Array.isArray(conversation.messages) ? conversation.messages : [];
for (let index = messages.length - 1; index >= 0; index -= 1) {
const message = messages[index];
const traceId = safeTraceIdLocal(message?.traceId);
if (message?.role === "agent" && traceId && conversationNeedsTerminalRepair(conversation, traceId)) return traceId;
}
return "";
}
function terminalConversationFallbackRepairTraceId(conversation) {
if (!conversation) return "";
const messages = Array.isArray(conversation.messages) ? conversation.messages : [];
const lastTraceId = safeTraceIdLocal(conversation.lastTraceId ?? conversation.session?.lastTraceId ?? conversation.snapshot?.lastTraceId);
const lastTraceMessage = messages.find((message) => message?.traceId === lastTraceId && message?.role === "agent") ?? null;
if (lastTraceId && isLegacyFinalResponseFallbackText(lastTraceMessage?.text)) return lastTraceId;
for (let index = messages.length - 1; index >= 0; index -= 1) {
const message = messages[index];
const traceId = safeTraceIdLocal(message?.traceId);
if (message?.role === "agent" && traceId && isLegacyFinalResponseFallbackText(message.text)) return traceId;
}
return "";
}
function isLegacyFinalResponseFallbackText(value) {
return textOr(value, "").trim() === "Code Agent 仍在处理，可以继续 steer 或等待 trace 完成。";
}
function mergeTerminalConversationMessages(existingMessages = [], result = {}, context = {}) {
const messages = Array.isArray(existingMessages) ? existingMessages.map(redactConversationMessage).filter(Boolean) : [];