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

fix: sync terminal workbench traces

Browse Source
This commit is contained in:
Codex
2026-06-02 01:55:19 +08:00
parent 84a49eb0db
commit 577622d0ab
3 changed files with 227 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/access-control.test.ts
+125
View File
@@ -324,6 +324,131 @@ test("workbench workspace permits a new turn after AgentRun active trace reaches
}
});
test("workbench workspace status clears completed AgentRun active trace on read", async () => {
const agentRunCalls = [];
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_status/events") {
return send({ items: [
{ id: "evt_status_done", runId: "run_workspace_status", seq: 1, type: "terminal_status", payload: { commandId: "cmd_workspace_status", terminalStatus: "completed" }, createdAt: "2026-06-01T00:00:01.000Z" }
] });
}
if (request.method === "GET" && url.pathname === "/api/v1/runs/run_workspace_status/commands/cmd_workspace_status/result") {
return send({
runId: "run_workspace_status",
commandId: "cmd_workspace_status",
status: "completed",
runStatus: "completed",
commandState: "completed",
terminalStatus: "completed",
completed: true,
reply: "previous turn completed",
lastSeq: 1,
sessionRef: { sessionId: "ses_issue664_status", conversationId: "cnv_issue664_status", threadId: "thread-issue-664" }
});
}
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_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-status", password: "alice-pass" }, adminLogin.cookie);
assert.equal(alice.status, 201);
const aliceLogin = await postJson(port, "/auth/login", { username: "alice-ws-status", 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_issue664_status", {
projectId: "prj_device_pod_workbench",
sessionId: "ses_issue664_status",
threadId: "thread-issue-664",
sessionStatus: "active",
lastTraceId: "trc_issue664_previous",
messages: [{ role: "agent", text: "previous turn", traceId: "trc_issue664_previous" }]
}, aliceLogin.cookie);
assert.equal(conversation.status, 200);
const previousPayload = {
messageId: "msg_workspace_status_done",
agentRun: {
runId: "run_workspace_status",
commandId: "cmd_workspace_status",
backendProfile: "deepseek",
managerUrl: `http://127.0.0.1:${agentRunPort}`,
lastSeq: 0,
sessionId: "ses_issue664_status",
conversationId: "cnv_issue664_status",
threadId: "thread-issue-664"
}
};
agentSessions.set("trc_issue664_status_done", {
id: "ses_issue664_status",
ownerUserId: alice.body.user.id,
conversationId: "cnv_issue664_status",
threadId: "thread-issue-664",
status: "active",
session: previousPayload,
updatedAt: "2026-06-01T00:00:00.000Z"
});
const update = await patchJson(port, `/v1/workbench/workspace/${workspace.body.workspace.workspaceId}`, {
expectedRevision: 1,
selectedConversationId: "cnv_issue664_status",
selectedAgentSessionId: "ses_issue664_status",
activeTraceId: "trc_issue664_status_done",
providerProfile: "deepseek",
sessionStatus: "running",
updatedByClient: "test-suite"
}, aliceLogin.cookie);
assert.equal(update.status, 200);
assert.equal(update.body.workspace.activeTraceId, "trc_issue664_status_done");
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.workspace.lastTraceId, "trc_issue664_status_done");
assert.equal(restored.body.workspace.workspace.sessionStatus, "idle");
assert.equal(restored.body.workspace.revision, 3);
assert.ok(agentRunCalls.some((call) => call.path === "/api/v1/runs/run_workspace_status/commands/cmd_workspace_status/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("cloud api access control grants visible device pods and requires device-pod executor", async () => {
let directGatewayDispatches = 0;
const gatewayRegistry = {
internal/cloud/access-control.ts
+98 -3
View File
@@ -1,6 +1,12 @@
import { createHash, randomBytes, randomUUID } from "node:crypto";
import { CLOUD_API_SERVICE_ID } from "../audit/index.mjs";
import {
codeAgentAgentRunAdapterEnabled,
loadPersistedAgentRunResult,
syncAgentRunChatResult
} from "./code-agent-agentrun-adapter.ts";
import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts";
import { getHeader, readBody, sendJson, truthyFlag } from "./server-http-utils.ts";
const SESSION_COOKIE = "hwlab_session";
@@ -169,11 +175,14 @@ function accessStoreForRuntime(runtimeStore, options = {}) {
}
class AccessController {
constructor({ store, env = process.env, gatewayRegistry = null, fetchImpl = fetch, devicePodExecutorUrl = env.HWLAB_DEVICE_POD_URL, devicePodExecutorTimeoutMs = 1200, now = () => new Date().toISOString(), required = truthyFlag(env.HWLAB_ACCESS_CONTROL_REQUIRED) } = {}) {
constructor({ store, env = process.env, gatewayRegistry = null, fetchImpl = fetch, traceStore = null, codeAgentChatResults = null, devicePodExecutorUrl = env.HWLAB_DEVICE_POD_URL, devicePodExecutorTimeoutMs = 1200, now = () => new Date().toISOString(), required = truthyFlag(env.HWLAB_ACCESS_CONTROL_REQUIRED) } = {}) {
this.store = store;
this.env = env;
this.gatewayRegistry = gatewayRegistry;
this.fetchImpl = fetchImpl;
this.traceStore = traceStore;
this.codeAgentChatResults = codeAgentChatResults;
this.codeAgentEnv = env;
this.devicePodExecutorUrl = normalizeBaseUrl(devicePodExecutorUrl);
this.devicePodInternalToken = textOr(env.HWLAB_DEVICE_POD_INTERNAL_TOKEN, "");
this.devicePodExecutorTimeoutMs = Number.parseInt(String(devicePodExecutorTimeoutMs), 10) || 1200;
@@ -183,6 +192,13 @@ class AccessController {
this.codeAgentDevicePodSession = null;
}
configureCodeAgentWorkspaceContext({ env = null, fetchImpl = null, traceStore = null, codeAgentChatResults = null } = {}) {
if (env) this.codeAgentEnv = env;
if (fetchImpl) this.fetchImpl = fetchImpl;
if (traceStore) this.traceStore = traceStore;
if (codeAgentChatResults) this.codeAgentChatResults = codeAgentChatResults;
}
async handleAuthRoute(request, response, url) {
try {
await this.ensureBootstrap();
@@ -670,11 +686,12 @@ class AccessController {
const auth = await this.authenticate(request, { required: true });
if (!auth.ok) return sendJson(response, auth.status, auth);
const projectId = textOr(url.searchParams.get("projectId"), "prj_device_pod_workbench");
const workspace = await this.store.getOrCreateDefaultWorkspace?.({
let workspace = await this.store.getOrCreateDefaultWorkspace?.({
ownerUserId: auth.actor.id,
projectId,
now: this.now()
});
workspace = await this.syncTerminalWorkbenchWorkspace(workspace, auth.actor);
return sendJson(response, 200, await this.workbenchWorkspacePayload(workspace, auth.actor));
}
@@ -801,8 +818,9 @@ class AccessController {
await this.ensureBootstrap();
const auth = await this.authenticate(request, { required: true });
if (!auth.ok) return sendJson(response, auth.status, auth);
const workspace = await this.store.getWorkspaceForUser?.({ workspaceId, ownerUserId: auth.actor.id, actorRole: auth.actor.role });
let workspace = await this.store.getWorkspaceForUser?.({ workspaceId, ownerUserId: auth.actor.id, actorRole: auth.actor.role });
if (!workspace) return sendJson(response, 404, errorPayload("workbench_workspace_not_found", "Workbench workspace is not visible to the current actor", 404));
workspace = await this.syncTerminalWorkbenchWorkspace(workspace, auth.actor);
const afterRevision = Number.parseInt(String(url.searchParams.get("afterRevision") ?? ""), 10) || 0;
const changed = workspace.revision > afterRevision;
return sendJson(response, 200, {
@@ -826,6 +844,65 @@ class AccessController {
};
}
async syncTerminalWorkbenchWorkspace(workspace, actor) {
const activeTraceId = safeTraceIdLocal(workspace?.activeTraceId);
const codeAgentEnv = this.codeAgentEnv ?? this.env;
if (!workspace || !activeTraceId || !codeAgentAgentRunAdapterEnabled(codeAgentEnv)) return workspace;
const options = {
env: codeAgentEnv,
fetchImpl: this.fetchImpl,
accessController: this,
codeAgentChatResults: this.codeAgentChatResults,
traceStore: this.traceStore ?? defaultCodeAgentTraceStore
};
try {
const cached = this.codeAgentChatResults?.get?.(activeTraceId) ?? null;
const persisted = cached ? null : await loadPersistedAgentRunResult(activeTraceId, options);
const current = cached ?? persisted ?? null;
if (!current || !canActorReadWorkspaceTrace(current, actor)) return workspace;
const synced = current?.agentRun?.runId && current.status === "running"
? await syncAgentRunChatResult({ traceId: activeTraceId, currentResult: current, options, traceStore: options.traceStore })
: { result: current };
const result = synced.result ?? current;
if (!isTerminalCodeAgentResult(result)) return workspace;
const now = this.now();
const workspaceJson = normalizeObject(workspace.workspace);
const selectedConversationId = safeConversationIdLocal(result.conversationId) ? result.conversationId : workspace.selectedConversationId;
const selectedAgentSessionId = safeAgentSessionId(result.sessionId ?? result.session?.sessionId ?? result.sessionReuse?.sessionId) || workspace.selectedAgentSessionId;
const updated = await this.store.updateWorkspace?.({
workspaceId: workspace.id,
ownerUserId: workspace.ownerUserId,
actorRole: actor?.role ?? "user",
selectedConversationId,
selectedAgentSessionId,
selectedDevicePodId: workspace.selectedDevicePodId,
activeTraceId: null,
providerProfile: workspace.providerProfile,
patch: {
...workspaceJson,
selectedConversationId,
selectedAgentSessionId,
activeTraceId: null,
sessionStatus: terminalWorkbenchSessionStatus(result),
lastTraceId: activeTraceId,
syncedTraceStatus: result.status ?? result.agentRun?.terminalStatus ?? null,
updatedAt: now,
source: "workbench-status-sync",
valuesRedacted: true,
secretMaterialStored: false
},
updatedBySessionId: workspace.updatedBySessionId,
updatedByClient: "workbench-status-sync",
now
});
return updated ?? workspace;
} catch {
return workspace;
}
}
async publicWorkbenchWorkspace(workspace, actor) {
const conversation = workspace?.selectedConversationId
? await this.visibleConversationForActor(actor, workspace.selectedConversationId, workspace.projectId)
@@ -1848,6 +1925,24 @@ function normalizeConversationSnapshot(body = {}, actor = null) {
valuesRedacted: true
};
}
function isTerminalCodeAgentResult(result = {}) {
const status = textOr(result.status, "").toLowerCase();
const terminalStatus = textOr(result.agentRun?.terminalStatus, "").toLowerCase();
return [status, terminalStatus].some((value) => ["completed", "failed", "blocked", "canceled", "cancelled", "timeout", "error"].includes(value));
}
function terminalWorkbenchSessionStatus(result = {}) {
const sessionStatus = textOr(result.session?.status ?? result.sessionSummary?.status ?? result.sessionLifecycleStatus ?? result.runnerTrace?.sessionStatus, "").toLowerCase();
if (sessionStatus && sessionStatus !== "running" && sessionStatus !== "busy" && sessionStatus !== "pending") return sessionStatus === "cancelled" ? "canceled" : sessionStatus;
const status = textOr(result.status ?? result.agentRun?.terminalStatus, "").toLowerCase();
if (status === "completed") return "idle";
if (status === "cancelled") return "canceled";
return status || "idle";
}
function canActorReadWorkspaceTrace(result = {}, actor = null) {
const ownerUserId = textOr(result.ownerUserId, "");
if (!ownerUserId || actor?.role === "admin") return true;
return ownerUserId === actor?.id;
}
function redactConversationMessage(message) {
if (!message || typeof message !== "object") return null;
const runnerTrace = message.runnerTrace && typeof message.runnerTrace === "object" ? message.runnerTrace : null;
internal/cloud/server.ts
+4 -1
View File
@@ -81,7 +81,10 @@ export function createCloudApiServer(options = {}) {
staleMs: parsePositiveInteger(env.HWLAB_GATEWAY_DEMO_STALE_MS, 30000),
dispatchTimeoutMs: parsePositiveInteger(env.HWLAB_GATEWAY_DISPATCH_TIMEOUT_MS, 120000)
});
const accessController = options.accessController || createAccessController({ ...options, env, runtimeStore, gatewayRegistry });
const accessController = options.accessController || createAccessController({ ...options, env, runtimeStore, gatewayRegistry, traceStore, codeAgentChatResults });
if (typeof accessController.configureCodeAgentWorkspaceContext === "function") {
accessController.configureCodeAgentWorkspaceContext({ env, fetchImpl: options.fetchImpl, traceStore, codeAgentChatResults });
}
return createServer(async (request, response) => {
try {
await routeRequest(request, response, { ...options, env, runtimeStore, gatewayRegistry, accessController, sessionRegistry, traceStore, codexStdioManager, codeAgentChatResults });