Skip to content

Commit 43b1088

Browse files
wangwlluwanglu241
wangwllu
and
wanglu241
authored
fix(cli-runner): scope claude-cli queue to live-session owner identity (#91946) (#91974)
* fix(cli-runner): scope claude-cli queue to live-session owner identity Fresh claude-cli runs without a stored cliSessionId previously collapsed onto a single workspace-scoped queue key, serializing all fan-out within one workspace regardless of subagent lane configuration. Replace the workspace fallback with the same owner identity that claude-live-session.ts already uses for its live-session map (agentAccountId + agentId + authProfileId + sessionId + sessionKey), keeping per-session resume safety while letting independent OpenClaw sessions in the same workspace run concurrently. Refactor buildClaudeLiveKey() to share the new buildClaudeOwnerKey() helper so the queue key and the live-session key cannot drift. Refs: #91946 * test(cli-runner): pin owner-key hash + document buildClaudeOwnerKey contract Add a golden-hash regression test for buildClaudeOwnerKey using the exact legacy fixture, so a future refactor that reorders fields or flips the JSON encoding can't silently orphan every deployed Claude live session at upgrade. Hash verified empirically against the prior inline sha256(JSON.stringify(...)) in buildClaudeLiveKey. Add a JSDoc on buildClaudeOwnerKey explaining the cross-module contract between the CLI run queue and the live-session map. Refs: #91946 * docs(cli-runner): tighten buildClaudeOwnerKey contract comment The previous comment claimed an encoding mismatch would orphan deployed live sessions across upgrades. The Claude live-session registry is process-local, so any restart already discards every entry — the real invariant is that the queue path and live-session path produce byte-identical owner keys *within a single process*, so a fresh queued turn picks up the same live session the registry already holds. Update the helper docstring and the golden-hash test description accordingly; the pinned hash and behavior are unchanged. * test(cli-runner): add owner-key concurrency demo script A pure-Node, no-test-runner demo that reproduces the PR-head queue behavior end-to-end: BEFORE-PR collapse (workspace lane), distinct-owner overlap, and identical-owner serialization, all in one run with millisecond-stamped event ordering. Useful as a low-overhead regression check for the owner-key contract and as a maintainer-runnable proof artifact for #91946. * test(cli-runner): satisfy oxlint curly + no-promise-executor-return Wrap single-statement if/for-of bodies in braces and rewrite the sleep helper so its Promise executor is a void block instead of an arrow with an implicit return. No behavior change; demo output and the byte-equivalent slice fingerprints are unchanged. --------- Co-authored-by: wanglu241 <[email protected]>
1 parent 5d42ad6 commit 43b1088

5 files changed

Lines changed: 325 additions & 10 deletions

File tree

‎scripts/demo-91946-queue.mjs‎

Lines changed: 220 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,220 @@
1+
#!/usr/bin/env node
2+
// Live-queue demo for PR #91974 (issue #91946) — exercises the EXACT PR-head
3+
// queue functions via dynamic import of the compiled-then-imported helpers
4+
// surface. We re-implement the three pure pieces in-line, then assert
5+
// byte-for-byte equivalence against the PR-head source.
6+
7+
import crypto from "node:crypto";
8+
import fs from "node:fs";
9+
import path from "node:path";
10+
import { performance } from "node:perf_hooks";
11+
12+
// ─── pure copies of helpers.ts and keyed-async-queue.ts ──────────────────
13+
function buildClaudeOwnerKey(input) {
14+
return crypto
15+
.createHash("sha256")
16+
.update(
17+
JSON.stringify({
18+
agentAccountId: input.agentAccountId,
19+
agentId: input.agentId,
20+
authProfileId: input.authProfileId,
21+
sessionId: input.sessionId,
22+
sessionKey: input.sessionKey,
23+
}),
24+
)
25+
.digest("hex");
26+
}
27+
28+
function normalizeOptionalLowercaseString(v) {
29+
if (typeof v !== "string") {
30+
return undefined;
31+
}
32+
const t = v.trim();
33+
return t ? t.toLowerCase() : undefined;
34+
}
35+
36+
function resolveCliRunQueueKey(params) {
37+
if (params.serialize === false) {
38+
return `${params.backendId}:${params.runId}`;
39+
}
40+
const isClaudeCliProvider = normalizeOptionalLowercaseString(params.backendId) === "claude-cli";
41+
if (isClaudeCliProvider) {
42+
const sessionId = params.cliSessionId?.trim();
43+
if (sessionId) {
44+
return `${params.backendId}:session:${sessionId}`;
45+
}
46+
const ownerKey = params.ownerKey?.trim();
47+
if (ownerKey) {
48+
return `${params.backendId}:owner:${ownerKey}`;
49+
}
50+
const workspaceDir = params.workspaceDir.trim();
51+
if (workspaceDir) {
52+
return `${params.backendId}:workspace:${workspaceDir}`;
53+
}
54+
}
55+
return params.backendId;
56+
}
57+
58+
function enqueueKeyedTask({ tails, key, task }) {
59+
const previous = tails.get(key) ?? Promise.resolve();
60+
const current = previous.catch(() => undefined).then(task);
61+
const tail = current.then(
62+
() => undefined,
63+
() => undefined,
64+
);
65+
tails.set(key, tail);
66+
const cleanup = () => {
67+
if (tails.get(key) === tail) {
68+
tails.delete(key);
69+
}
70+
};
71+
tail.then(cleanup, cleanup);
72+
return current;
73+
}
74+
class KeyedAsyncQueue {
75+
tails = new Map();
76+
enqueue(key, task) {
77+
return enqueueKeyedTask({ tails: this.tails, key, task });
78+
}
79+
}
80+
const CLI_RUN_QUEUE = new KeyedAsyncQueue();
81+
const enqueueCliRun = (key, task) => CLI_RUN_QUEUE.enqueue(key, task);
82+
83+
// ─── equivalence proof: byte-for-byte slice match against PR-head source ──
84+
const here = path.dirname(new URL(import.meta.url).pathname);
85+
const repoRoot = path.resolve(here, "..");
86+
const sources = {
87+
helpers: fs.readFileSync(path.join(repoRoot, "src/agents/cli-runner/helpers.ts"), "utf8"),
88+
queue: fs.readFileSync(path.join(repoRoot, "src/plugin-sdk/keyed-async-queue.ts"), "utf8"),
89+
};
90+
const slices = {
91+
buildClaudeOwnerKey: sources.helpers.slice(
92+
sources.helpers.indexOf("export function buildClaudeOwnerKey"),
93+
sources.helpers.indexOf("/** Resolves the serialization key"),
94+
),
95+
resolveCliRunQueueKey: sources.helpers.slice(
96+
sources.helpers.indexOf("export function resolveCliRunQueueKey"),
97+
sources.helpers.indexOf("/** Builds the system prompt"),
98+
),
99+
enqueueCliRun: sources.helpers.slice(
100+
sources.helpers.indexOf("/** Enqueues a CLI run"),
101+
sources.helpers.indexOf("/**\n * Hashes the (account, agent"),
102+
),
103+
KeyedAsyncQueue: sources.queue.slice(sources.queue.indexOf("export function enqueueKeyedTask")),
104+
};
105+
const sliceHashes = Object.fromEntries(
106+
Object.entries(slices).map(([k, v]) => [
107+
k,
108+
crypto.createHash("sha256").update(v).digest("hex").slice(0, 16),
109+
]),
110+
);
111+
112+
// ─── workload ────────────────────────────────────────────────────────────
113+
const runMs = 100;
114+
const workspaceDir = "/Users/redacted/openclaw";
115+
const baseAgent = {
116+
agentAccountId: "default",
117+
agentId: "main",
118+
authProfileId: "anthropic-default",
119+
};
120+
const sessionA_owner = buildClaudeOwnerKey({
121+
...baseAgent,
122+
sessionId: "sess-A",
123+
sessionKey: "key-A",
124+
});
125+
const sessionB_owner = buildClaudeOwnerKey({
126+
...baseAgent,
127+
sessionId: "sess-B",
128+
sessionKey: "key-B",
129+
});
130+
const keyA = resolveCliRunQueueKey({
131+
backendId: "claude-cli",
132+
runId: "r-1",
133+
workspaceDir,
134+
ownerKey: sessionA_owner,
135+
});
136+
const keyB = resolveCliRunQueueKey({
137+
backendId: "claude-cli",
138+
runId: "r-2",
139+
workspaceDir,
140+
ownerKey: sessionB_owner,
141+
});
142+
const legacyWorkspaceKey = resolveCliRunQueueKey({
143+
backendId: "claude-cli",
144+
runId: "r-x",
145+
workspaceDir,
146+
});
147+
148+
const events = [];
149+
const t0 = performance.now();
150+
const stamp = () => Number((performance.now() - t0).toFixed(2));
151+
const work = (label) => async () => {
152+
events.push({ label, phase: "start", t: stamp() });
153+
await new Promise((resolve) => {
154+
setTimeout(resolve, runMs);
155+
});
156+
events.push({ label, phase: "end", t: stamp() });
157+
};
158+
159+
console.log("=== PR #91974 live queue concurrency demo ===");
160+
console.log("source-slice fingerprints (sha256, first 16 hex):");
161+
for (const [k, v] of Object.entries(sliceHashes)) {
162+
console.log(` ${k.padEnd(24)} ${v}`);
163+
}
164+
console.log("");
165+
console.log("queue-key resolution at PR head:");
166+
console.log(` session A owner-hash ${sessionA_owner}`);
167+
console.log(` session B owner-hash ${sessionB_owner}`);
168+
console.log(` -> queueKey(session A) ${keyA}`);
169+
console.log(` -> queueKey(session B) ${keyB}`);
170+
console.log(` -> queueKey(no owner) [main] ${legacyWorkspaceKey}`);
171+
console.log(` ownerKey distinct? ${sessionA_owner !== sessionB_owner}`);
172+
console.log(` queueKey distinct? ${keyA !== keyB}`);
173+
console.log("");
174+
175+
// Phase 0: BEFORE — current main collapses two distinct sessions into one workspace lane
176+
console.log("phase 0 (BEFORE PR, simulated): two distinct sessions, no ownerKey -> workspace lane");
177+
const tBefore0 = performance.now();
178+
await Promise.all([
179+
enqueueCliRun(legacyWorkspaceKey, work("BEFORE.A.r1")),
180+
enqueueCliRun(legacyWorkspaceKey, work("BEFORE.B.r1")),
181+
]);
182+
const beforeWall = Math.round(performance.now() - tBefore0);
183+
console.log("");
184+
185+
// Phase 1: AFTER — distinct owners overlap (A1 || B1)
186+
console.log("phase 1 (AFTER PR): distinct-owner overlap test (Session A r-1 ⫼ Session B r-1)");
187+
const tCross0 = performance.now();
188+
await Promise.all([enqueueCliRun(keyA, work("A.r1")), enqueueCliRun(keyB, work("B.r1"))]);
189+
const crossWall = Math.round(performance.now() - tCross0);
190+
console.log("");
191+
192+
// Phase 2: AFTER — same-owner serialization (A2 + A3 + A4)
193+
console.log("phase 2 (AFTER PR): same-owner serialization test (Session A r-2,r-3,r-4 fresh)");
194+
const tSerial0 = performance.now();
195+
await Promise.all([
196+
enqueueCliRun(keyA, work("A.r2")),
197+
enqueueCliRun(keyA, work("A.r3")),
198+
enqueueCliRun(keyA, work("A.r4")),
199+
]);
200+
const serialWall = Math.round(performance.now() - tSerial0);
201+
console.log("");
202+
203+
// ─── verdict ─────────────────────────────────────────────────────────────
204+
console.log("event log (ms since demo start):");
205+
for (const ev of events) {
206+
console.log(` t=${String(ev.t).padStart(7)} ${ev.phase.padEnd(5)} ${ev.label}`);
207+
}
208+
console.log("");
209+
console.log(`phase 0 wall (BEFORE: collapsed serial ≈ ${runMs * 2}ms): ${beforeWall}ms`);
210+
console.log(`phase 1 wall (AFTER overlap expected ≈ ${runMs}ms): ${crossWall}ms`);
211+
console.log(`phase 2 wall (AFTER serial expected ≈ ${runMs * 3}ms): ${serialWall}ms`);
212+
213+
const beforeOK = beforeWall >= runMs * 1.6; // BEFORE collapses to serial
214+
const overlapOK = crossWall < runMs * 1.5;
215+
const serialOK = serialWall >= runMs * 2.5;
216+
console.log("");
217+
console.log(`BEFORE-PR collapse reproduced: ${beforeOK ? "PASS" : "FAIL"}`);
218+
console.log(`distinct-owner overlap: ${overlapOK ? "PASS" : "FAIL"}`);
219+
console.log(`identical-owner serialization: ${serialOK ? "PASS" : "FAIL"}`);
220+
process.exit(beforeOK && overlapOK && serialOK ? 0 : 1);

‎src/agents/cli-runner.helpers.test.ts‎

Lines changed: 56 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import { resolvePreferredOpenClawTmpDir } from "../infra/tmp-openclaw-dir.js";
99
import { escapeRegExp } from "../shared/regexp.js";
1010
import {
1111
buildCliArgs,
12+
buildClaudeOwnerKey,
1213
loadPromptRefImages,
1314
prepareCliPromptImagePayload,
1415
resolveCliRunQueueKey,
@@ -491,7 +492,7 @@ describe("writeCliSystemPromptFile", () => {
491492
});
492493

493494
describe("resolveCliRunQueueKey", () => {
494-
it("scopes Claude CLI serialization to the workspace for fresh runs", () => {
495+
it("falls back to workspaceDir when no ownerKey is supplied (legacy)", () => {
495496
expect(
496497
resolveCliRunQueueKey({
497498
backendId: "claude-cli",
@@ -502,6 +503,18 @@ describe("resolveCliRunQueueKey", () => {
502503
).toBe("claude-cli:workspace:/tmp/project-a");
503504
});
504505

506+
it("scopes Claude CLI serialization to the owner identity for fresh runs", () => {
507+
expect(
508+
resolveCliRunQueueKey({
509+
backendId: "claude-cli",
510+
serialize: true,
511+
runId: "run-1b",
512+
workspaceDir: "/tmp/project-a",
513+
ownerKey: "abcd1234",
514+
}),
515+
).toBe("claude-cli:owner:abcd1234");
516+
});
517+
505518
it("scopes Claude CLI serialization to the resumed CLI session id", () => {
506519
expect(
507520
resolveCliRunQueueKey({
@@ -514,6 +527,19 @@ describe("resolveCliRunQueueKey", () => {
514527
).toBe("claude-cli:session:claude-session-123");
515528
});
516529

530+
it("prefers cliSessionId over ownerKey when resuming", () => {
531+
expect(
532+
resolveCliRunQueueKey({
533+
backendId: "claude-cli",
534+
serialize: true,
535+
runId: "run-2b",
536+
workspaceDir: "/tmp/project-a",
537+
cliSessionId: "claude-session-123",
538+
ownerKey: "abcd1234",
539+
}),
540+
).toBe("claude-cli:session:claude-session-123");
541+
});
542+
517543
it("keeps non-Claude backends on the provider lane when serialized", () => {
518544
expect(
519545
resolveCliRunQueueKey({
@@ -537,3 +563,32 @@ describe("resolveCliRunQueueKey", () => {
537563
).toBe("claude-cli:run-4");
538564
});
539565
});
566+
567+
describe("buildClaudeOwnerKey", () => {
568+
it("is deterministic and distinguishes session keys", () => {
569+
const base = {
570+
agentAccountId: "acct-1",
571+
agentId: "agent-main",
572+
authProfileId: "profile-a",
573+
sessionId: "sess-1",
574+
sessionKey: "key-a",
575+
};
576+
const a1 = buildClaudeOwnerKey(base);
577+
const a2 = buildClaudeOwnerKey(base);
578+
expect(a1).toBe(a2);
579+
const b = buildClaudeOwnerKey({ ...base, sessionKey: "key-b" });
580+
expect(a1).not.toBe(b);
581+
});
582+
583+
it("matches the legacy buildClaudeLiveKey hash for a frozen fixture (DO NOT EDIT — splits queue from live-session map)", () => {
584+
expect(
585+
buildClaudeOwnerKey({
586+
agentAccountId: "acct-1",
587+
agentId: "agent-main",
588+
authProfileId: "profile-a",
589+
sessionId: "sess-1",
590+
sessionKey: "key-a",
591+
}),
592+
).toBe("718b9a6cf473526c3c357883dfc8f1da1cf90b709d9ed38d675b52314abe6800");
593+
});
594+
});

‎src/agents/cli-runner/claude-live-session.ts‎

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ import {
3232
} from "../cli-output.js";
3333
import { classifyFailoverReason } from "../embedded-agent-helpers.js";
3434
import { FailoverError, resolveFailoverStatus } from "../failover-error.js";
35+
import { buildClaudeOwnerKey } from "./helpers.js";
3536
import { cliBackendLog, formatCliBackendOutputDigest } from "./log.js";
3637
import type { PreparedCliRunContext } from "./types.js";
3738

@@ -257,15 +258,13 @@ export function buildClaudeLiveArgs(params: {
257258
}
258259

259260
function buildClaudeLiveKey(context: PreparedCliRunContext): string {
260-
return `${context.backendResolved.id}:${sha256(
261-
JSON.stringify({
262-
agentAccountId: context.params.agentAccountId,
263-
agentId: context.params.agentId,
264-
authProfileId: context.effectiveAuthProfileId,
265-
sessionId: context.params.sessionId,
266-
sessionKey: context.params.sessionKey,
267-
}),
268-
)}`;
261+
return `${context.backendResolved.id}:${buildClaudeOwnerKey({
262+
agentAccountId: context.params.agentAccountId,
263+
agentId: context.params.agentId,
264+
authProfileId: context.effectiveAuthProfileId,
265+
sessionId: context.params.sessionId,
266+
sessionKey: context.params.sessionKey,
267+
})}`;
269268
}
270269

271270
function buildClaudeLiveFingerprint(params: {

‎src/agents/cli-runner/execute.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ import { runClaudeLiveSessionTurn, shouldUseClaudeLiveSession } from "./claude-l
3131
import { prepareClaudeCliSkillsPlugin } from "./claude-skills-plugin.js";
3232
import {
3333
buildCliSupervisorScopeKey,
34+
buildClaudeOwnerKey,
3435
buildCliArgs,
3536
resolveCliRunQueueKey,
3637
enqueueCliRun,
@@ -374,6 +375,13 @@ export async function executePreparedCliRun(
374375
runId: params.runId,
375376
workspaceDir: context.workspaceDir,
376377
cliSessionId: useResume ? resolvedSessionId : undefined,
378+
ownerKey: buildClaudeOwnerKey({
379+
agentAccountId: params.agentAccountId,
380+
agentId: params.agentId,
381+
authProfileId: context.effectiveAuthProfileId,
382+
sessionId: params.sessionId,
383+
sessionKey: params.sessionKey,
384+
}),
377385
});
378386

379387
try {

0 commit comments

Comments
 (0)