Skip to content

Commit 07c9176

Browse files
tripolskypetrtripolskypetr
committed
patch
1 parent 0ce7375 commit 07c9176

5 files changed

Lines changed: 221 additions & 21 deletions

File tree

‎src/client/ClientAgent.ts‎

Lines changed: 16 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -876,8 +876,12 @@ export class ClientAgent implements IAgent {
876876
const seen = new Set<string>();
877877
const agentToolList: IAgentTool[] = [];
878878
if (this.params.tools) {
879-
await Promise.all(
880-
this.params.tools.map(async (tool) => {
879+
// Resolve availability/dynamic schemas concurrently, but keep the RESULT
880+
// in declaration order: pushing from within the concurrent map would order
881+
// tools by resolution time, making the tool list, the seen-based dedup and
882+
// the single-commit-action filter all non-deterministic across runs.
883+
const resolvedTools = await Promise.all(
884+
this.params.tools.map(async (tool): Promise<IAgentTool | null> => {
881885
const {
882886
toolName,
883887
function: upperFn,
@@ -894,7 +898,7 @@ export class ClientAgent implements IAgent {
894898
)
895899
)
896900
) {
897-
return;
901+
return null;
898902
}
899903
} catch (error) {
900904
// A throwing isAvailable would reject the queued EXECUTE_FN and
@@ -905,7 +909,7 @@ export class ClientAgent implements IAgent {
905909
)}`
906910
);
907911
await errorSubject.next([this.params.clientId, error as Error]);
908-
return;
912+
return null;
909913
}
910914
let fn: typeof upperFn;
911915
try {
@@ -922,15 +926,20 @@ export class ClientAgent implements IAgent {
922926
)}`
923927
);
924928
await errorSubject.next([this.params.clientId, error as Error]);
925-
return;
929+
return null;
926930
}
927-
agentToolList.push({
931+
return {
928932
...other,
929933
toolName,
930934
function: fn,
931-
});
935+
};
932936
})
933937
);
938+
for (const tool of resolvedTools) {
939+
if (tool) {
940+
agentToolList.push(tool);
941+
}
942+
}
934943
}
935944
let mcpToolList: Awaited<ReturnType<typeof this.params.mcp.listTools>> = [];
936945
try {

‎src/client/ClientState.ts‎

Lines changed: 20 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import { getErrorMessage, queued, singleshot, Subject } from "functools-kit";
2+
import { AsyncLocalStorage } from "async_hooks";
23
import { errorSubject } from "../config/emitters";
34
import {
45
IState,
@@ -101,11 +102,22 @@ export class ClientState<State extends IStateData = IStateData>
101102
public readonly stateChanged = new Subject<StateName>();
102103

103104
/**
104-
* True while a queued dispatch (read/write) is executing for this instance.
105-
* getState checks it to serve reentrant reads (getState inside a setState
106-
* dispatchFn) without re-entering the queue, which would deadlock.
105+
* Marks the async execution context of a running dispatch (read/write).
106+
* getState checks it to serve reentrant reads (getState called from INSIDE a
107+
* setState dispatchFn/middleware) without re-entering the queue, which would
108+
* deadlock. Scoping this per async-context — instead of a plain instance flag —
109+
* ensures an UNRELATED concurrent getState still queues and observes writes in
110+
* order, rather than reading a stale field while some other write is in flight.
107111
*/
108-
_inDispatch = false;
112+
private _dispatchContext = new AsyncLocalStorage<true>();
113+
114+
/**
115+
* True only while the caller runs inside the async context of an active
116+
* dispatch on this instance (i.e. a reentrant call from within a dispatchFn).
117+
*/
118+
get _inDispatch(): boolean {
119+
return this._dispatchContext.getStore() === true;
120+
}
109121

110122
/**
111123
* The current state data, initialized as null and set during waitForInit.
@@ -118,12 +130,10 @@ export class ClientState<State extends IStateData = IStateData>
118130
* Ensures thread-safe state operations, supporting concurrent access from ClientAgent or tools.
119131
*/
120132
dispatch = queued(async (action: Action, payload) => {
121-
this._inDispatch = true;
122-
try {
123-
return await DISPATCH_FN<State>(action, this, payload);
124-
} finally {
125-
this._inDispatch = false;
126-
}
133+
return await this._dispatchContext.run(
134+
true,
135+
async () => await DISPATCH_FN<State>(action, this, payload)
136+
);
127137
}) as (action: Action, payload?: DispatchFn<State>) => Promise<State>;
128138

129139
/**

‎test/spec/state.test.mjs‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -339,3 +339,115 @@ test("Will keep state order even if not awaited", async ({
339339

340340
pass();
341341
});
342+
343+
test("Will observe an in-flight write from an unrelated concurrent getState", async ({
344+
pass,
345+
fail,
346+
}) => {
347+
const TEST_COMPLETION = addCompletion({
348+
completionName: "test_completion",
349+
getCompletion: async ({ agentName, messages }) => {
350+
const [{ content }] = messages.slice(-1);
351+
return { agentName, content, role: "assistant" };
352+
},
353+
});
354+
355+
const TEST_STATE = addState({
356+
stateName: "test_state",
357+
getDefaultState: () => "foo",
358+
});
359+
360+
const TEST_AGENT = addAgent({
361+
agentName: "test_agent",
362+
completion: TEST_COMPLETION,
363+
prompt: "x",
364+
states: [TEST_STATE],
365+
});
366+
367+
const TEST_SWARM = addSwarm({
368+
swarmName: "test_swarm",
369+
agentList: [TEST_AGENT],
370+
defaultAgent: TEST_AGENT,
371+
});
372+
373+
const CLIENT_ID = randomString();
374+
session(CLIENT_ID, TEST_SWARM);
375+
const ref = { clientId: CLIENT_ID, agentName: TEST_AGENT, stateName: TEST_STATE };
376+
377+
await State.setState(() => "foo", ref);
378+
379+
// A slow write, not awaited: it holds the dispatch queue for 200ms.
380+
const writePromise = State.setState(async () => {
381+
await sleep(200);
382+
return "WRITTEN";
383+
}, ref);
384+
385+
// Let the write enter the dispatch, then issue an UNRELATED getState.
386+
// It must queue behind the write and observe "WRITTEN", not the stale "foo".
387+
await sleep(20);
388+
const readDuringWrite = await State.getState(ref);
389+
await writePromise;
390+
391+
if (readDuringWrite !== "WRITTEN") {
392+
fail(`concurrent getState bypassed the dispatch queue: got ${readDuringWrite}`);
393+
}
394+
395+
pass();
396+
});
397+
398+
test("Will serve a reentrant getState from inside a setState without deadlock", async ({
399+
pass,
400+
fail,
401+
}) => {
402+
const TEST_COMPLETION = addCompletion({
403+
completionName: "test_completion",
404+
getCompletion: async ({ agentName, messages }) => {
405+
const [{ content }] = messages.slice(-1);
406+
return { agentName, content, role: "assistant" };
407+
},
408+
});
409+
410+
const TEST_STATE = addState({
411+
stateName: "test_state",
412+
getDefaultState: () => "init",
413+
});
414+
415+
const TEST_AGENT = addAgent({
416+
agentName: "test_agent",
417+
completion: TEST_COMPLETION,
418+
prompt: "x",
419+
states: [TEST_STATE],
420+
});
421+
422+
const TEST_SWARM = addSwarm({
423+
swarmName: "test_swarm",
424+
agentList: [TEST_AGENT],
425+
defaultAgent: TEST_AGENT,
426+
});
427+
428+
const CLIENT_ID = randomString();
429+
session(CLIENT_ID, TEST_SWARM);
430+
const ref = { clientId: CLIENT_ID, agentName: TEST_AGENT, stateName: TEST_STATE };
431+
432+
let reentrantValue = "NOT_RUN";
433+
const writePromise = State.setState(async () => {
434+
// getState called from INSIDE the dispatchFn must not deadlock behind
435+
// this very write; it reads the current value directly.
436+
reentrantValue = await State.getState(ref);
437+
return "after-reentrant";
438+
}, ref);
439+
440+
const settled = await Promise.race([
441+
writePromise.then(() => "SETTLED"),
442+
sleep(2_000).then(() => "DEADLOCK"),
443+
]);
444+
445+
if (settled !== "SETTLED") {
446+
fail("reentrant getState deadlocked behind its enclosing setState");
447+
}
448+
if (reentrantValue !== "init") {
449+
fail(`reentrant getState returned the wrong value: ${reentrantValue}`);
450+
}
451+
452+
pass();
453+
});

‎test/spec/toolcall.test.mjs‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -615,3 +615,64 @@ test("Will name the default agent in triage navigation tool output", async ({ pa
615615
}
616616
fail(`accept=${accept} outputs=${JSON.stringify(toolOutputs)}`);
617617
});
618+
619+
test("Will present tools in declaration order regardless of isAvailable timing", async ({ pass, fail }) => {
620+
setConfig({ CC_PERSIST_ENABLED_BY_DEFAULT: false });
621+
const CLIENT_ID = randomString();
622+
let seenToolNames = null;
623+
624+
const MOCK_COMPLETION = addCompletion({
625+
completionName: "mock-completion",
626+
getCompletion: async ({ agentName, messages, tools }) => {
627+
seenToolNames = (tools || []).map((t) => t.function.name);
628+
const [last] = messages.slice(-1);
629+
return { agentName, content: last.content || "ok", role: "assistant" };
630+
},
631+
});
632+
633+
// Declaration order a,b,c,d. isAvailable resolves LAST-declared FIRST, so the
634+
// old completion-order push would reverse them; the fix must keep a,b,c,d.
635+
const makeTool = (letter, delay) =>
636+
addTool({
637+
toolName: `ord-tool-${letter}`,
638+
isAvailable: async () => {
639+
await new Promise((r) => setTimeout(r, delay));
640+
return true;
641+
},
642+
validate: () => true,
643+
type: "function",
644+
function: {
645+
name: `ord_fn_${letter}`,
646+
description: "",
647+
parameters: { type: "object", properties: {}, required: [] },
648+
},
649+
call: async () => {},
650+
});
651+
652+
const TEST_AGENT = addAgent({
653+
agentName: "ord-agent",
654+
completion: MOCK_COMPLETION,
655+
prompt: "",
656+
tools: [
657+
makeTool("a", 40),
658+
makeTool("b", 30),
659+
makeTool("c", 20),
660+
makeTool("d", 5),
661+
],
662+
});
663+
664+
const TEST_SWARM = addSwarm({
665+
swarmName: "ord-swarm",
666+
agentList: [TEST_AGENT],
667+
defaultAgent: TEST_AGENT,
668+
});
669+
670+
await complete("start", CLIENT_ID, TEST_SWARM);
671+
672+
const got = (seenToolNames || []).join(",");
673+
if (got === "ord_fn_a,ord_fn_b,ord_fn_c,ord_fn_d") {
674+
pass();
675+
return;
676+
}
677+
fail(`tool order not preserved: ${got}`);
678+
});

‎types.d.ts‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -6483,11 +6483,19 @@ declare class ClientState<State extends IStateData = IStateData> implements ISta
64836483
readonly params: IStateParams<State>;
64846484
readonly stateChanged: Subject<string>;
64856485
/**
6486-
* True while a queued dispatch (read/write) is executing for this instance.
6487-
* getState checks it to serve reentrant reads (getState inside a setState
6488-
* dispatchFn) without re-entering the queue, which would deadlock.
6486+
* Marks the async execution context of a running dispatch (read/write).
6487+
* getState checks it to serve reentrant reads (getState called from INSIDE a
6488+
* setState dispatchFn/middleware) without re-entering the queue, which would
6489+
* deadlock. Scoping this per async-context — instead of a plain instance flag —
6490+
* ensures an UNRELATED concurrent getState still queues and observes writes in
6491+
* order, rather than reading a stale field while some other write is in flight.
64896492
*/
6490-
_inDispatch: boolean;
6493+
private _dispatchContext;
6494+
/**
6495+
* True only while the caller runs inside the async context of an active
6496+
* dispatch on this instance (i.e. a reentrant call from within a dispatchFn).
6497+
*/
6498+
get _inDispatch(): boolean;
64916499
/**
64926500
* The current state data, initialized as null and set during waitForInit.
64936501
* Updated by setState and clearState, persisted via params.setState if provided.

0 commit comments

Comments
 (0)