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

Merge pull request #719 from pikasTech/fix-v02-code-agent-concurrency-717

Browse Source 
修复 Code Agent 默认 workspace 并发和 stale continuation 摩擦
This commit is contained in:
Lyon
2026-06-02 21:57:38 +08:00
committed by GitHub
parent e469ea9afd 9f287fa0d9
commit c1350bb7ac
7 changed files with 51 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/reference/spec-v02-deepseek-proxy.md
+1
View File
@@ -7,6 +7,7 @@
- 为 `hwlab-cloud-api` 提供 OpenAI Responses 兼容入口 `/v1/responses`，内部转到 Moon Bridge，再转 DeepSeek upstream。
- 只服务 Code Agent provider profile，不向浏览器或普通用户直接暴露。
- 读取 `hwlab-v02-code-agent-provider/openai-api-key` 作为 `DEEPSEEK_API_KEY` 渲染 Moon Bridge config；Secret 更新后必须滚动本 Deployment。
- DeepSeek 余额不足、外部额度耗尽或该 provider profile 明确不可用时，Code Agent 非 DeepSeek 专项验收可以临时切到 `minimax-m3` 继续闭环；issue/验收评论必须写明实际 `providerProfile=minimax-m3`，不得把它记为 DeepSeek 通道通过。
## 内部架构
docs/reference/spec-v02-hwlab-cli.md
+1 -1
View File
@@ -31,7 +31,7 @@
- `client agent send --from-trace` 只能用 inspect 恢复 `conversationId/sessionId/threadId/retryOf` 和提交本轮原始消息；不得把 inspect 的 messages/facts 作为 `conversationContext`、`messages` 或 prompt 前缀提交。CLI 的 continuation 摘要只用于可见性，不是模型上下文。
- `client agent composer status|submit` 是 Cloud Web composer 的状态机等价入口。`status` 必须先恢复账号 workspace，再用与 Web 相同的 composer policy 输出 `locked`、`disabled`、`submitMode`、`route`、`targetTraceId`、conversation/session/thread 和 workspace revision；运行中 turn 必须显示 `locked=false`、`disabled=false`、`submitMode=steer`。`submit` 必须按该 policy 自动选择 `/v1/agent/chat` 或 `/v1/agent/chat/steer`，不得要求用户手动判断 URL 或绕过 Cloud Web 同源 path。
- Code Agent continuation 的 thread 字段只有 `threadId` 一个标准名称。CLI 读取 inspect、`--from-trace` 回放、手动 `--thread-id` 提交和输出摘要都必须以该字段为唯一 thread identity；服务端响应也应保持同一字段口径。
- `client agent send` 默认先恢复账号 workspace，从 selected conversation/session 和 workspace JSON 中读取标准 `conversationId/sessionId/threadId`，再向 `/v1/agent/chat` 发送这些 IDs、`workspaceId` 和 `expectedWorkspaceRevision`；服务端接受后 CLI 保存新的 workspace revision，终态轮询后再 PATCH workspace 清理终态 `activeTraceId`。默认 workspace 恢复只恢复 ID 和 revision，不恢复 messages/facts，不生成 `conversationContext`，也不得把历史文本拼入 prompt。只有显式 `--no-workspace` 才跳过这一默认恢复路径。
- `client agent send` 默认先恢复账号 workspace，从 selected conversation/session 和 workspace JSON 中读取标准 `conversationId/sessionId/threadId`，再向 `/v1/agent/chat` 发送这些 IDs、`workspaceId` 和 `expectedWorkspaceRevision`；服务端接受后 CLI 保存新的 workspace revision，终态轮询后再 PATCH workspace 清理终态 `activeTraceId`。默认 workspace 恢复只恢复 ID 和 revision，不恢复 messages/facts，不生成 `conversationContext`，也不得把历史文本拼入 prompt。显式传入新的 `--conversation-id` 且没有显式 `--session-id/--thread-id` 时，CLI 必须把它视为新独立 conversation，不得从默认 workspace 继承旧 session/thread，避免 stale continuation 误续接。只有显式 `--no-workspace` 才跳过这一默认恢复路径。
- `client agent steer <traceId>` 是运行中引导入口，必须调用 Cloud Web 同源 `POST /v1/agent/chat/steer`，把 steer 文本装配成 AgentRun `type=steer` command 作用到目标 trace 的 active turn。CLI 不手动穿内部 URL；验收使用当前 runtime namespace/lane 自动解析的 `19666` Web 入口，并通过原 trace 的 result/trace 观察 steer 是否被 runner 接收和应用。
- `client agent trace <traceId> --render web` 必须调用 Cloud Web trace row 的同一纯转换路径，输出 `render="web"`、renderer 标识、source event count、rendered row count、默认压制的 noise event count 和 row 摘要。浏览器 trace 展示错乱时，必须先用该 CLI 入口确认 Web 渲染转换是否已经乱序、重复、缺 final response、吞掉关键 row 或只显示泛化 tool call，再继续修浏览器 DOM/CSS。
- AgentRun v0.1 短连接 runner 已要求支持同 run/runner 多轮 command。CLI 仍应把 Web 提交的 `conversationId/sessionId/threadId` 原样送到 Cloud Web API，用于验证 adapter 是否在 runner reuse window 有效时复用同一个 AgentRun `runId` / `jobName` 并创建新 `commandId`；每轮都新建 runner 或重新 bundle 不是通过状态，trace 中的原因说明只能用于定位。
docs/reference/spec-v02-hwlab-cloud-api.md
+2 -2
View File
@@ -36,7 +36,7 @@
| `GET /auth/session`、`POST /auth/login`、`POST /auth/logout` | v0.2 本地用户 session 入口。 |
| `GET /v1/auth/session`、`GET /v1/users/me`、`GET /v1/access/status`、`GET /v1/setup/status` | v0.2 用户/session/setup 的 REST 状态和兼容入口；响应不得暴露 password hash、session token 原文或 Secret 值。 |
| `POST /v1/admin/users`、`POST/PUT /v1/admin/device-pods`、`POST/DELETE /v1/admin/device-pod-grants...` | `admin` 管理用户、device pod profile 和 grant 的入口。 |
| `GET/PATCH /v1/workbench/workspace`、`POST /v1/workbench/workspace/{id}/reset`、`GET /events` | 账号级共享 workspace authority；所有读写按 ownerUserId 隔离，写入使用 revision 防冲突，active trace 用于阻止同账号并发 Code Agent turn。 |
| `GET/PATCH /v1/workbench/workspace`、`POST /v1/workbench/workspace/{id}/reset`、`GET /events` | 账号级共享 workspace authority；所有读写按 ownerUserId 隔离，写入使用 revision 观测冲突，active trace 只表示最近活动/当前选中 trace，不作为同账号 Code Agent 并发互斥锁。 |
| `GET /v1/m3/status`、`POST /v1/m3/io` | M3 只读/受控 IO 入口；写操作必须有明确 approval。 |
| `GET /v1/diagnostics/gate`、`GET /v1/live-builds` | 诊断和 live build inventory。 |
| `GET /v1/gateway/sessions`、`POST /v1/gateway/poll`、`POST /v1/gateway/result` | gateway 主动出站注册、取任务和回传结果。 |
@@ -77,5 +77,5 @@
| gateway outbound poll/result | 已实现 | 支持 gateway 主动轮询和 `hardware.invoke.shell` 分发。 |
| device-pod 正式权限/profile/job | 部分实现 | profile/grant/list/status/job 持久化在 cloud-api；用户态 probe GET 已收敛为只读 job；已提供内部 gateway dispatch route 供 `hwlab-device-pod` executor 下发到 device-host-cli，无在线 gateway/device-host-cli 时返回 blocker。 |
| v0.2 admin/user 权限模型 | 部分实现 | `/auth/*`、admin user/device-pod/grant API 和 Code Agent owner binding 已接入；生产 bootstrap 依赖 SecretRef。 |
| 账号共享 workspace authority | 已实现 | `account_workspaces` 持久化同账号 Web/CLI 共享 workspace，支持 revision 冲突、账号隔离、active trace 并发保护和 reset。 |
| 账号共享 workspace authority | 已实现 | `account_workspaces` 持久化同账号 Web/CLI 共享 workspace，支持 revision 冲突可见性、账号隔离、最近 active trace 记录和 reset；同一用户不同 session/run 允许并发，互斥只在 session/thread/run 层处理。 |
internal/cloud/access-control.test.ts
+24 -12
View File
@@ -62,7 +62,7 @@ test("Postgres workspace update query preserves parameter numbering", async () =
assert.equal(update.params[17], "user");
});
test("workbench workspace is account-scoped, persistent, and gates active agent turns", async () => {
test("workbench workspace is account-scoped, persistent, and permits independent Code Agent turns", async () => {
const server = createCloudApiServer({
env: {
HWLAB_ACCESS_CONTROL_REQUIRED: "1",
@@ -146,19 +146,30 @@ test("workbench workspace is account-scoped, persistent, and gates active agent
assert.equal(events.body.status, "changed");
assert.equal(events.body.latestRevision, 2);
const busy = await postJson(port, "/v1/agent/chat", {
message: "should be blocked by active workspace turn",
traceId: "trc_issue655_next",
conversationId: "cnv_issue655_shared",
sessionId: "ses_issue655_shared",
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_issue655_next" });
assert.equal(busy.status, 409);
assert.equal(busy.body.error.code, "workspace_agent_busy");
assert.equal(busy.body.activeTraceId, "trc_issue655_active");
assert.equal(busy.body.workspaceRevision, 2);
}, 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,
@@ -315,7 +326,8 @@ test("workbench workspace permits a new turn after AgentRun active trace reaches
assert.equal(next.status, 202);
assert.equal(next.body.status, "running");
assert.equal(next.body.traceId, "trc_issue655_after_done");
assert.ok(agentRunCalls.some((call) => call.path === "/api/v1/runs/run_workspace_done/commands/cmd_workspace_done/result"));
await waitForCondition(() => agentRunCalls.some((call) => call.method === "POST" && call.path === "/api/v1/runs"));
assert.equal(agentRunCalls.some((call) => call.path === "/api/v1/runs/run_workspace_done/commands/cmd_workspace_done/result"), false);
const restored = await getJson(port, "/v1/workbench/workspace?projectId=prj_device_pod_workbench", aliceLogin.cookie);
assert.equal(restored.body.workspace.activeTraceId, "trc_issue655_after_done");
internal/cloud/access-control.ts
+3
View File
@@ -1784,9 +1784,12 @@ function redactedWorkspaceJson(value = {}) {
selectedAgentSessionId: textOr(workspace.selectedAgentSessionId, ""),
selectedDevicePodId: textOr(workspace.selectedDevicePodId, ""),
activeTraceId: textOr(workspace.activeTraceId, ""),
previousActiveTraceId: textOr(workspace.previousActiveTraceId, ""),
providerProfile: textOr(workspace.providerProfile, ""),
sessionStatus: textOr(workspace.sessionStatus, ""),
lastTraceId: textOr(workspace.lastTraceId, ""),
workspaceRevisionConflict: workspace.workspaceRevisionConflict === true,
expectedWorkspaceRevision: Number.isFinite(Number(workspace.expectedWorkspaceRevision)) ? Number(workspace.expectedWorkspaceRevision) : undefined,
staleContinuationCleared: workspace.staleContinuationCleared === true,
staleContinuationReason: textOr(workspace.staleContinuationReason, ""),
staleConversationId: textOr(workspace.staleConversationId, ""),
internal/cloud/server-code-agent-http.ts
+13 -56
View File
@@ -1072,31 +1072,14 @@ async function claimWorkbenchWorkspaceTurn({ params = {}, options = {}, traceId,
return { blocked: true };
}
const expectedRevision = Number.parseInt(String(params.expectedWorkspaceRevision ?? params.workspaceRevision ?? ""), 10);
if (Number.isInteger(expectedRevision) && expectedRevision > 0 && workspace.revision !== expectedRevision) {
sendJson(response, 409, {
ok: false,
status: "revision_conflict",
error: { code: "workspace_revision_conflict", message: "Workbench workspace revision changed; restore workspace before submitting a new turn." },
workspaceId,
expectedRevision,
workspaceRevision: workspace.revision
});
return { blocked: true };
}
const workspaceRevisionConflict = Number.isInteger(expectedRevision) && expectedRevision > 0 && workspace.revision !== expectedRevision;
const requestedConversationId = safeConversationId(params.conversationId) || null;
const requestedSessionId = safeSessionId(params.sessionId) || null;
const nextSelectedConversationId = requestedConversationId || workspace.selectedConversationId;
const nextSelectedAgentSessionId = requestedSessionId || (requestedConversationId && requestedConversationId !== workspace.selectedConversationId ? null : workspace.selectedAgentSessionId);
const activeTraceId = safeTraceId(workspace.activeTraceId);
const activeResult = activeTraceId ? await resolveWorkbenchActiveTraceResult(activeTraceId, options) : null;
const activeTraceSnapshot = activeTraceId ? (options.traceStore ?? defaultCodeAgentTraceStore).snapshot(activeTraceId) : null;
const activeTraceOrphaned = Boolean(activeTraceId && !activeResult && isWorkbenchOrphanActiveTrace(activeTraceSnapshot));
const selectedConversationId = safeConversationId(workspace.selectedConversationId);
const shouldInspectSelectedConversation = Boolean(
activeTraceId && !activeResult && !activeTraceOrphaned && selectedConversationId && options.accessController?.visibleConversationForActor
);
const selectedConversation = shouldInspectSelectedConversation
? await options.accessController.visibleConversationForActor(options.actor, selectedConversationId, workspace.projectId)
: null;
const activeTraceBlocksTurn = Boolean(
activeTraceId && !activeTraceOrphaned && (activeResult?.status === "running" || (!activeResult && isActiveWorkbenchConversation(selectedConversation)))
);
const activeTraceOrphaned = Boolean(activeTraceId && isWorkbenchOrphanActiveTrace(activeTraceSnapshot));
if (activeTraceOrphaned) {
await store.updateWorkspace?.({
workspaceId,
@@ -1123,37 +1106,20 @@ async function claimWorkbenchWorkspaceTurn({ params = {}, options = {}, traceId,
now: new Date().toISOString()
});
}
if (activeTraceBlocksTurn) {
const conversationId = safeConversationId(params.conversationId) || selectedConversationId || activeResult?.conversationId || null;
const sessionId = safeSessionId(params.sessionId) || safeSessionId(workspace.selectedAgentSessionId) || activeResult?.sessionId || null;
const threadId = safeOpaqueId(params.threadId) || activeResult?.threadId || null;
sendJson(response, 409, {
ok: false,
status: "workspace_agent_busy",
error: { code: "workspace_agent_busy", message: "Workbench workspace already has an active Code Agent turn; follow the active trace instead of opening a new conversation." },
workspaceId,
workspaceRevision: workspace.revision,
activeTraceId,
conversationId,
sessionId,
threadId,
providerTrace: activeResult?.providerTrace ?? activeResult?.agentRun?.providerTrace ?? null,
resultUrl: `/v1/agent/chat/result/${encodeURIComponent(activeTraceId)}`,
traceUrl: `/v1/agent/chat/trace/${encodeURIComponent(activeTraceId)}`
});
return { blocked: true };
}
const updated = await store.updateWorkspace?.({
workspaceId,
ownerUserId: options.actor.id,
actorRole: options.actor.role,
selectedConversationId: safeConversationId(params.conversationId) || workspace.selectedConversationId,
selectedAgentSessionId: safeSessionId(params.sessionId) || workspace.selectedAgentSessionId,
selectedConversationId: nextSelectedConversationId,
selectedAgentSessionId: nextSelectedAgentSessionId,
activeTraceId: traceId,
providerProfile: textValue(params.providerProfile) || workspace.providerProfile,
patch: {
selectedConversationId: safeConversationId(params.conversationId) || workspace.selectedConversationId,
selectedAgentSessionId: safeSessionId(params.sessionId) || workspace.selectedAgentSessionId,
selectedConversationId: nextSelectedConversationId,
selectedAgentSessionId: nextSelectedAgentSessionId,
previousActiveTraceId: activeTraceId || null,
workspaceRevisionConflict,
expectedWorkspaceRevision: Number.isInteger(expectedRevision) && expectedRevision > 0 ? expectedRevision : null,
activeTraceId: traceId,
providerProfile: textValue(params.providerProfile) || workspace.providerProfile,
sessionStatus: "running",
@@ -1169,15 +1135,6 @@ async function claimWorkbenchWorkspaceTurn({ params = {}, options = {}, traceId,
return { workspace: updated ?? workspace };
}
function isActiveWorkbenchConversation(conversation = null) {
if (!conversation) return false;
const values = [conversation.status, conversation.session?.status, conversation.snapshot?.sessionStatus];
for (const message of Array.isArray(conversation.messages) ? conversation.messages.slice(-5) : []) {
values.push(message?.status, message?.runnerTrace?.sessionStatus, message?.sessionStatus);
}
return values.some((value) => ["running", "busy", "pending", "queued"].includes(textValue(value).toLowerCase()));
}
function isWorkbenchOrphanActiveTrace(snapshot) {
if (!snapshot || typeof snapshot !== "object") return false;
const last = snapshot.lastEvent && typeof snapshot.lastEvent === "object" ? snapshot.lastEvent : null;
tools/src/hwlab-cli-lib.ts
+7 -3
View File
@@ -1038,9 +1038,13 @@ async function agentSend(context: any) {
const replay = await resolveAgentReplayContext(context);
const traceId = text(parsed.traceId) || makeId("trc");
const workspaceState = parsed.noWorkspace === true ? null : await restoreWorkbenchWorkspace(context, { quiet: true });
const conversationId = text(parsed.conversationId) || replay.conversationId || text(workspaceState?.selectedConversationId) || makeId("cnv");
const sessionId = text(parsed.sessionId) || replay.sessionId || text(workspaceState?.selectedAgentSessionId);
const threadId = text(parsed.threadId) || replay.threadId || text(workspaceState?.threadId);
const explicitConversationId = text(parsed.conversationId);
const replayConversationId = replay.conversationId;
const workspaceConversationId = text(workspaceState?.selectedConversationId);
const conversationId = explicitConversationId || replayConversationId || workspaceConversationId || makeId("cnv");
const explicitNewConversation = Boolean(explicitConversationId && explicitConversationId !== replayConversationId && explicitConversationId !== workspaceConversationId);
const sessionId = text(parsed.sessionId) || replay.sessionId || (explicitNewConversation ? "" : text(workspaceState?.selectedAgentSessionId));
const threadId = text(parsed.threadId) || replay.threadId || (explicitNewConversation ? "" : text(workspaceState?.threadId));
const retryOf = text(parsed.retryOf) || replay.retryOf;
const continuation = agentContinuationSummary({
conversationId,