Skip to content

Commit 0ce7375

Browse files
tripolskypetrtripolskypetr
committed
wip
1 parent 3fdd27a commit 0ce7375

8 files changed

Lines changed: 466 additions & 142 deletions

File tree

‎src/client/ClientAgent.ts‎

Lines changed: 122 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -188,13 +188,31 @@ const createToolCall = async (
188188
// handles this late error (e.g. changeToAgent recursion guard thrown after
189189
// commitToolOutput). Without a recovery emission the pending waitForOutput
190190
// of the enclosing execution would hang forever.
191-
const result = await self._resurrectModel(
192-
mode,
193-
`Late tool error after commit: name=${
194-
tool.function.name
195-
} error=${getErrorMessage(error)}`
196-
);
197-
await self._emitOutput(mode, result, outputEpoch);
191+
try {
192+
const result = await self._resurrectModel(
193+
mode,
194+
`Late tool error after commit: name=${
195+
tool.function.name
196+
} error=${getErrorMessage(error)}`
197+
);
198+
await self._emitOutput(mode, result, outputEpoch);
199+
} catch (recoveryError) {
200+
// createToolCall is fired without await: a throwing recovery
201+
// (_emitOutput throws by contract when validation fails twice) would
202+
// be an unhandled rejection. Resolve the waiter directly instead.
203+
console.error(
204+
`agent-swarm late tool error recovery failed functionName=${
205+
tool.function.name
206+
} error=${getErrorMessage(recoveryError)}`
207+
);
208+
await errorSubject.next([
209+
self.params.clientId,
210+
recoveryError as Error,
211+
]);
212+
if (outputEpoch === self._outputEpoch) {
213+
await self._outputSubject.next(createPlaceholder());
214+
}
215+
}
198216
}
199217
} finally {
200218
self._runningToolCalls -= 1;
@@ -422,26 +440,34 @@ const EXECUTE_FN = async (
422440
if (!toolCalls.length) {
423441
// maxToolCalls=0 dropped every call: the model wanted tools but none may
424442
// run. Emit the (tool-stripped) content as the answer instead of leaving
425-
// the pending waitForOutput hanging forever.
426-
const result = await self.params.transform(
427-
message.content,
428-
self.params.clientId,
429-
self.params.agentName
430-
);
443+
// the pending waitForOutput hanging forever. _emitOutput owns the
444+
// transform, so pass the raw content — transforming here too would
445+
// apply a non-idempotent transform twice.
431446
await self.params.history.push({
432447
...message,
433448
tool_calls: [],
434449
agentName: self.params.agentName,
435450
});
436-
await self._emitOutput(mode, result, outputEpoch);
451+
await self._emitOutput(mode, message.content, outputEpoch);
437452
return;
438453
}
439454

440455
{
441-
const priorityTool = toolCalls.find((call) =>
442-
swarm.navigationSchemaService.hasTool(call.function.name) ||
443-
swarm.actionSchemaService.hasTool(call.function.name)
444-
);
456+
// Navigation/action tools are registered by toolName, while the model
457+
// calls tools by function.name — check both, since they diverge for MCP
458+
// tools (mcp_ prefix) and dynamic tool schemas.
459+
const priorityTool = toolCalls.find((call) => {
460+
const targetFn = toolList.find(
461+
(t) => t.function.name === call.function.name
462+
);
463+
return (
464+
swarm.navigationSchemaService.hasTool(call.function.name) ||
465+
swarm.actionSchemaService.hasTool(call.function.name) ||
466+
(targetFn &&
467+
(swarm.navigationSchemaService.hasTool(targetFn.toolName) ||
468+
swarm.actionSchemaService.hasTool(targetFn.toolName)))
469+
);
470+
});
445471
if (priorityTool) {
446472
toolCalls = [priorityTool];
447473
}
@@ -656,30 +682,59 @@ const EXECUTE_FN = async (
656682
self.params.logger.debug(
657683
`ClientAgent agentName=${self.params.agentName} clientId=${self.params.clientId} functionName=${tool.function.name} the next tool execution stopped due to the call error`
658684
);
659-
const result = await self._resurrectModel(
660-
mode,
661-
`Function call failed with error: name=${
662-
tool.function.name
663-
} arguments=${JSON.stringify(tool.function.arguments)}`
664-
);
665-
GLOBAL_CONFIG.CC_LOGGER_ENABLE_DEBUG &&
666-
self.params.logger.debug(
667-
`ClientAgent agentName=${self.params.agentName} clientId=${self.params.clientId} execute end result=${result}`
685+
try {
686+
const result = await self._resurrectModel(
687+
mode,
688+
`Function call failed with error: name=${
689+
tool.function.name
690+
} arguments=${JSON.stringify(tool.function.arguments)}`
691+
);
692+
GLOBAL_CONFIG.CC_LOGGER_ENABLE_DEBUG &&
693+
self.params.logger.debug(
694+
`ClientAgent agentName=${self.params.agentName} clientId=${self.params.clientId} execute end result=${result}`
695+
);
696+
await self._emitOutput(mode, result, outputEpoch);
697+
} catch (recoveryError) {
698+
// Nobody awaits the status chain: a throwing recovery
699+
// (_emitOutput throws by contract when validation fails twice)
700+
// would reject it unhandled. Resolve the waiter directly instead.
701+
console.error(
702+
`agent-swarm tool error recovery failed functionName=${
703+
tool.function.name
704+
} error=${getErrorMessage(recoveryError)}`
668705
);
669-
await self._emitOutput(mode, result, outputEpoch);
706+
await errorSubject.next([
707+
self.params.clientId,
708+
recoveryError as Error,
709+
]);
710+
if (outputEpoch === self._outputEpoch) {
711+
await self._outputSubject.next(createPlaceholder());
712+
}
713+
}
670714
}
671715
return status;
672716
});
673717
}
674-
lastToolStatusRef.finally(() => {
675-
self._activeToolChains -= 1;
676-
self.params.onAfterToolCalls &&
677-
self.params.onAfterToolCalls(
678-
self.params.clientId,
679-
self.params.agentName,
680-
toolCalls
718+
lastToolStatusRef
719+
.catch((error) => {
720+
// The chain is fire-and-forget; a rejected link (e.g. a throwing
721+
// params.map/bus adapter) must not surface as an unhandled rejection.
722+
console.error(
723+
`agent-swarm tool status chain error agentName=${
724+
self.params.agentName
725+
} clientId=${self.params.clientId} error=${getErrorMessage(error)}`
681726
);
682-
});
727+
return null;
728+
})
729+
.finally(() => {
730+
self._activeToolChains -= 1;
731+
self.params.onAfterToolCalls &&
732+
self.params.onAfterToolCalls(
733+
self.params.clientId,
734+
self.params.agentName,
735+
toolCalls
736+
);
737+
});
683738
run(true);
684739
return;
685740
}
@@ -689,34 +744,18 @@ const EXECUTE_FN = async (
689744
`ClientAgent agentName=${self.params.agentName} clientId=${self.params.clientId} execute no tool calls detected`
690745
);
691746
}
692-
const result = await self.params.transform(
693-
message.content,
694-
self.params.clientId,
695-
self.params.agentName
696-
);
697747
await self.params.history.push({
698748
...message,
699749
agentName: self.params.agentName,
700750
});
701-
let validation: string | null = null;
702-
if ((validation = await self.params.validate(result))) {
703-
GLOBAL_CONFIG.CC_LOGGER_ENABLE_DEBUG &&
704-
self.params.logger.debug(
705-
`ClientAgent agentName=${self.params.agentName} clientId=${self.params.clientId} execute invalid tool call detected: ${validation}`,
706-
{ result }
707-
);
708-
const result1 = await self._resurrectModel(
709-
mode,
710-
`Invalid model output: ${result}`
711-
);
712-
await self._emitOutput(mode, result1, outputEpoch);
713-
return;
714-
}
715751
GLOBAL_CONFIG.CC_LOGGER_ENABLE_DEBUG &&
716752
self.params.logger.debug(
717-
`ClientAgent agentName=${self.params.agentName} clientId=${self.params.clientId} execute end result=${result}`
753+
`ClientAgent agentName=${self.params.agentName} clientId=${self.params.clientId} execute end result=${message.content}`
718754
);
719-
await self._emitOutput(mode, result, outputEpoch);
755+
// _emitOutput owns transform + validate (with resurrect on failure), so the
756+
// raw content goes straight through — transforming/validating here as well
757+
// ran every transform twice per answer and broke non-idempotent transforms.
758+
await self._emitOutput(mode, message.content, outputEpoch);
720759
};
721760

