Skip to content

Commit 07da843

Browse files
tripolskypetrtripolskypetr
committed
wip
1 parent 07c9176 commit 07da843

9 files changed

Lines changed: 489 additions & 27 deletions

File tree

‎src/client/ClientOperator.ts‎

Lines changed: 38 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,9 @@
11
import { IAgent } from "../interfaces/Agent.interface";
22
import { IOperatorParams } from "../interfaces/Operator.interface";
33
import { ExecutionMode } from "../interfaces/Session.interface";
4-
import { sleep, Subject } from "functools-kit";
4+
import { getErrorMessage, sleep, Subject } from "functools-kit";
55
import { GLOBAL_CONFIG } from "../config/params";
6+
import { errorSubject } from "../config/emitters";
67

78
/**
89
* Type definition for dispose function.
@@ -49,7 +50,17 @@ class OperatorSignal {
4950
if (this._disposeRef) {
5051
const disposeRef = this._disposeRef;
5152
this._disposeRef = null;
52-
disposeRef();
53+
try {
54+
disposeRef();
55+
} catch (error) {
56+
// A throwing user dispose must not prevent the next signal from
57+
// opening — the previous one is torn down on a best-effort basis.
58+
console.error(
59+
`agent-swarm operator signal dispose error agentName=${
60+
this.params.agentName
61+
} clientId=${this.params.clientId} error=${getErrorMessage(error)}`
62+
);
63+
}
5364
}
5465
this._disposeRef = this._signalFactory(message, next);
5566
}
@@ -65,7 +76,17 @@ class OperatorSignal {
6576
if (this._disposeRef) {
6677
const disposeRef = this._disposeRef;
6778
this._disposeRef = null;
68-
await disposeRef();
79+
try {
80+
await disposeRef();
81+
} catch (error) {
82+
// dispose is awaited from waitForOutput's timeout path: a throwing
83+
// user dispose would reject waitForOutput and hang the swarm race.
84+
console.error(
85+
`agent-swarm operator signal dispose error agentName=${
86+
this.params.agentName
87+
} clientId=${this.params.clientId} error=${getErrorMessage(error)}`
88+
);
89+
}
6990
}
7091
}
7192
}
@@ -124,7 +145,20 @@ export class ClientOperator implements IAgent {
124145
`ClientOperator agentName=${this.params.agentName} clientId=${this.params.clientId} execute tool mode forwarded to operator`
125146
);
126147
}
127-
this._operatorSignal.sendMessage(input, this._outgoingSubject.next);
148+
try {
149+
this._operatorSignal.sendMessage(input, this._outgoingSubject.next);
150+
} catch (error) {
151+
// execute is fired without await from ClientSession.execute: a throwing
152+
// operator connector would surface as an unhandled rejection and crash
153+
// the host process. Surface the error instead; the pending waitForOutput
154+
// resolves through the operator timeout.
155+
console.error(
156+
`agent-swarm operator connector error agentName=${
157+
this.params.agentName
158+
} clientId=${this.params.clientId} error=${getErrorMessage(error)}`
159+
);
160+
await errorSubject.next([this.params.clientId, error as Error]);
161+
}
128162
GLOBAL_CONFIG.CC_LOGGER_ENABLE_DEBUG &&
129163
this.params.logger.debug(
130164
`ClientOperator agentName=${this.params.agentName} clientId=${this.params.clientId} execute end`,

‎src/client/ClientPolicy.ts‎

Lines changed: 28 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -48,10 +48,33 @@ export class ClientPolicy implements IPolicy {
4848
* the earlier one — one ban is silently lost in memory and in the persisted
4949
* store. queued() runs the mutations one at a time so each observes the result
5050
* of the previous one.
51+
*
52+
* The wrapped function must never reject: queued() chains every call on the
53+
* previous promise, so a rejection inside the queue (e.g. a throwing
54+
* setBannedClients adapter) would reject every already-queued ban/unban with
55+
* this foreign error WITHOUT running it — a concurrent ban of another client
56+
* would be silently lost. Errors are boxed here and rethrown by the caller
57+
* outside the queue.
5158
*/
52-
private _banQueue = queued(
53-
async (fn: () => Promise<void>): Promise<void> => await fn()
54-
) as (fn: () => Promise<void>) => Promise<void>;
59+
private _banQueue = queued(async (fn: () => Promise<void>) => {
60+
try {
61+
await fn();
62+
return null;
63+
} catch (error) {
64+
return { error };
65+
}
66+
}) as (fn: () => Promise<void>) => Promise<null | { error: unknown }>;
67+
68+
/**
69+
* Runs a ban-set mutation through _banQueue, rethrowing its boxed error
70+
* outside the queue so only the failing caller observes it.
71+
*/
72+
private async _runBanQueue(fn: () => Promise<void>): Promise<void> {
73+
const result = await this._banQueue(fn);
74+
if (result) {
75+
throw result.error;
76+
}
77+
}
5578

