File size: 5,543 Bytes
eb3f11e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 | /**
* Nested agent-step executor.
*
* Sends annotated inter-session messages through in-process or Gateway execution and reads the assistant reply.
*/
import crypto from "node:crypto";
import { getRuntimeConfig } from "../../config/config.js";
import { resolveSessionStorePathCore } from "../../config/sessions/paths.js";
import { annotateInterSessionPromptText } from "../../sessions/input-provenance.js";
import { recordSessionParticipantBestEffort } from "../../sessions/session-participant-recording.js";
import { INTERNAL_MESSAGE_CHANNEL } from "../../utils/message-channel.js";
import { resolveNestedAgentLaneForSession } from "../lanes.js";
import { waitForAgentRunReply } from "../run-wait.js";
import {
callAgentToolGatewayRequest,
type AgentToolGatewayRequestCaller,
} from "./in-process-gateway.js";
type GatewayCaller = AgentToolGatewayRequestCaller;
type AgentCommandRunner = typeof import("../../commands/agent.js").agentCommandFromIngress;
const defaultAgentStepDeps = {
agentCommandFromIngress: (async (...args) => {
const { agentCommandFromIngress } = await import("../../commands/agent.js");
return await agentCommandFromIngress(...args);
}) as AgentCommandRunner,
};
let agentStepDeps: {
agentCommandFromIngress: AgentCommandRunner;
} = defaultAgentStepDeps;
function extractAgentCommandReply(
result: Awaited<ReturnType<AgentCommandRunner>>,
): string | undefined {
const error = result?.meta.error;
// Plain incomplete-turn output is a control failure; trusted terminal tool presentations remain deliverable.
if (error?.kind === "incomplete_turn" && error.terminalPresentation !== true) {
return undefined;
}
const texts = result?.payloads
?.map((payload) => payload.text)
.filter((text): text is string => Boolean(text?.trim()));
return texts?.length ? texts.join("\n\n") : undefined;
}
/** Sends one annotated message to a target session and returns the resulting assistant text. */
export async function runAgentStep(params: {
agentId?: string;
sessionKey: string;
message: string;
extraSystemPrompt: string;
timeoutMs: number;
channel?: string;
lane?: string;
transcriptMessage?: string;
sourceAgentId?: string;
sourceSessionKey?: string;
sourceChannel?: string;
sourceTool?: string;
callGateway?: GatewayCaller;
}): Promise<string | undefined> {
const promptedAt = Date.now();
const stepIdem = crypto.randomUUID();
const inputProvenance = {
kind: "inter_session" as const,
sourceSessionKey: params.sourceSessionKey,
sourceChannel: params.sourceChannel,
sourceTool: params.sourceTool ?? "sessions_send",
};
// Mark inter-session prompts so downstream transcripts can distinguish tool-routed text.
const message = annotateInterSessionPromptText(params.message, inputProvenance);
const lane = params.lane ?? resolveNestedAgentLaneForSession(params.sessionKey);
const channel = params.channel ?? INTERNAL_MESSAGE_CHANNEL;
const gatewayCall = params.callGateway ?? callAgentToolGatewayRequest;
if (params.transcriptMessage !== undefined) {
// Intentional direct in-process exception: the public agent schema rejects transcriptMessage.
// Keep announce bookkeeping off the wire without expanding the model-authored RPC surface.
const result = await agentStepDeps.agentCommandFromIngress({
message,
...(params.agentId ? { agentId: params.agentId } : {}),
transcriptMessage: params.transcriptMessage,
sessionKey: params.sessionKey,
deliver: false,
sourceReplyDeliveryMode: "message_tool_only",
channel,
lane,
runId: stepIdem,
extraSystemPrompt: params.extraSystemPrompt,
inputProvenance,
allowModelOverride: false,
});
return extractAgentCommandReply(result);
}
const response = await gatewayCall({
method: "agent",
params: {
message,
...(params.agentId ? { agentId: params.agentId } : {}),
sessionKey: params.sessionKey,
idempotencyKey: stepIdem,
deliver: false,
sourceReplyDeliveryMode: "message_tool_only",
channel,
lane,
extraSystemPrompt: params.extraSystemPrompt,
inputProvenance,
},
timeoutMs: 10_000,
});
if (params.sourceAgentId && params.agentId) {
recordSessionParticipantBestEffort({
identity: { type: "agent", id: params.sourceAgentId },
agentId: params.agentId,
sessionKey: params.sessionKey,
storePath: resolveSessionStorePathCore(getRuntimeConfig().session?.store, {
agentId: params.agentId,
}),
promptedAt,
});
}
const stepRunId = typeof response?.runId === "string" && response.runId ? response.runId : "";
const resolvedRunId = stepRunId || stepIdem;
const result = await waitForAgentRunReply({
runId: resolvedRunId,
timeoutMs: Math.min(params.timeoutMs, 60_000),
callGateway: gatewayCall,
untilTerminal: true,
});
if (result.status !== "ok") {
return undefined;
}
return result.replyText;
}
/** Test-only dependency overrides for gateway and in-process command execution. */
const testing = {
setDepsForTest(
overrides?: Partial<{
agentCommandFromIngress: AgentCommandRunner;
}>,
) {
agentStepDeps = overrides
? {
...defaultAgentStepDeps,
...overrides,
}
: defaultAgentStepDeps;
},
};
if (process.env.VITEST || process.env.NODE_ENV === "test") {
(globalThis as Record<PropertyKey, unknown>)[Symbol.for("openclaw.agentStepTestApi")] = {
testing,
};
}
|