mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-24 11:25:50 -06:00
Merge remote-tracking branch 'origin/main' into fix/inline-slash-skills
This commit is contained in:
Jesse Merhi
@@ -380,6 +380,7 @@ export async function runBridgeRequest(params: {
|
||||
parentToolCallId: string;
|
||||
codeModeRunId: string;
|
||||
maxOutputBytes: number;
|
||||
remainingMs: number;
|
||||
ctx: ToolSearchToolContext;
|
||||
request: PendingBridgeRequest;
|
||||
signal?: AbortSignal;
|
||||
@@ -439,7 +440,24 @@ export async function runBridgeRequest(params: {
|
||||
if (!binding) {
|
||||
throw new ToolInputError(`Unknown catalog function: ${callableName}.`);
|
||||
}
|
||||
const called = await params.runtime.callExactId(binding.id, values[1] ?? {}, {
|
||||
let input = values[1] ?? {};
|
||||
if (
|
||||
binding.source === "openclaw" &&
|
||||
binding.name === "exec" &&
|
||||
binding.input?.includes("yieldMs") === true &&
|
||||
isRecord(input) &&
|
||||
input.background !== true &&
|
||||
input.yieldMs === undefined
|
||||
) {
|
||||
// The shell's 10s default equals Code Mode's default budget. Yield
|
||||
// within the remaining shared deadline so late sequential calls can
|
||||
// still return their process handle and resume the guest inline.
|
||||
input = {
|
||||
...input,
|
||||
yieldMs: Math.max(1, Math.min(1_000, Math.floor(params.remainingMs / 4))),
|
||||
};
|
||||
}
|
||||
const called = await params.runtime.callExactId(binding.id, input, {
|
||||
parentToolCallId: params.parentToolCallId,
|
||||
signal: params.signal,
|
||||
onUpdate: params.onUpdate,
|
||||
|
||||
@@ -334,6 +334,7 @@ async function settleCodeModeResult(params: {
|
||||
namespaceRuntime: params.namespaceRuntime,
|
||||
parentToolCallId: params.parentToolCallId,
|
||||
codeModeRunId: params.codeModeReplayId,
|
||||
deadlineMs: settleDeadline,
|
||||
activeRunId,
|
||||
ctx: params.ctx,
|
||||
signal: params.signal,
|
||||
@@ -469,6 +470,7 @@ async function settleCodeModeResult(params: {
|
||||
namespaceRuntime: params.namespaceRuntime,
|
||||
parentToolCallId: params.parentToolCallId,
|
||||
codeModeRunId: params.codeModeReplayId,
|
||||
deadlineMs: settleDeadline,
|
||||
activeRunId,
|
||||
ctx: params.ctx,
|
||||
signal: params.signal,
|
||||
@@ -512,6 +514,7 @@ async function settleCodeModeResult(params: {
|
||||
catalogProjection: params.catalogProjection,
|
||||
namespaceRuntime: params.namespaceRuntime,
|
||||
output,
|
||||
deadlineMs: settleDeadline,
|
||||
deliveredOutputCount,
|
||||
reservedActiveRunSlot: params.reservedActiveRunSlot,
|
||||
replaySafe: params.replaySafe,
|
||||
|
||||
@@ -307,6 +307,7 @@ export async function runCodeModeScriptHeadless(params: {
|
||||
namespaceRuntime,
|
||||
parentToolCallId,
|
||||
codeModeRunId,
|
||||
deadlineMs: deadline,
|
||||
ctx: params.ctx,
|
||||
signal: abortScope.signal,
|
||||
}),
|
||||
|
||||
@@ -238,6 +238,7 @@ export function snapshotState(params: {
|
||||
catalogProjection: CodeModeCatalogProjection;
|
||||
namespaceRuntime: CodeModeNamespaceRuntime;
|
||||
output: unknown[];
|
||||
deadlineMs: number;
|
||||
deliveredOutputCount?: number;
|
||||
reservedActiveRunSlot?: boolean;
|
||||
replaySafe: boolean;
|
||||
@@ -321,6 +322,7 @@ export function createPendingBridgeStates(params: {
|
||||
namespaceRuntime: CodeModeNamespaceRuntime;
|
||||
parentToolCallId: string;
|
||||
codeModeRunId: string;
|
||||
deadlineMs: number;
|
||||
activeRunId?: string;
|
||||
ctx: ToolSearchToolContext;
|
||||
signal?: AbortSignal;
|
||||
@@ -342,6 +344,7 @@ export function createPendingBridgeStates(params: {
|
||||
parentToolCallId: params.parentToolCallId,
|
||||
codeModeRunId: params.codeModeRunId,
|
||||
maxOutputBytes: params.config.maxOutputBytes,
|
||||
remainingMs: Math.max(1, params.deadlineMs - Date.now()),
|
||||
ctx: params.ctx,
|
||||
request,
|
||||
signal,
|
||||
|
||||
@@ -155,6 +155,122 @@ describe("Code Mode bridge settlement and cancellation", () => {
|
||||
expect(testing.activeRuns.size).toBe(0);
|
||||
});
|
||||
|
||||
it("yields nested exec before the Code Mode deadline when continuation args are omitted", async () => {
|
||||
const catalogRef = createToolSearchCatalogRef();
|
||||
const config = {
|
||||
tools: { codeMode: { enabled: true, timeoutMs: 10_000 } },
|
||||
} as never;
|
||||
const ctx = {
|
||||
config,
|
||||
runtimeConfig: config,
|
||||
sessionId: "session-code-mode",
|
||||
sessionKey: "agent:main:main",
|
||||
runId: "run-code-mode",
|
||||
catalogRef,
|
||||
};
|
||||
const codeModeTools = createCodeModeTools(ctx);
|
||||
const shell = pluginToolWithExecute("exec", "Run shell", async (_toolCallId, input) =>
|
||||
jsonResult(input),
|
||||
);
|
||||
shell.parameters = Type.Object({
|
||||
command: Type.String(),
|
||||
yieldMs: Type.Optional(Type.Number()),
|
||||
background: Type.Optional(Type.Boolean()),
|
||||
});
|
||||
applyCodeModeCatalog({
|
||||
tools: [...codeModeTools, shell],
|
||||
config,
|
||||
sessionId: "session-code-mode",
|
||||
sessionKey: "agent:main:main",
|
||||
runId: "run-code-mode",
|
||||
catalogRef,
|
||||
});
|
||||
|
||||
const details = resultDetails(
|
||||
await expectDefined(codeModeTools[0], "Code Mode exec test invariant").execute(
|
||||
"code-call-shell-yield",
|
||||
{
|
||||
code: `return [
|
||||
await exec({ command: "default" }),
|
||||
await exec({ command: "explicit", yieldMs: 4_000 }),
|
||||
await exec({ command: "background", background: true }),
|
||||
];`,
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
expect(details).toMatchObject({
|
||||
status: "completed",
|
||||
value: [
|
||||
{ command: "default", yieldMs: 1_000 },
|
||||
{ command: "explicit", yieldMs: 4_000 },
|
||||
{ command: "background", background: true },
|
||||
],
|
||||
});
|
||||
expect(testing.activeRuns.size).toBe(0);
|
||||
});
|
||||
|
||||
it("bounds nested exec yield by the shared remaining deadline", async () => {
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] });
|
||||
const catalogRef = createToolSearchCatalogRef();
|
||||
const config = {
|
||||
tools: { codeMode: { enabled: true, timeoutMs: 10_000 } },
|
||||
} as never;
|
||||
const ctx = {
|
||||
config,
|
||||
runtimeConfig: config,
|
||||
sessionId: "session-code-mode",
|
||||
sessionKey: "agent:main:main",
|
||||
runId: "run-code-mode",
|
||||
catalogRef,
|
||||
};
|
||||
const codeModeTools = createCodeModeTools(ctx);
|
||||
const consumeBudget = pluginToolWithExecute(
|
||||
"fake_consume_budget",
|
||||
"Consume most of the shared Code Mode deadline",
|
||||
async () => {
|
||||
vi.advanceTimersByTime(9_600);
|
||||
return jsonResult({ consumed: true });
|
||||
},
|
||||
);
|
||||
const shell = pluginToolWithExecute("exec", "Run shell", async (_toolCallId, input) =>
|
||||
jsonResult(input),
|
||||
);
|
||||
shell.parameters = Type.Object({
|
||||
command: Type.String(),
|
||||
yieldMs: Type.Optional(Type.Number()),
|
||||
background: Type.Optional(Type.Boolean()),
|
||||
});
|
||||
applyCodeModeCatalog({
|
||||
tools: [...codeModeTools, consumeBudget, shell],
|
||||
config,
|
||||
sessionId: "session-code-mode",
|
||||
sessionKey: "agent:main:main",
|
||||
runId: "run-code-mode",
|
||||
catalogRef,
|
||||
});
|
||||
|
||||
const details = resultDetails(
|
||||
await expectDefined(codeModeTools[0], "Code Mode exec test invariant").execute(
|
||||
"code-call-late-shell-yield",
|
||||
{
|
||||
code: `
|
||||
await fake_consume_budget({});
|
||||
return await exec({ command: "late" });
|
||||
`,
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
expect(details).toMatchObject({
|
||||
status: "completed",
|
||||
value: { command: "late", yieldMs: 100 },
|
||||
});
|
||||
expect(consumeBudget.execute).toHaveBeenCalledOnce();
|
||||
expect(shell.execute).toHaveBeenCalledOnce();
|
||||
expect(testing.activeRuns.size).toBe(0);
|
||||
});
|
||||
|
||||
it("supports a guest timer between an action and its observation", async () => {
|
||||
const catalogRef = createToolSearchCatalogRef();
|
||||
const config = {
|
||||
|
||||
@@ -74,6 +74,9 @@ export function extractMessagingToolSourceReplyPayload(
|
||||
if (idempotencyKey) {
|
||||
payload.idempotencyKey = idempotencyKey;
|
||||
}
|
||||
if (details.sourceReplyTranscriptOwner === true) {
|
||||
payload.transcriptOwner = true;
|
||||
}
|
||||
return Object.keys(payload).length > 0 ? payload : undefined;
|
||||
}
|
||||
|
||||
|
||||
@@ -29,6 +29,7 @@ export type MessagingToolSourceReplyPayload = Pick<
|
||||
| "text"
|
||||
> & {
|
||||
idempotencyKey?: string;
|
||||
transcriptOwner?: true;
|
||||
/** Current-source progress (`false`) or completed reply (`true`). */
|
||||
sourceReplyFinal?: boolean;
|
||||
};
|
||||
|
||||
@@ -530,6 +530,9 @@ export function buildEmbeddedRunPayloads(params: {
|
||||
if (item.sourceReplyMirror.idempotencyKey) {
|
||||
sourceReplyTranscriptMirror.idempotencyKey = item.sourceReplyMirror.idempotencyKey;
|
||||
}
|
||||
if (item.sourceReplyMirror.transcriptOwner) {
|
||||
sourceReplyTranscriptMirror.transcriptOwner = true;
|
||||
}
|
||||
setReplyPayloadMetadata(payload, {
|
||||
sourceReplyTranscriptMirror,
|
||||
});
|
||||
|
||||
@@ -23,7 +23,7 @@ type EmbeddedRunReplyItem = {
|
||||
interactive?: ReplyPayload["interactive"];
|
||||
channelData?: Record<string, unknown>;
|
||||
nonTerminalToolErrorWarning?: boolean;
|
||||
sourceReplyMirror?: { idempotencyKey?: string };
|
||||
sourceReplyMirror?: { idempotencyKey?: string; transcriptOwner?: true };
|
||||
};
|
||||
|
||||
/** Builds transcript mirrors and completion evidence for message-tool source replies. */
|
||||
@@ -70,6 +70,7 @@ export function buildSourceReplyPayloadState(params: {
|
||||
idempotencyKey:
|
||||
payload.idempotencyKey ??
|
||||
(params.runId ? `${params.runId}:internal-source-reply:${index}` : undefined),
|
||||
...(payload.transcriptOwner ? { transcriptOwner: true as const } : {}),
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
@@ -5,6 +5,13 @@ import path from "node:path";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { getReplyPayloadMetadata } from "../../auto-reply/reply-payload.js";
|
||||
import { buildReplyPayloads } from "../../auto-reply/reply/agent-runner-payloads.js";
|
||||
import { mirrorDeliveredReplyToTranscript } from "../../auto-reply/reply/dispatch-from-config.transcript.js";
|
||||
import {
|
||||
loadTranscriptEvents,
|
||||
replaceSessionEntry,
|
||||
} from "../../config/sessions/session-accessor.js";
|
||||
import { resolveManagedOutgoingMediaArtifactDownload } from "../../gateway/managed-image-attachments.js";
|
||||
import { listManagedImageRecordEntries } from "../../gateway/managed-image-record-store.js";
|
||||
import { withOpenClawTestState } from "../../test-utils/openclaw-test-state.js";
|
||||
import { extractMessagingToolSourceReplyPayload } from "../embedded-agent-messaging-extraction.js";
|
||||
import { buildEmbeddedRunPayloads } from "../embedded-agent-runner/run/payloads.js";
|
||||
@@ -28,6 +35,9 @@ function createCurrentSourceMessageTool(params: { workspaceDir?: string } = {})
|
||||
});
|
||||
}
|
||||
|
||||
const TINY_PNG_BASE64 =
|
||||
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9Y9ZQmcAAAAASUVORK5CYII=";
|
||||
|
||||
describe("WebChat message tool internal source reply", () => {
|
||||
it("projects a real targetless send and preserves the automatic final reply", async () => {
|
||||
const tool = createCurrentSourceMessageTool();
|
||||
@@ -132,4 +142,119 @@ describe("WebChat message tool internal source reply", () => {
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it("persists one managed image for overlapping internal source replies", async () => {
|
||||
await withOpenClawTestState(
|
||||
{ layout: "state-only", prefix: "openclaw-internal-source-reply-" },
|
||||
async (state) => {
|
||||
const stateDir = state.stateDir;
|
||||
const workspaceDir = state.workspaceDir;
|
||||
const storePath = path.join(stateDir, "agents", "main", "sessions", "sessions.json");
|
||||
const sessionKey = "agent:main:webchat:dm:restart-proof";
|
||||
const sessionId = "restart-proof-session";
|
||||
const imagePath = path.join(workspaceDir, "restart-proof.png");
|
||||
await fs.mkdir(workspaceDir, { recursive: true });
|
||||
await fs.writeFile(imagePath, Buffer.from(TINY_PNG_BASE64, "base64"));
|
||||
|
||||
await replaceSessionEntry(
|
||||
{ agentId: "main", sessionKey, storePath },
|
||||
{ sessionId, chatType: "direct", updatedAt: 1 },
|
||||
);
|
||||
const config = {
|
||||
agents: {
|
||||
entries: {
|
||||
main: { default: true, workspace: workspaceDir },
|
||||
},
|
||||
},
|
||||
};
|
||||
const tool = createMessageTool({
|
||||
config,
|
||||
currentChannelProvider: "webchat",
|
||||
agentSessionKey: sessionKey,
|
||||
runSessionKey: sessionKey,
|
||||
sessionId,
|
||||
agentId: "main",
|
||||
runId: "restart-proof-run",
|
||||
getScopedChannelsCommandSecretTargets: () => ({ targetIds: new Set<string>() }),
|
||||
resolveCommandSecretRefsViaGateway: async () => ({
|
||||
resolvedConfig: config,
|
||||
diagnostics: [],
|
||||
targetStatesByPath: {},
|
||||
hadUnresolvedTargets: false,
|
||||
}),
|
||||
});
|
||||
|
||||
const sendParams = {
|
||||
action: "send" as const,
|
||||
message: "Durable image reply",
|
||||
media: imagePath,
|
||||
};
|
||||
const [toolResult, overlappingResult] = await Promise.all([
|
||||
tool.execute("restart-proof-call", sendParams),
|
||||
tool.execute("restart-proof-call", sendParams),
|
||||
]);
|
||||
const sourceReply = extractMessagingToolSourceReplyPayload(toolResult);
|
||||
expect(sourceReply).toMatchObject({ transcriptOwner: true });
|
||||
expect(overlappingResult.details).toMatchObject({
|
||||
idempotencyKey: sourceReply?.idempotencyKey,
|
||||
sourceReplyTranscriptOwner: true,
|
||||
});
|
||||
const sourcePayloads = buildEmbeddedRunPayloads({
|
||||
assistantTexts: [],
|
||||
lastAssistant: undefined,
|
||||
currentAssistant: undefined,
|
||||
sessionKey,
|
||||
agentId: "main",
|
||||
sourceReplyDeliveryMode: "message_tool_only",
|
||||
messagingToolSourceReplyPayloads: sourceReply ? [sourceReply] : [],
|
||||
runId: "restart-proof-run",
|
||||
verboseLevel: "off",
|
||||
reasoningLevel: "off",
|
||||
toolResultFormat: "plain",
|
||||
});
|
||||
const mirror = getReplyPayloadMetadata(
|
||||
sourcePayloads[0] as object,
|
||||
)?.sourceReplyTranscriptMirror;
|
||||
expect(mirror).toMatchObject({ transcriptOwner: true });
|
||||
await mirrorDeliveredReplyToTranscript({
|
||||
metadata: mirror ? { ...mirror, expectedSessionId: sessionId, storePath } : undefined,
|
||||
cfg: config,
|
||||
});
|
||||
const events = await loadTranscriptEvents({
|
||||
agentId: "main",
|
||||
sessionId,
|
||||
sessionKey,
|
||||
storePath,
|
||||
});
|
||||
const assistants = events
|
||||
.map((event) => (event as { message?: Record<string, unknown> }).message)
|
||||
.filter((message) => message?.role === "assistant");
|
||||
expect(assistants).toHaveLength(1);
|
||||
const assistant = assistants[0];
|
||||
const content = Array.isArray(assistant?.content)
|
||||
? (assistant.content as Array<Record<string, unknown>>)
|
||||
: [];
|
||||
const image = content.find((block) => block.type === "image");
|
||||
expect(toolResult.details).toMatchObject({
|
||||
sourceReplySink: "internal-ui",
|
||||
idempotencyKey: expect.any(String),
|
||||
});
|
||||
expect(content[0]).toEqual({ type: "text", text: "Durable image reply" });
|
||||
expect(image).toMatchObject({
|
||||
type: "image",
|
||||
artifactId: expect.stringMatching(/^artifact_managed_image_/u),
|
||||
});
|
||||
expect(JSON.stringify(assistant)).not.toContain(imagePath);
|
||||
expect(listManagedImageRecordEntries({ stateDir, sessionKey })).toHaveLength(1);
|
||||
await expect(
|
||||
resolveManagedOutgoingMediaArtifactDownload({
|
||||
sessionKey,
|
||||
agentId: "main",
|
||||
artifactId: String(image?.artifactId),
|
||||
stateDir,
|
||||
}),
|
||||
).resolves.toMatchObject({ type: "image" });
|
||||
},
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -294,6 +294,8 @@ export type ReplyPayloadMetadata = {
|
||||
expectedSessionId?: string;
|
||||
/** Delivery stays live, but neither side may be appended to a transcript. */
|
||||
transcriptWriteBlocked?: boolean;
|
||||
/** The visible reply already owns its durable transcript row. */
|
||||
transcriptOwner?: boolean;
|
||||
text?: string;
|
||||
mediaUrls?: string[];
|
||||
idempotencyKey?: string;
|
||||
|
||||
@@ -30,7 +30,7 @@ export async function mirrorDeliveredReplyToTranscript(params: {
|
||||
cfg: OpenClawConfig;
|
||||
}): Promise<void> {
|
||||
const mirror = params.metadata;
|
||||
if (!mirror) {
|
||||
if (!mirror || mirror.transcriptOwner) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
import type { ReplyPayload } from "../auto-reply/reply-payload.js";
|
||||
import { appendAssistantMessageToSessionTranscript } from "../config/sessions.js";
|
||||
import { resolveSessionStorePathCore } from "../config/sessions/paths.js";
|
||||
import {
|
||||
findTranscriptEvent,
|
||||
readTranscriptEventMessage,
|
||||
} from "../config/sessions/session-accessor.sqlite-read.js";
|
||||
import { getOwnedSessionTranscriptWriterFence } from "../config/sessions/transcript-write-context.js";
|
||||
import type { OpenClawConfig } from "../config/types.openclaw.js";
|
||||
import { getAgentScopedMediaLocalRootsForSources } from "../media/local-roots.js";
|
||||
import { createKeyedFifoLeaseRegistry } from "../shared/keyed-fifo-lease.js";
|
||||
import { isOpenClawDeliveryMirrorAssistantMessage } from "../shared/transcript-only-openclaw-assistant.js";
|
||||
import {
|
||||
attachManagedOutgoingMediaToMessage,
|
||||
createManagedOutgoingMediaBlocks,
|
||||
} from "./managed-image-attachments.js";
|
||||
import { prepareGatewayInjectedAssistantContent } from "./server-methods/chat-transcript-inject.js";
|
||||
|
||||
const internalSourceReplyPersistenceLeases = createKeyedFifoLeaseRegistry(
|
||||
Symbol.for("openclaw.internalSourceReplyPersistenceLeases"),
|
||||
);
|
||||
|
||||
function collectSourceReplyMediaUrls(payload: ReplyPayload): string[] {
|
||||
return Array.from(
|
||||
new Set([...(payload.mediaUrl ? [payload.mediaUrl] : []), ...(payload.mediaUrls ?? [])]),
|
||||
).filter((value) => value.trim().length > 0);
|
||||
}
|
||||
|
||||
async function hasPersistedInternalSourceReply(params: {
|
||||
cfg: OpenClawConfig;
|
||||
sessionKey: string;
|
||||
expectedSessionId?: string;
|
||||
agentId?: string;
|
||||
idempotencyKey?: string;
|
||||
}): Promise<boolean> {
|
||||
if (!params.expectedSessionId || !params.idempotencyKey) {
|
||||
return false;
|
||||
}
|
||||
const storePath = resolveSessionStorePathCore(params.cfg.session?.store, {
|
||||
agentId: params.agentId,
|
||||
});
|
||||
const found = await findTranscriptEvent(
|
||||
{
|
||||
agentId: params.agentId,
|
||||
sessionId: params.expectedSessionId,
|
||||
sessionKey: params.sessionKey,
|
||||
storePath,
|
||||
},
|
||||
(event) => {
|
||||
const message = readTranscriptEventMessage(event);
|
||||
return (
|
||||
message?.idempotencyKey === params.idempotencyKey &&
|
||||
isOpenClawDeliveryMirrorAssistantMessage(message)
|
||||
);
|
||||
},
|
||||
);
|
||||
return found !== undefined;
|
||||
}
|
||||
|
||||
function resolveInternalSourceReplyPersistenceLeaseKey(params: {
|
||||
sessionKey: string;
|
||||
expectedSessionId?: string;
|
||||
agentId?: string;
|
||||
idempotencyKey?: string;
|
||||
}): string | undefined {
|
||||
if (!params.idempotencyKey) {
|
||||
return undefined;
|
||||
}
|
||||
return JSON.stringify([
|
||||
params.agentId ?? "",
|
||||
params.sessionKey,
|
||||
params.expectedSessionId ?? "",
|
||||
params.idempotencyKey,
|
||||
]);
|
||||
}
|
||||
|
||||
/** Persist the private WebChat source reply before its successful tool result becomes visible. */
|
||||
export async function persistInternalSourceReply(params: {
|
||||
cfg: OpenClawConfig;
|
||||
sessionKey: string;
|
||||
expectedSessionId?: string;
|
||||
agentId?: string;
|
||||
payload: ReplyPayload;
|
||||
idempotencyKey?: string;
|
||||
sourceReplyFinal?: boolean;
|
||||
toolCallId?: string;
|
||||
sourceTurnId?: string;
|
||||
}): Promise<void> {
|
||||
const leaseKey = resolveInternalSourceReplyPersistenceLeaseKey(params);
|
||||
const lease = leaseKey ? internalSourceReplyPersistenceLeases.reserve([leaseKey]) : undefined;
|
||||
await lease?.wait();
|
||||
try {
|
||||
if (await hasPersistedInternalSourceReply(params)) {
|
||||
return;
|
||||
}
|
||||
const mediaUrls = collectSourceReplyMediaUrls(params.payload);
|
||||
const mediaBlocks = await createManagedOutgoingMediaBlocks({
|
||||
sessionKey: params.sessionKey,
|
||||
agentId: params.agentId,
|
||||
mediaUrls,
|
||||
localRoots: getAgentScopedMediaLocalRootsForSources({
|
||||
cfg: params.cfg,
|
||||
agentId: params.agentId,
|
||||
mediaSources: mediaUrls,
|
||||
}),
|
||||
});
|
||||
const content: Array<Record<string, unknown>> = [
|
||||
...(params.payload.text ? [{ type: "text", text: params.payload.text }] : []),
|
||||
...mediaBlocks,
|
||||
];
|
||||
const writerFence = getOwnedSessionTranscriptWriterFence();
|
||||
const appended = await appendAssistantMessageToSessionTranscript({
|
||||
agentId: params.agentId,
|
||||
sessionKey: params.sessionKey,
|
||||
...(params.expectedSessionId ? { expectedSessionId: params.expectedSessionId } : {}),
|
||||
...(writerFence?.expectedLifecycleRevision !== undefined
|
||||
? { expectedLifecycleRevision: writerFence.expectedLifecycleRevision }
|
||||
: {}),
|
||||
...(writerFence ? { expectedWriterRunId: writerFence.expectedWriterRunId } : {}),
|
||||
content: prepareGatewayInjectedAssistantContent(content),
|
||||
idempotencyKey: params.idempotencyKey,
|
||||
...(params.sourceReplyFinal !== undefined
|
||||
? {
|
||||
deliveryMirror: {
|
||||
kind: "message-tool-source-reply" as const,
|
||||
final: params.sourceReplyFinal,
|
||||
...(params.toolCallId ? { toolCallId: params.toolCallId } : {}),
|
||||
...(params.sourceTurnId ? { sourceTurnId: params.sourceTurnId } : {}),
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
config: params.cfg,
|
||||
});
|
||||
if (!appended.ok) {
|
||||
throw new Error(`Internal source reply persistence failed: ${appended.reason}`);
|
||||
}
|
||||
if (
|
||||
mediaBlocks.length > 0 &&
|
||||
!attachManagedOutgoingMediaToMessage({
|
||||
messageId: appended.messageId,
|
||||
blocks: mediaBlocks,
|
||||
})
|
||||
) {
|
||||
throw new Error("Internal source reply media ownership could not be persisted");
|
||||
}
|
||||
} finally {
|
||||
lease?.release();
|
||||
}
|
||||
}
|
||||
@@ -14,6 +14,7 @@ import { getAgentScopedMediaLocalRoots } from "../../media/local-roots.js";
|
||||
import { resolveAgentScopedOutboundMediaAccess } from "../../media/read-capability.js";
|
||||
import { readBooleanParam } from "../../plugin-sdk/boolean-param.js";
|
||||
import { hasPollCreationParams } from "../../poll-params.js";
|
||||
import { createLazyRuntimeModule } from "../../shared/lazy-runtime.js";
|
||||
import { INTERNAL_MESSAGE_CHANNEL } from "../../utils/message-channel.js";
|
||||
import { formatErrorMessage } from "../errors.js";
|
||||
import { throwIfAborted } from "./abort.js";
|
||||
@@ -49,6 +50,10 @@ import {
|
||||
} from "./outbound-policy.js";
|
||||
import { getRuntimeVisibleChannelPlugin } from "./runtime-visible-channels.js";
|
||||
|
||||
const loadInternalSourceReplyPersistence = createLazyRuntimeModule(
|
||||
() => import("../../gateway/internal-source-reply-persistence.js"),
|
||||
);
|
||||
|
||||
export function getToolResult(result: MessageActionResult): AgentToolResult<unknown> | undefined {
|
||||
return "toolResult" in result ? result.toolResult : undefined;
|
||||
}
|
||||
@@ -294,12 +299,37 @@ async function handleInternalSourceReplySendAction(
|
||||
}
|
||||
const sourceReplyMediaUrls = resolveSendableOutboundReplyParts(sourceReplyPayload).mediaUrls;
|
||||
const sourceReplyMessage = sourceReplyPayload.text ?? sourceReply.message;
|
||||
const idempotencyKey = normalizeOptionalString(params.idempotencyKey);
|
||||
let persistedIdempotencyKey: string | undefined;
|
||||
let persistedTranscriptOwner = false;
|
||||
if (!dryRun && input.sessionId) {
|
||||
const sessionKey = input.sourceReplySessionKey ?? input.sessionKey;
|
||||
if (!sessionKey) {
|
||||
throw new Error("Internal source reply requires a session key");
|
||||
}
|
||||
const { persistInternalSourceReply } = await loadInternalSourceReplyPersistence();
|
||||
await persistInternalSourceReply({
|
||||
cfg: input.cfg,
|
||||
sessionKey,
|
||||
expectedSessionId: input.sessionId,
|
||||
agentId: input.agentId ?? resolveSessionAgentId({ sessionKey, config: input.cfg }),
|
||||
payload: sourceReplyPayload,
|
||||
idempotencyKey,
|
||||
sourceReplyFinal: input.sourceReplyFinal,
|
||||
toolCallId: input.sourceReplyToolCallId,
|
||||
sourceTurnId: input.messageActionAuthorization?.toolContext?.currentSourceTurnId,
|
||||
});
|
||||
persistedIdempotencyKey = idempotencyKey;
|
||||
persistedTranscriptOwner = true;
|
||||
}
|
||||
const payload = {
|
||||
status: "ok",
|
||||
deliveryStatus: dryRun ? "dry_run" : "sent",
|
||||
channel: INTERNAL_MESSAGE_CHANNEL,
|
||||
target: "current-run",
|
||||
sourceReplyDeliveryMode: input.sourceReplyDeliveryMode,
|
||||
...(persistedIdempotencyKey ? { idempotencyKey: persistedIdempotencyKey } : {}),
|
||||
...(persistedTranscriptOwner ? { sourceReplyTranscriptOwner: true as const } : {}),
|
||||
...(dryRun ? {} : { sourceReplySink: "internal-ui" as const }),
|
||||
sourceReply: sourceReplyPayload,
|
||||
...(sourceReplyMessage ? { message: sourceReplyMessage } : {}),
|
||||
@@ -328,6 +358,8 @@ function buildInternalSourceReplyToolResult(payload: {
|
||||
channel: ChannelId;
|
||||
target: string;
|
||||
sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
|
||||
idempotencyKey?: string;
|
||||
sourceReplyTranscriptOwner?: true;
|
||||
sourceReplySink?: "internal-ui";
|
||||
sourceReply: ReplyPayload;
|
||||
message?: string;
|
||||
@@ -340,6 +372,8 @@ function buildInternalSourceReplyToolResult(payload: {
|
||||
channel: ChannelId;
|
||||
target: string;
|
||||
sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
|
||||
idempotencyKey?: string;
|
||||
sourceReplyTranscriptOwner?: true;
|
||||
sourceReplySink?: "internal-ui";
|
||||
sourceReply: ReplyPayload;
|
||||
message?: string;
|
||||
@@ -364,6 +398,8 @@ function buildInternalSourceReplyToolResult(payload: {
|
||||
...(payload.sourceReplyDeliveryMode
|
||||
? { sourceReplyDeliveryMode: payload.sourceReplyDeliveryMode }
|
||||
: {}),
|
||||
...(payload.idempotencyKey ? { idempotencyKey: payload.idempotencyKey } : {}),
|
||||
...(payload.sourceReplyTranscriptOwner ? { sourceReplyTranscriptOwner: true as const } : {}),
|
||||
...(payload.sourceReplySink ? { sourceReplySink: payload.sourceReplySink } : {}),
|
||||
sourceReply: payload.sourceReply,
|
||||
...(payload.message ? { message: payload.message } : {}),
|
||||
|
||||
Reference in New Issue
Block a user