github/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-05-06 21:41:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

ACP: carry dedupe/projector updates onto configurable acpx branch

Browse Source
This commit is contained in:
Onur
2026-03-01 09:19:11 +01:00
committed by Onur Solmaz
parent f88bc09f85
commit 2466a9bb13
14 changed files with 1076 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
src/auto-reply/reply/abort.test.ts
View File

@@ -43,6 +43,26 @@ vi.mock("../../agents/subagent-registry.js", () => ({
markSubagentRunTerminated: subagentRegistryMocks.markSubagentRunTerminated,
}));
const acpManagerMocks = vi.hoisted(() => ({
resolveSession: vi.fn<
() =>
| { kind: "none" }
| {
kind: "ready";
sessionKey: string;
meta: unknown;
}
>(() => ({ kind: "none" })),
cancelSession: vi.fn(async () => {}),
}));
vi.mock("../../acp/control-plane/manager.js", () => ({
getAcpSessionManager: () => ({
resolveSession: acpManagerMocks.resolveSession,
cancelSession: acpManagerMocks.cancelSession,
}),
}));
describe("abort detection", () => {
async function writeSessionStore(
storePath: string,
@@ -106,6 +126,8 @@ describe("abort detection", () => {
afterEach(() => {
resetAbortMemoryForTest();
acpManagerMocks.resolveSession.mockReset().mockReturnValue({ kind: "none" });
acpManagerMocks.cancelSession.mockReset().mockResolvedValue(undefined);
});
it("triggerBodyNormalized extracts /stop from RawBody for abort detection", async () => {
@@ -355,6 +377,85 @@ describe("abort detection", () => {
expect(commandQueueMocks.clearCommandLane).toHaveBeenCalledWith(`session:${sessionKey}`);
});
it("plain-language stop on ACP-bound session triggers ACP cancel", async () => {
const sessionKey = "agent:codex:acp:test-1";
const sessionId = "session-123";
const { cfg } = await createAbortConfig({
sessionIdsByKey: { [sessionKey]: sessionId },
});
acpManagerMocks.resolveSession.mockReturnValue({
kind: "ready",
sessionKey,
meta: {} as never,
});
const result = await runStopCommand({
cfg,
sessionKey,
from: "telegram:123",
to: "telegram:123",
targetSessionKey: sessionKey,
});
expect(result.handled).toBe(true);
expect(acpManagerMocks.cancelSession).toHaveBeenCalledWith({
cfg,
sessionKey,
reason: "fast-abort",
});
});
it("ACP cancel failures do not skip queue and lane cleanup", async () => {
const sessionKey = "agent:codex:acp:test-2";
const sessionId = "session-456";
const { root, cfg } = await createAbortConfig({
sessionIdsByKey: { [sessionKey]: sessionId },
});
const followupRun: FollowupRun = {
prompt: "queued",
enqueuedAt: Date.now(),
run: {
agentId: "main",
agentDir: path.join(root, "agent"),
sessionId,
sessionKey,
messageProvider: "telegram",
agentAccountId: "acct",
sessionFile: path.join(root, "session.jsonl"),
workspaceDir: path.join(root, "workspace"),
config: cfg,
provider: "anthropic",
model: "claude-opus-4-5",
timeoutMs: 1000,
blockReplyBreak: "text_end",
},
};
enqueueFollowupRun(
sessionKey,
followupRun,
{ mode: "collect", debounceMs: 0, cap: 20, dropPolicy: "summarize" },
"none",
);
acpManagerMocks.resolveSession.mockReturnValue({
kind: "ready",
sessionKey,
meta: {} as never,
});
acpManagerMocks.cancelSession.mockRejectedValueOnce(new Error("cancel failed"));
const result = await runStopCommand({
cfg,
sessionKey,
from: "telegram:123",
to: "telegram:123",
targetSessionKey: sessionKey,
});
expect(result.handled).toBe(true);
expect(getFollowupQueueDepth(sessionKey)).toBe(0);
expect(commandQueueMocks.clearCommandLane).toHaveBeenCalledWith(`session:${sessionKey}`);
});
it("persists abort cutoff metadata on /stop when command and target session match", async () => {
const sessionKey = "telegram:123";
const sessionId = "session-123";

24
src/auto-reply/reply/abort.ts
View File

@@ -1,3 +1,4 @@
import { getAcpSessionManager } from "../../acp/control-plane/manager.js";
import { resolveSessionAgentId } from "../../agents/agent-scope.js";
import { abortEmbeddedPiRun } from "../../agents/pi-embedded.js";
import {
@@ -301,9 +302,28 @@ export async function tryFastAbortFromMessage(params: {
const storePath = resolveStorePath(cfg.session?.store, { agentId });
const store = loadSessionStore(storePath);
const { entry, key } = resolveSessionEntryForKey(store, targetKey);
const resolvedTargetKey = key ?? targetKey;
const acpManager = getAcpSessionManager();
const acpResolution = acpManager.resolveSession({
cfg,
sessionKey: resolvedTargetKey,
});
if (acpResolution.kind !== "none") {
try {
await acpManager.cancelSession({
cfg,
sessionKey: resolvedTargetKey,
reason: "fast-abort",
});
} catch (error) {
logVerbose(
`abort: ACP cancel failed for ${resolvedTargetKey}: ${error instanceof Error ? error.message : String(error)}`,
);
}
}
const sessionId = entry?.sessionId;
const aborted = sessionId ? abortEmbeddedPiRun(sessionId) : false;
const cleared = clearSessionQueues([key ?? targetKey, sessionId]);
const cleared = clearSessionQueues([resolvedTargetKey, sessionId]);
if (cleared.followupCleared > 0 || cleared.laneCleared > 0) {
logVerbose(
`abort: cleared followups=${cleared.followupCleared} lane=${cleared.laneCleared} keys=${cleared.keys.join(",")}`,
@@ -311,7 +331,7 @@ export async function tryFastAbortFromMessage(params: {
}
const abortCutoff = shouldPersistAbortCutoff({
commandSessionKey: ctx.SessionKey,
targetSessionKey: key ?? targetKey,
targetSessionKey: resolvedTargetKey,
})
? resolveAbortCutoffFromContext(ctx)
: undefined;

333
src/auto-reply/reply/acp-projector.test.ts
View File

@@ -1,5 +1,6 @@
import { describe, expect, it, vi } from "vitest";
import { describe, expect, it } from "vitest";
import type { OpenClawConfig } from "../../config/config.js";
import { prefixSystemMessage } from "../../infra/system-message.js";
import { createAcpReplyProjector } from "./acp-projector.js";
function createCfg(overrides?: Partial<OpenClawConfig>): OpenClawConfig {
@@ -8,7 +9,7 @@ function createCfg(overrides?: Partial<OpenClawConfig>): OpenClawConfig {
enabled: true,
stream: {
coalesceIdleMs: 0,
maxChunkChars: 50,
maxChunkChars: 64,
},
},
...overrides,
@@ -29,71 +30,123 @@ describe("createAcpReplyProjector", () => {
await projector.onEvent({
type: "text_delta",
text: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
text: "a".repeat(70),
tag: "agent_message_chunk",
});
await projector.flush(true);
expect(deliveries).toEqual([
{ kind: "block", text: "a".repeat(64) },
{ kind: "block", text: "a".repeat(6) },
]);
});
it("supports deliveryMode=final_only by buffering deltas until done", async () => {
const deliveries: Array<{ kind: string; text?: string }> = [];
const projector = createAcpReplyProjector({
cfg: createCfg({
acp: {
enabled: true,
stream: {
coalesceIdleMs: 0,
maxChunkChars: 512,
deliveryMode: "final_only",
},
},
}),
shouldSendToolSummaries: true,
deliver: async (kind, payload) => {
deliveries.push({ kind, text: payload.text });
return true;
},
});
await projector.onEvent({
type: "text_delta",
text: "What",
tag: "agent_message_chunk",
});
await projector.onEvent({
type: "text_delta",
text: "bbbbbbbbbb",
text: " now?",
tag: "agent_message_chunk",
});
await projector.flush(true);
expect(deliveries).toEqual([
{
kind: "block",
text: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
},
{ kind: "block", text: "aabbbbbbbbbb" },
]);
});
it("buffers tiny token deltas and flushes once at turn end", async () => {
const deliveries: Array<{ kind: string; text?: string }> = [];
const projector = createAcpReplyProjector({
cfg: createCfg({
acp: {
enabled: true,
stream: {
coalesceIdleMs: 0,
maxChunkChars: 256,
},
},
}),
shouldSendToolSummaries: true,
provider: "discord",
deliver: async (kind, payload) => {
deliveries.push({ kind, text: payload.text });
return true;
},
});
await projector.onEvent({ type: "text_delta", text: "What" });
await projector.onEvent({ type: "text_delta", text: " do" });
await projector.onEvent({ type: "text_delta", text: " you want to work on?" });
expect(deliveries).toEqual([]);
await projector.flush(true);
expect(deliveries).toEqual([{ kind: "block", text: "What do you want to work on?" }]);
await projector.onEvent({ type: "done" });
expect(deliveries).toEqual([{ kind: "block", text: "What now?" }]);
});
it("filters thought stream text and suppresses tool summaries when disabled", async () => {
const deliver = vi.fn(async () => true);
const projector = createAcpReplyProjector({
it("suppresses usage_update by default and allows deduped usage when enabled", async () => {
const hidden: Array<{ kind: string; text?: string }> = [];
const hiddenProjector = createAcpReplyProjector({
cfg: createCfg(),
shouldSendToolSummaries: false,
deliver,
shouldSendToolSummaries: true,
deliver: async (kind, payload) => {
hidden.push({ kind, text: payload.text });
return true;
},
});
await hiddenProjector.onEvent({
type: "status",
text: "usage updated: 10/100",
tag: "usage_update",
used: 10,
size: 100,
});
expect(hidden).toEqual([]);
const shown: Array<{ kind: string; text?: string }> = [];
const shownProjector = createAcpReplyProjector({
cfg: createCfg({
acp: {
enabled: true,
stream: {
coalesceIdleMs: 0,
maxChunkChars: 64,
showUsage: true,
tagVisibility: {
usage_update: true,
},
},
},
}),
shouldSendToolSummaries: true,
deliver: async (kind, payload) => {
shown.push({ kind, text: payload.text });
return true;
},
});
await projector.onEvent({ type: "text_delta", text: "internal", stream: "thought" });
await projector.onEvent({ type: "status", text: "running tool" });
await projector.onEvent({ type: "tool_call", text: "ls" });
await projector.flush(true);
await shownProjector.onEvent({
type: "status",
text: "usage updated: 10/100",
tag: "usage_update",
used: 10,
size: 100,
});
await shownProjector.onEvent({
type: "status",
text: "usage updated: 10/100",
tag: "usage_update",
used: 10,
size: 100,
});
await shownProjector.onEvent({
type: "status",
text: "usage updated: 11/100",
tag: "usage_update",
used: 11,
size: 100,
});
expect(deliver).not.toHaveBeenCalled();
expect(shown).toEqual([
{ kind: "tool", text: prefixSystemMessage("usage updated: 10/100") },
{ kind: "tool", text: prefixSystemMessage("usage updated: 11/100") },
]);
});
it("emits status and tool_call summaries when enabled", async () => {
it("dedupes repeated tool lifecycle updates in minimal mode", async () => {
const deliveries: Array<{ kind: string; text?: string }> = [];
const projector = createAcpReplyProjector({
cfg: createCfg(),
@@ -104,16 +157,70 @@ describe("createAcpReplyProjector", () => {
},
});
await projector.onEvent({ type: "status", text: "planning" });
await projector.onEvent({ type: "tool_call", text: "exec ls" });
await projector.onEvent({
type: "tool_call",
tag: "tool_call",
toolCallId: "call_1",
status: "in_progress",
title: "List files",
text: "List files (in_progress)",
});
await projector.onEvent({
type: "tool_call",
tag: "tool_call_update",
toolCallId: "call_1",
status: "in_progress",
title: "List files",
text: "List files (in_progress)",
});
await projector.onEvent({
type: "tool_call",
tag: "tool_call_update",
toolCallId: "call_1",
status: "completed",
title: "List files",
text: "List files (completed)",
});
await projector.onEvent({
type: "tool_call",
tag: "tool_call_update",
toolCallId: "call_1",
status: "completed",
title: "List files",
text: "List files (completed)",
});
expect(deliveries).toEqual([
{ kind: "tool", text: "⚙️ planning" },
{ kind: "tool", text: "🧰 exec ls" },
]);
expect(deliveries.length).toBe(2);
expect(deliveries[0]?.kind).toBe("tool");
expect(deliveries[0]?.text).toContain("Tool Call");
expect(deliveries[1]?.kind).toBe("tool");
expect(deliveries[1]?.text).toContain("Tool Call");
});
it("flushes pending streamed text before tool/status updates", async () => {
it("renders fallback tool labels without leaking call ids as primary label", async () => {
const deliveries: Array<{ kind: string; text?: string }> = [];
const projector = createAcpReplyProjector({
cfg: createCfg(),
shouldSendToolSummaries: true,
deliver: async (kind, payload) => {
deliveries.push({ kind, text: payload.text });
return true;
},
});
await projector.onEvent({
type: "tool_call",
tag: "tool_call",
toolCallId: "call_ABC123",
status: "in_progress",
text: "call_ABC123 (in_progress)",
});
expect(deliveries[0]?.text).toContain("Tool Call");
expect(deliveries[0]?.text).not.toContain("call_ABC123 (");
});
it("respects metaMode=off and still streams assistant text", async () => {
const deliveries: Array<{ kind: string; text?: string }> = [];
const projector = createAcpReplyProjector({
cfg: createCfg({
@@ -122,24 +229,118 @@ describe("createAcpReplyProjector", () => {
stream: {
coalesceIdleMs: 0,
maxChunkChars: 256,
metaMode: "off",
},
},
}),
shouldSendToolSummaries: true,
provider: "discord",
deliver: async (kind, payload) => {
deliveries.push({ kind, text: payload.text });
return true;
},
});
await projector.onEvent({ type: "text_delta", text: "Hello" });
await projector.onEvent({ type: "text_delta", text: " world" });
await projector.onEvent({ type: "status", text: "running tool" });
await projector.onEvent({
type: "status",
text: "available commands updated",
tag: "available_commands_update",
});
await projector.onEvent({
type: "tool_call",
text: "tool call",
tag: "tool_call",
toolCallId: "x",
status: "in_progress",
});
await projector.onEvent({
type: "text_delta",
text: "hello",
tag: "agent_message_chunk",
});
await projector.flush(true);
expect(deliveries).toEqual([{ kind: "block", text: "hello" }]);
});
it("truncates oversized turns once and emits one truncation notice", async () => {
const deliveries: Array<{ kind: string; text?: string }> = [];
const projector = createAcpReplyProjector({
cfg: createCfg({
acp: {
enabled: true,
stream: {
coalesceIdleMs: 0,
maxChunkChars: 256,
maxTurnChars: 5,
metaMode: "minimal",
},
},
}),
shouldSendToolSummaries: true,
deliver: async (kind, payload) => {
deliveries.push({ kind, text: payload.text });
return true;
},
});
await projector.onEvent({
type: "text_delta",
text: "hello world",
tag: "agent_message_chunk",
});
await projector.onEvent({
type: "text_delta",
text: "ignored tail",
tag: "agent_message_chunk",
});
await projector.flush(true);
expect(deliveries).toEqual([
{ kind: "block", text: "Hello world" },
{ kind: "tool", text: "⚙️ running tool" },
{ kind: "block", text: "hello" },
{ kind: "tool", text: prefixSystemMessage("output truncated") },
]);
});
it("supports tagVisibility overrides for tool updates", async () => {
const deliveries: Array<{ kind: string; text?: string }> = [];
const projector = createAcpReplyProjector({
cfg: createCfg({
acp: {
enabled: true,
stream: {
coalesceIdleMs: 0,
maxChunkChars: 256,
tagVisibility: {
tool_call_update: false,
},
},
},
}),
shouldSendToolSummaries: true,
deliver: async (kind, payload) => {
deliveries.push({ kind, text: payload.text });
return true;
},
});
await projector.onEvent({
type: "tool_call",
tag: "tool_call",
toolCallId: "c1",
status: "in_progress",
title: "Run tests",
text: "Run tests (in_progress)",
});
await projector.onEvent({
type: "tool_call",
tag: "tool_call_update",
toolCallId: "c1",
status: "completed",
title: "Run tests",
text: "Run tests (completed)",
});
expect(deliveries.length).toBe(1);
expect(deliveries[0]?.text).toContain("Tool Call");
});
});

366
src/auto-reply/reply/acp-projector.ts
View File

@@ -1,6 +1,8 @@
import type { AcpRuntimeEvent } from "../../acp/runtime/types.js";
import type { AcpRuntimeEvent, AcpSessionUpdateTag } from "../../acp/runtime/types.js";
import { EmbeddedBlockChunker } from "../../agents/pi-embedded-block-chunker.js";
import { formatToolSummary, resolveToolDisplay } from "../../agents/tool-display.js";
import type { OpenClawConfig } from "../../config/config.js";
import { prefixSystemMessage } from "../../infra/system-message.js";
import type { ReplyPayload } from "../types.js";
import { createBlockReplyPipeline } from "./block-reply-pipeline.js";
import { resolveEffectiveBlockStreamingConfig } from "./block-streaming.js";
@@ -8,8 +10,57 @@ import type { ReplyDispatchKind } from "./reply-dispatcher.js";
const DEFAULT_ACP_STREAM_COALESCE_IDLE_MS = 350;
const DEFAULT_ACP_STREAM_MAX_CHUNK_CHARS = 1800;
const DEFAULT_ACP_META_MODE = "minimal";
const DEFAULT_ACP_SHOW_USAGE = false;
const DEFAULT_ACP_DELIVERY_MODE = "live";
const DEFAULT_ACP_MAX_TURN_CHARS = 24_000;
const DEFAULT_ACP_MAX_TOOL_SUMMARY_CHARS = 320;
const DEFAULT_ACP_MAX_STATUS_CHARS = 320;
const DEFAULT_ACP_MAX_META_EVENTS_PER_TURN = 64;
const ACP_BLOCK_REPLY_TIMEOUT_MS = 15_000;
const ACP_TAG_VISIBILITY_DEFAULTS: Record<string, boolean> = {
agent_message_chunk: true,
tool_call: true,
tool_call_update: true,
usage_update: false,
available_commands_update: false,
current_mode_update: false,
config_option_update: false,
session_info_update: false,
plan: false,
agent_thought_chunk: false,
};
const TERMINAL_TOOL_STATUSES = new Set(["completed", "failed", "cancelled", "done", "error"]);
export type AcpProjectedDeliveryMeta = {
tag?: AcpSessionUpdateTag;
toolCallId?: string;
toolStatus?: string;
allowEdit?: boolean;
};
type AcpDeliveryMode = "live" | "final_only";
type AcpMetaMode = "off" | "minimal" | "verbose";
type AcpProjectionSettings = {
deliveryMode: AcpDeliveryMode;
metaMode: AcpMetaMode;
showUsage: boolean;
maxTurnChars: number;
maxToolSummaryChars: number;
maxStatusChars: number;
maxMetaEventsPerTurn: number;
tagVisibility: Partial<Record<AcpSessionUpdateTag, boolean>>;
};
type ToolLifecycleState = {
started: boolean;
terminal: boolean;
lastRenderedHash?: string;
};
function clampPositiveInteger(
value: unknown,
fallback: number,
@@ -28,6 +79,21 @@ function clampPositiveInteger(
return rounded;
}
function clampBoolean(value: unknown, fallback: boolean): boolean {
return typeof value === "boolean" ? value : fallback;
}
function resolveAcpDeliveryMode(value: unknown): AcpDeliveryMode {
return value === "final_only" ? "final_only" : DEFAULT_ACP_DELIVERY_MODE;
}
function resolveAcpMetaMode(value: unknown): AcpMetaMode {
if (value === "off" || value === "minimal" || value === "verbose") {
return value;
}
return DEFAULT_ACP_META_MODE;
}
function resolveAcpStreamCoalesceIdleMs(cfg: OpenClawConfig): number {
return clampPositiveInteger(
cfg.acp?.stream?.coalesceIdleMs,
@@ -46,6 +112,40 @@ function resolveAcpStreamMaxChunkChars(cfg: OpenClawConfig): number {
});
}
function resolveAcpProjectionSettings(cfg: OpenClawConfig): AcpProjectionSettings {
const stream = cfg.acp?.stream;
return {
deliveryMode: resolveAcpDeliveryMode(stream?.deliveryMode),
metaMode: resolveAcpMetaMode(stream?.metaMode),
showUsage: clampBoolean(stream?.showUsage, DEFAULT_ACP_SHOW_USAGE),
maxTurnChars: clampPositiveInteger(stream?.maxTurnChars, DEFAULT_ACP_MAX_TURN_CHARS, {
min: 1,
max: 500_000,
}),
maxToolSummaryChars: clampPositiveInteger(
stream?.maxToolSummaryChars,
DEFAULT_ACP_MAX_TOOL_SUMMARY_CHARS,
{
min: 64,
max: 8_000,
},
),
maxStatusChars: clampPositiveInteger(stream?.maxStatusChars, DEFAULT_ACP_MAX_STATUS_CHARS, {
min: 64,
max: 8_000,
}),
maxMetaEventsPerTurn: clampPositiveInteger(
stream?.maxMetaEventsPerTurn,
DEFAULT_ACP_MAX_META_EVENTS_PER_TURN,
{
min: 1,
max: 2_000,
},
),
tagVisibility: stream?.tagVisibility ?? {},
};
}
function resolveAcpStreamingConfig(params: {
cfg: OpenClawConfig;
provider?: string;
@@ -60,6 +160,66 @@ function resolveAcpStreamingConfig(params: {
});
}
function truncateText(input: string, maxChars: number): string {
if (input.length <= maxChars) {
return input;
}
if (maxChars <= 1) {
return input.slice(0, maxChars);
}
return `${input.slice(0, maxChars - 1)}…`;
}
function hashText(text: string): string {
return text.trim();
}
function normalizeToolStatus(status: string | undefined): string | undefined {
if (!status) {
return undefined;
}
const normalized = status.trim().toLowerCase();
return normalized || undefined;
}
function isTagVisible(
settings: AcpProjectionSettings,
tag: AcpSessionUpdateTag | undefined,
): boolean {
if (!tag) {
return true;
}
const override = settings.tagVisibility[tag];
if (typeof override === "boolean") {
return override;
}
if (Object.prototype.hasOwnProperty.call(ACP_TAG_VISIBILITY_DEFAULTS, tag)) {
return ACP_TAG_VISIBILITY_DEFAULTS[tag];
}
return true;
}
function renderToolSummaryText(event: Extract<AcpRuntimeEvent, { type: "tool_call" }>): string {
const detailParts: string[] = [];
const title = event.title?.trim();
if (title) {
detailParts.push(title);
}
const status = event.status?.trim();
if (status) {
detailParts.push(`status=${status}`);
}
const fallback = event.text?.trim();
if (detailParts.length === 0 && fallback) {
detailParts.push(fallback);
}
const display = resolveToolDisplay({
name: "tool_call",
meta: detailParts.join(" · ") || "tool call",
});
return formatToolSummary(display);
}
export type AcpReplyProjector = {
onEvent: (event: AcpRuntimeEvent) => Promise<void>;
flush: (force?: boolean) => Promise<void>;
@@ -68,10 +228,15 @@ export type AcpReplyProjector = {
export function createAcpReplyProjector(params: {
cfg: OpenClawConfig;
shouldSendToolSummaries: boolean;
deliver: (kind: ReplyDispatchKind, payload: ReplyPayload) => Promise<boolean>;
deliver: (
kind: ReplyDispatchKind,
payload: ReplyPayload,
meta?: AcpProjectedDeliveryMeta,
) => Promise<boolean>;
provider?: string;
accountId?: string;
}): AcpReplyProjector {
const settings = resolveAcpProjectionSettings(params.cfg);
const streaming = resolveAcpStreamingConfig({
cfg: params.cfg,
provider: params.provider,
@@ -86,7 +251,28 @@ export function createAcpReplyProjector(params: {
});
const chunker = new EmbeddedBlockChunker(streaming.chunking);
let emittedTurnChars = 0;
let emittedMetaEvents = 0;
let truncationNoticeEmitted = false;
let lastStatusHash: string | undefined;
let lastToolHash: string | undefined;
let lastUsageTuple: string | undefined;
const toolLifecycleById = new Map<string, ToolLifecycleState>();
const resetTurnState = () => {
emittedTurnChars = 0;
emittedMetaEvents = 0;
truncationNoticeEmitted = false;
lastStatusHash = undefined;
lastToolHash = undefined;
lastUsageTuple = undefined;
toolLifecycleById.clear();
};
const drainChunker = (force: boolean) => {
if (settings.deliveryMode === "final_only" && !force) {
return;
}
chunker.drain({
force,
emit: (chunk) => {
@@ -100,13 +286,132 @@ export function createAcpReplyProjector(params: {
await blockReplyPipeline.flush({ force });
};
const emitToolSummary = async (prefix: string, text: string): Promise<void> => {
if (!params.shouldSendToolSummaries || !text) {
const consumeMetaQuota = (force: boolean): boolean => {
if (force) {
return true;
}
if (emittedMetaEvents >= settings.maxMetaEventsPerTurn) {
return false;
}
emittedMetaEvents += 1;
return true;
};
const emitSystemStatus = async (
text: string,
meta?: AcpProjectedDeliveryMeta,
opts?: { force?: boolean; dedupe?: boolean },
) => {
if (!params.shouldSendToolSummaries) {
return;
}
// Keep tool summaries ordered after any pending streamed text.
await flush(true);
await params.deliver("tool", { text: `${prefix} ${text}` });
if (settings.metaMode === "off" && opts?.force !== true) {
return;
}
const bounded = truncateText(text.trim(), settings.maxStatusChars);
if (!bounded) {
return;
}
const formatted = prefixSystemMessage(bounded);
const hash = hashText(formatted);
const shouldDedupe = opts?.dedupe !== false;
if (shouldDedupe && lastStatusHash === hash) {
return;
}
if (!consumeMetaQuota(opts?.force === true)) {
return;
}
if (settings.deliveryMode === "live") {
await flush(true);
}
await params.deliver("tool", { text: formatted }, meta);
lastStatusHash = hash;
};
const emitToolSummary = async (
event: Extract<AcpRuntimeEvent, { type: "tool_call" }>,
opts?: { force?: boolean },
) => {
if (!params.shouldSendToolSummaries || settings.metaMode === "off") {
return;
}
if (!isTagVisible(settings, event.tag)) {
return;
}
const toolSummary = truncateText(renderToolSummaryText(event), settings.maxToolSummaryChars);
const hash = hashText(toolSummary);
const toolCallId = event.toolCallId?.trim() || undefined;
const status = normalizeToolStatus(event.status);
const isTerminal = status ? TERMINAL_TOOL_STATUSES.has(status) : false;
const isStart = status === "in_progress" || event.tag === "tool_call";
if (settings.metaMode === "verbose") {
if (lastToolHash === hash) {
return;
}
} else if (settings.metaMode === "minimal") {
if (toolCallId) {
const state = toolLifecycleById.get(toolCallId) ?? {
started: false,
terminal: false,
};
if (isTerminal && state.terminal) {
return;
}
if (isStart && state.started) {
return;
}
if (state.lastRenderedHash === hash) {
return;
}
if (isStart) {
state.started = true;
}
if (isTerminal) {
state.terminal = true;
}
state.lastRenderedHash = hash;
toolLifecycleById.set(toolCallId, state);
} else if (lastToolHash === hash) {
return;
}
}
if (!consumeMetaQuota(opts?.force === true)) {
return;
}
if (settings.deliveryMode === "live") {
await flush(true);
}
await params.deliver(
"tool",
{ text: toolSummary },
{
...(event.tag ? { tag: event.tag } : {}),
...(toolCallId ? { toolCallId } : {}),
...(status ? { toolStatus: status } : {}),
allowEdit: Boolean(toolCallId && event.tag === "tool_call_update"),
},
);
lastToolHash = hash;
};
const emitTruncationNotice = async () => {
if (truncationNoticeEmitted) {
return;
}
truncationNoticeEmitted = true;
await emitSystemStatus(
"output truncated",
{
tag: "session_info_update",
},
{
force: true,
dedupe: false,
},
);
};
const onEvent = async (event: AcpRuntimeEvent): Promise<void> => {
@@ -114,22 +419,61 @@ export function createAcpReplyProjector(params: {
if (event.stream && event.stream !== "output") {
return;
}
if (event.text) {
chunker.append(event.text);
if (!isTagVisible(settings, event.tag)) {
return;
}
const text = event.text;
if (!text) {
return;
}
if (emittedTurnChars >= settings.maxTurnChars) {
await emitTruncationNotice();
return;
}
const remaining = settings.maxTurnChars - emittedTurnChars;
const accepted = remaining < text.length ? text.slice(0, remaining) : text;
if (accepted.length > 0) {
chunker.append(accepted);
emittedTurnChars += accepted.length;
drainChunker(false);
}
if (accepted.length < text.length) {
await emitTruncationNotice();
}
return;
}
if (event.type === "status") {
await emitToolSummary("⚙️", event.text);
if (!isTagVisible(settings, event.tag)) {
return;
}
if (event.tag === "usage_update") {
if (!settings.showUsage) {
return;
}
const usageTuple =
typeof event.used === "number" && typeof event.size === "number"
? `${event.used}/${event.size}`
: hashText(event.text);
if (usageTuple === lastUsageTuple) {
return;
}
lastUsageTuple = usageTuple;
}
await emitSystemStatus(event.text, event.tag ? { tag: event.tag } : undefined, {
dedupe: true,
});
return;
}
if (event.type === "tool_call") {
await emitToolSummary("🧰", event.text);
await emitToolSummary(event);
return;
}
if (event.type === "done" || event.type === "error") {
await flush(true);
resetTurnState();
}
};

81
src/auto-reply/reply/dispatch-acp.ts
View File

@@ -11,6 +11,7 @@ import { readAcpSessionEntry } from "../../acp/runtime/session-meta.js";
import type { OpenClawConfig } from "../../config/config.js";
import type { TtsAutoMode } from "../../config/types.tts.js";
import { logVerbose } from "../../globals.js";
import { runMessageAction } from "../../infra/outbound/message-action-runner.js";
import { getSessionBindingService } from "../../infra/outbound/session-binding-service.js";
import { generateSecureUuid } from "../../infra/secure-random.js";
import { prefixSystemMessage } from "../../infra/system-message.js";
@@ -157,6 +158,7 @@ export async function tryDispatchAcpReply(params: {
originatingTo?: string;
shouldSendToolSummaries: boolean;
bypassForCommand: boolean;
onReplyStart?: () => Promise<void> | void;
recordProcessed: DispatchProcessedRecorder;
markIdle: (reason: string) => void;
}): Promise<AcpDispatchAttemptResult | null> {
@@ -182,9 +184,69 @@ export async function tryDispatchAcpReply(params: {
let queuedFinal = false;
let acpAccumulatedBlockText = "";
let acpBlockCount = 0;
let startedReplyLifecycle = false;
const toolUpdateMessageById = new Map<
string,
{
channel: string;
accountId?: string;
to: string;
threadId?: string | number;
messageId: string;
}
>();
const ensureReplyLifecycleStarted = async () => {
if (startedReplyLifecycle) {
return;
}
startedReplyLifecycle = true;
await params.onReplyStart?.();
};
const tryEditToolUpdate = async (payload: ReplyPayload, toolCallId: string): Promise<boolean> => {
if (!params.shouldRouteToOriginating || !params.originatingChannel || !params.originatingTo) {
return false;
}
const handle = toolUpdateMessageById.get(toolCallId);
if (!handle?.messageId) {
return false;
}
const message = payload.text?.trim();
if (!message) {
return false;
}
try {
await runMessageAction({
cfg: params.cfg,
action: "edit",
params: {
channel: handle.channel,
accountId: handle.accountId,
to: handle.to,
threadId: handle.threadId,
messageId: handle.messageId,
message,
},
sessionKey: params.ctx.SessionKey,
});
routedCounts.tool += 1;
return true;
} catch (error) {
logVerbose(
`dispatch-acp: tool message edit failed for ${toolCallId}: ${error instanceof Error ? error.message : String(error)}`,
);
return false;
}
};
const deliverAcpPayload = async (
kind: ReplyDispatchKind,
payload: ReplyPayload,
meta?: {
toolCallId?: string;
allowEdit?: boolean;
},
): Promise<boolean> => {
if (kind === "block" && payload.text?.trim()) {
if (acpAccumulatedBlockText.length > 0) {
@@ -193,6 +255,9 @@ export async function tryDispatchAcpReply(params: {
acpAccumulatedBlockText += payload.text;
acpBlockCount += 1;
}
if ((payload.text?.trim() ?? "").length > 0 || payload.mediaUrl || payload.mediaUrls?.length) {
await ensureReplyLifecycleStarted();
}
const ttsPayload = await maybeApplyTtsToPayload({
payload,
@@ -204,6 +269,13 @@ export async function tryDispatchAcpReply(params: {
});
if (params.shouldRouteToOriginating && params.originatingChannel && params.originatingTo) {
const toolCallId = meta?.toolCallId?.trim();
if (kind === "tool" && meta?.allowEdit === true && toolCallId) {
const edited = await tryEditToolUpdate(ttsPayload, toolCallId);
if (edited) {
return true;
}
}
const result = await routeReply({
payload: ttsPayload,
channel: params.originatingChannel,
@@ -219,6 +291,15 @@ export async function tryDispatchAcpReply(params: {
);
return false;
}
if (kind === "tool" && meta?.toolCallId && result.messageId) {
toolUpdateMessageById.set(meta.toolCallId, {
channel: params.originatingChannel,
accountId: params.ctx.AccountId,
to: params.originatingTo,
...(params.ctx.MessageThreadId != null ? { threadId: params.ctx.MessageThreadId } : {}),
messageId: result.messageId,
});
}
routedCounts[kind] += 1;
return true;
}

1
src/auto-reply/reply/dispatch-from-config.ts
View File

@@ -371,6 +371,7 @@ export async function dispatchReplyFromConfig(params: {
originatingTo,
shouldSendToolSummaries,
bypassForCommand: bypassAcpForCommand,
onReplyStart: params.replyOptions?.onReplyStart,
recordProcessed,
markIdle,
});