5679
/**
5780
* Constructs a ClientPolicy instance with the provided parameters.
@@ -259,7 +282,7 @@ export class ClientPolicy implements IPolicy {
259282
swarmName,
260283
}
261284
);
262-
await this._banQueue(async () => {
285+
await this._runBanQueue(async () => {
263286
const banSet = await this._getBanSet(swarmName);
264287
if (banSet.has(clientId)) {
265288
// Already banned: skip the mutation AND the notifications, matching
@@ -310,7 +333,7 @@ export class ClientPolicy implements IPolicy {
310333
swarmName,
311334
}
312335
);
313-
await this._banQueue(async () => {
336+
await this._runBanQueue(async () => {
314337
const banSet = await this._getBanSet(swarmName);
315338
if (!banSet.has(clientId)) {
316339
// Not banned: skip the mutation and the notifications, matching the

‎src/client/ClientState.ts‎

Lines changed: 40 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -126,15 +126,47 @@ export class ClientState<State extends IStateData = IStateData>
126126
_state: State = null as State;
127127

128128
/**
129-
* Queued dispatch function to read or write the state, delegating to DISPATCH_FN.
130-
* Ensures thread-safe state operations, supporting concurrent access from ClientAgent or tools.
129+
* Queued core of dispatch. The wrapped function must never reject: queued()
130+
* chains every call on the previous promise, so a rejection inside the queue
131+
* would reject every already-queued dispatch with this foreign error WITHOUT
132+
* running it — a concurrent setState would be silently dropped. Errors (e.g.
133+
* a throwing user dispatchFn or middleware) are boxed here and rethrown
134+
* outside the queue, so only the caller whose operation failed observes them.
131135
*/
132-
dispatch = queued(async (action: Action, payload) => {
133-
return await this._dispatchContext.run(
134-
true,
135-
async () => await DISPATCH_FN<State>(action, this, payload)
136-
);
137-
}) as (action: Action, payload?: DispatchFn<State>) => Promise<State>;
136+
private _dispatchQueue = queued(async (action: Action, payload) => {
137+
try {
138+
return {
139+
ok: true as const,
140+
state: await this._dispatchContext.run(
141+
true,
142+
async () => await DISPATCH_FN<State>(action, this, payload)
143+
),
144+
};
145+
} catch (error) {
146+
return { ok: false as const, error };
147+
}
148+
}) as (
149+
action: Action,
150+
payload?: DispatchFn<State>
151+
) => Promise<
152+
{ ok: true; state: State } | { ok: false; error: unknown }
153+
>;
154+
155+
/**
156+
* Dispatches a read or write of the state through the serialized queue,
157+
* delegating to DISPATCH_FN. Ensures thread-safe state operations, supporting
158+
* concurrent access from ClientAgent or tools.
159+
*/
160+
dispatch = async (
161+
action: Action,
162+
payload?: DispatchFn<State>
163+
): Promise<State> => {
164+
const result = await this._dispatchQueue(action, payload);
165+
if ("error" in result) {
166+
throw result.error;
167+
}
168+
return result.state;
169+
};
138170

139171
/**
140172
* Constructs a ClientState instance with the provided parameters.

‎src/client/ClientStorage.ts‎

Lines changed: 32 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -368,12 +368,39 @@ export class ClientStorage<T extends IStorageData = IStorageData>
368368
}
369369

370370
/**
371-
* Dispatches a storage action (upsert, remove, or clear) in a queued manner, delegating to DISPATCH_FN.
372-
* Ensures sequential execution of storage operations, supporting thread-safe updates from ClientAgent or tools.
371+
* Queued core of dispatch. The wrapped function must never reject: queued()
372+
* chains every call on the previous promise, so a rejection inside the queue
373+
* (e.g. a throwing createIndex/createEmbedding/setData) would reject every
374+
* already-queued dispatch with this foreign error WITHOUT running it — a
375+
* concurrent upsert/remove/clear would be silently dropped. Errors are boxed
376+
* here and rethrown outside the queue, so only the failing caller observes them.
373377
*/
374-
dispatch = queued(
375-
async (action, payload) => await DISPATCH_FN<T>(action, this, payload)
376-
) as (action: Action, payload: Partial<Payload<T>>) => Promise<void>;
378+
private _dispatchQueue = queued(async (action, payload) => {
379+
try {
380+
await DISPATCH_FN<T>(action, this, payload);
381+
return null;
382+
} catch (error) {
383+
return { error };
384+
}
385+
}) as (
386+
action: Action,
387+
payload: Partial<Payload<T>>
388+
) => Promise<null | { error: unknown }>;
389+
390+
/**
391+
* Dispatches a storage action (upsert, remove, or clear) through the serialized
392+
* queue, delegating to DISPATCH_FN. Ensures sequential execution of storage
393+
* operations, supporting thread-safe updates from ClientAgent or tools.
394+
*/
395+
dispatch = async (
396+
action: Action,
397+
payload: Partial<Payload<T>>
398+
): Promise<void> => {
399+
const result = await this._dispatchQueue(action, payload);
400+
if (result) {
401+
throw result.error;
402+
}
403+
};
377404

378405
/**
379406
* Creates embeddings for the given item, memoized by item ID to avoid redundant calculations via CREATE_EMBEDDING_FN.

‎src/client/ClientSwarm.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -280,7 +280,10 @@ export class ClientSwarm implements ISwarm {
280280
`ClientSwarm swarmName=${this.params.swarmName} clientId=${this.params.clientId} setBusy`,
281281
{ isBusy }
282282
);
283-
this._isBusy += isBusy ? 1 : -1;
283+
// Clamp at zero: an unmatched setBusy(false) (reachable through the public
284+
// SwarmPublicService.setBusy) would drive the counter negative, and a
285+
// negative counter reads as "busy" forever — deadlocking the session lock.
286+
this._isBusy = Math.max(0, this._isBusy + (isBusy ? 1 : -1));
284287
}
285288

286289
/**

‎test/index.fix3.mjs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
import { run } from 'worker-testbed';
2+
import "./spec/clientfix3.test.mjs";
3+
run(import.meta.url, () => {
4+
console.log("clientfix3 finished");
5+
process.exit(-1);
6+
});

‎test/index.mjs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ import "./spec/toolcontext.test.mjs";
4848
import "./spec/policyrace.test.mjs";
4949
import "./spec/clientfix.test.mjs";
5050
import "./spec/clientfix2.test.mjs";
51+
import "./spec/clientfix3.test.mjs";
5152

5253
run(import.meta.url, () => {
5354
console.log("All tests are finished");

0 commit comments

Comments
 (0)