722761
/**
@@ -1080,6 +1119,14 @@ export class ClientAgent implements IAgent {
10801119
}
10811120
this.params.onOutput &&
10821121
this.params.onOutput(this.params.clientId, this.params.agentName, result);
1122+
// Keep the callback surface symmetric with the resurrect branch above:
1123+
// the emitted output is an assistant message on both paths.
1124+
this.params.onAssistantMessage &&
1125+
this.params.onAssistantMessage(
1126+
this.params.clientId,
1127+
this.params.agentName,
1128+
result
1129+
);
10831130
await this._outputSubject.next(result);
10841131
await this.params.bus.emit<IBusEvent>(this.params.clientId, {
10851132
type: "emit-output",
@@ -1150,7 +1197,24 @@ export class ClientAgent implements IAgent {
11501197
await this._resqueSubject.next(MODEL_RESQUE_SYMBOL);
11511198
return placeholder;
11521199
}
1153-
const rawMessage = await this.getCompletion(mode, []);
1200+
let rawMessage: ISwarmMessage;
1201+
try {
1202+
rawMessage = await this.getCompletion(mode, []);
1203+
} catch (error) {
1204+
// _resurrectModel IS the recovery path: if the resque completion itself
1205+
// rejects (e.g. the provider is down — often the very reason we are
1206+
// here), rethrowing would surface as an unhandled rejection from the
1207+
// fire-and-forget callers (createToolCall, tool status chain). Degrade
1208+
// to the placeholder instead.
1209+
console.error(
1210+
`agent-swarm _resurrectModel completion error agentName=${
1211+
this.params.agentName
1212+
} clientId=${this.params.clientId} error=${getErrorMessage(error)}`
1213+
);
1214+
await errorSubject.next([this.params.clientId, error as Error]);
1215+
await this._resqueSubject.next(MODEL_RESQUE_SYMBOL);
1216+
return placeholder;
1217+
}
11541218
if (rawMessage.tool_calls?.length) {
11551219
GLOBAL_CONFIG.CC_LOGGER_ENABLE_DEBUG &&
11561220
this.params.logger.debug(

‎src/client/ClientHistory.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,8 @@ export class ClientHistory implements IHistory {
6868
} clientId=${this.params.clientId} error=${getErrorMessage(error)}`
6969
);
7070
await errorSubject.next([this.params.clientId, error as Error]);
71+
// The record was dropped: do not announce a successful push on the bus.
72+
return;
7173
}
7274
await this.params.bus.emit<IBusEvent>(this.params.clientId, {
7375
type: "push",

‎src/client/ClientOperator.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,14 @@ class OperatorSignal {
4343
`OperatorSignal agentName=${this.params.agentName} clientId=${this.params.clientId} sendMessage`,
4444
{ message }
4545
);
46+
// Dispose the previous signal before opening a new one: overwriting the
47+
// ref leaks the pending signal, and its late answer would resolve the
48+
// NEXT waitForOutput with a reply to the previous question.
49+
if (this._disposeRef) {
50+
const disposeRef = this._disposeRef;
51+
this._disposeRef = null;
52+
disposeRef();
53+
}
4654
this._disposeRef = this._signalFactory(message, next);
4755
}
4856

0 commit comments

Comments
 (0)