vrr/openclaw
2
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-10-03 09:39:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 30

Keep connected apps usable and Slack previews quiet

Browse source 
* fix(codex): keep apps available with stale approval overrides

* fix(slack): wait for a complete first preamble before notifying

* test(agents): protect intentional silence after reactions

* test: match app inventory and draft stream fixture contracts

* refactor(codex): filter approval overrides in one pass

* test: preserve prior QA outputs in declaration cache relocation

* Preserve asynchronous completion ownership for silent replies

* Assert the intermediate notification checks actually execute

* Preserve complete preambles across embedded and CLI delivery

* test: exercise real commentary completion delivery
This commit is contained in:
pash-openai 2026-09-02 01:39:49 -07:00 • committed by GitHub
parent 5dd1bffd31
commit a2bbcbf406
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 589 additions and 166 deletions
Download patch file Download diff file Expand all files Collapse all files

2
docs/channels/slack.md
View file

@ -1560,6 +1560,8 @@ Set `channels.slack.streaming.progress.nativeTaskCards` to `false` to fall back
Set `channels.slack.streaming.progress.style` to `"compact"` for one plain-text progress draft instead of either card surface. This is also the default when `progress.toolProgress` is `false` and no style is selected. With the other progress controls below, only authored commentary appears as italic text: plan checklists and tool-status lines never replace it. An eligible final text answer replaces that same Slack message. Set `style: "card"` explicitly to retain a card while hiding tool progress.
For streamed preambles, Slack waits for the first complete preamble before creating the message, so its notification contains the full thought rather than a single token. Once that message exists, later preambles can stream as edits without another notification.
```json5
{
channels: {

27
extensions/codex/src/app-server/event-projector-assistant.ts
View file

@ -26,7 +26,7 @@ export class CodexAssistantProjection {
private latestTerminalAssistantCandidateSuperseded = false;
private terminalAssistantCandidateEarlierActiveItemIds = new Set<string>();
private pendingRawTerminalAssistantEchoItemId: string | undefined;
private readonly lastCommentaryProgressTextByItem = new Map<string, string>();
private readonly lastCommentaryProgressEventByItem = new Map<string, string>();
private readonly lastAnswerCandidateEventByItem = new Map<string, string>();
private visibleAnswerCandidateItemId: string | undefined;
// Codex emits each typed item completion before its matching raw response item.
@ -93,7 +93,7 @@ export class CodexAssistantProjection {
return;
}
if (isCommentary) {
this.emitCommentaryProgress({ itemId, text });
this.emitCommentaryProgress({ itemId, text, phase: "update" });
return;
}
if (this.isFinalAnswerAssistantItem(itemId)) {
@ -193,7 +193,7 @@ export class CodexAssistantProjection {
this.rememberAssistantItem(item.id);
this.assistantTextByItem.set(item.id, item.text);
if (item.text && this.isCommentaryAssistantItem(item.id)) {
this.emitCommentaryProgress({ itemId: item.id, text: item.text });
this.emitCommentaryProgress({ itemId: item.id, text: item.text, phase: "end" });
this.pendingRawCommentaryEchoes += 1;
} else if (
item.text &&
@ -294,7 +294,7 @@ export class CodexAssistantProjection {
}
this.rawPromotedAssistantItemIds.add(itemId);
if (phase === "commentary") {
this.emitCommentaryProgress({ itemId, text });
this.emitCommentaryProgress({ itemId, text, phase: "end" });
} else {
this.markLatestTerminalAssistantCandidate(itemId, activeItemIds);
}
@ -505,22 +505,27 @@ export class CodexAssistantProjection {
return this.assistantPhaseByItem.get(itemId) === "final_answer";
}
private emitCommentaryProgress(params: { itemId: string; text: string }): void {
private emitCommentaryProgress(params: {
itemId: string;
text: string;
phase: "update" | "end";
}): void {
const progressText = params.text.trim();
if (
!progressText ||
this.lastCommentaryProgressTextByItem.get(params.itemId) === progressText
) {
// Codex completes an item with the same text as its last delta. Channels
// need that boundary before their first notifying post, so agents must not
// collapse completion into a text-only duplicate or invent a timer instead.
const signature = `${params.phase}\0${progressText}`;
if (!progressText || this.lastCommentaryProgressEventByItem.get(params.itemId) === signature) {
return;
}
this.lastCommentaryProgressTextByItem.set(params.itemId, progressText);
this.lastCommentaryProgressEventByItem.set(params.itemId, signature);
this.emitAgentEvent({
stream: "item",
data: {
itemId: params.itemId,
kind: "preamble",
title: "Preamble",
phase: "update",
phase: params.phase,
progressText,
source: "codex-app-server",
},

29
extensions/codex/src/app-server/event-projector.commentary.test.ts
View file

@ -203,7 +203,7 @@ describe("CodexAppServerEventProjector commentary projection", () => {
).toBeUndefined();
});
it("streams commentary agent messages as keyed progress events", async () => {
it("streams commentary with a distinct completion even when the last delta has the same text", async () => {
const onAgentEvent = vi.fn();
const onPartialReply = vi.fn();
const commentaryText = [
@ -238,6 +238,18 @@ describe("CodexAppServerEventProjector commentary projection", () => {
await projector.handleNotification(
agentMessageDelta(commentaryText.slice("Checking".length), "msg-commentary"),
);
// The completion boundary lets channels buffer their first notification;
// text-only dedupe must not erase it after the final identical snapshot.
await projector.handleNotification(
forCurrentTurn("item/completed", {
item: {
type: "agentMessage",
id: "msg-commentary",
phase: "commentary",
text: commentaryText,
},
}),
);
await projector.handleNotification(
turnCompleted([
{
@ -277,6 +289,14 @@ describe("CodexAppServerEventProjector commentary projection", () => {
progressText: commentaryText,
source: "codex-app-server",
},
{
itemId: "msg-commentary",
kind: "preamble",
title: "Preamble",
phase: "end",
progressText: commentaryText,
source: "codex-app-server",
},
]);
const result = projector.buildResult(buildEmptyToolTelemetry());
@ -431,7 +451,10 @@ describe("CodexAppServerEventProjector commentary projection", () => {
.map((call) => call[0])
.filter((event) => event.stream === "item" && event.data.kind === "preamble");
expect(preambles.map((event) => event.data.progressText)).toEqual(["Checking the workspace"]);
expect(preambles.map((event) => [event.data.phase, event.data.progressText])).toEqual([
["update", "Checking the workspace"],
["end", "Checking the workspace"],
]);
expect(preambles.every((event) => event.data.itemId === "msg-commentary")).toBe(true);
});
@ -505,9 +528,11 @@ describe("CodexAppServerEventProjector commentary projection", () => {
.filter((event) => event.stream === "item" && event.data.kind === "preamble");
expect(preambles.map((event) => event.data.itemId)).toEqual([
"msg-commentary",
"msg-commentary",
"raw-assistant-2",
]);
expect(preambles.map((event) => event.data.phase)).toEqual(["update", "end", "end"]);
});
it("pairs a raw commentary echo after a rewritten typed completion", async () => {

33
extensions/codex/src/app-server/plugin-inventory.ts
View file

@ -66,8 +66,8 @@ export type CodexPluginOwnedApp = {
accessible: boolean;
enabled: boolean;
needsAuth: boolean;
/** Tool config keys Codex explicitly classifies as read-only. */
readOnlyToolConfigKeys?: readonly string[];
/** Current non-read-only tool keys; absent when Codex omits tool metadata. */
approvalOverrideToolConfigKeys?: readonly string[];
};
/** Inventory record for one configured Codex plugin policy. */
@ -533,29 +533,28 @@ function resolveOwnedApps(params: {
// app/read metadata is the canonical connector access proof.
needsAuth: !info.isAccessible,
},
resolveOwnedAppReadOnlyToolConfigKeys(info),
resolveOwnedAppApprovalOverrideKeys(info),
);
})
.toSorted((left, right) => left.id.localeCompare(right.id));
}
/** Returns the config keys that Codex metadata proves belong to read-only tools. */
export function resolveOwnedAppReadOnlyToolConfigKeys(
/** Returns current tool keys whose overrides could bypass the requested reviewer. */
export function resolveOwnedAppApprovalOverrideKeys(
app: v2.AppInfo,
): Pick<CodexPluginOwnedApp, "readOnlyToolConfigKeys"> {
): Pick<CodexPluginOwnedApp, "approvalOverrideToolConfigKeys"> {
if (!app.toolSummaries) {
return {};
}
const appName = app.name.trim();
const appNameLower = appName.toLowerCase();
const tools = app.toolSummaries ?? [];
const writableToolConfigKeys = new Set(
tools
.filter((tool) => !tool.isReadOnly)
.flatMap((tool) => resolveAppToolConfigKeys({ appName, appNameLower, tool })),
);
const keys = tools
.filter((tool) => tool.isReadOnly)
.flatMap((tool) => resolveAppToolConfigKeys({ appName, appNameLower, tool }))
.filter((key) => !writableToolConfigKeys.has(key));
return keys.length > 0 ? { readOnlyToolConfigKeys: Array.from(new Set(keys)).toSorted() } : {};
// Agents: app/read includes disabled tools. Keep every non-read-only alias,
// including collisions with read-only titles; retired names cannot authorize
// a current tool and must not prevent the entire app from being admitted.
const keys = app.toolSummaries
.filter((tool) => !tool.isReadOnly)
.flatMap((tool) => resolveAppToolConfigKeys({ appName, appNameLower, tool }));
return { approvalOverrideToolConfigKeys: Array.from(new Set(keys)).toSorted() };
}
function resolveAppToolConfigKeys(params: {

4
extensions/codex/src/app-server/plugin-thread-app-admission.ts
View file

@ -7,7 +7,7 @@ import {
} from "./app-inventory-cache.js";
import type { ResolvedCodexPluginsPolicy } from "./config.js";
import {
resolveOwnedAppReadOnlyToolConfigKeys,
resolveOwnedAppApprovalOverrideKeys,
type CodexPluginInventory,
type CodexPluginInventoryRecord,
type CodexPluginOwnedApp,
@ -188,7 +188,7 @@ export function toCodexPluginOwnedAccountApp(app: v2.AppInfo): CodexPluginOwnedA
accessible: app.isAccessible,
enabled: app.isEnabled,
needsAuth: !app.isAccessible,
...resolveOwnedAppReadOnlyToolConfigKeys(app),
...resolveOwnedAppApprovalOverrideKeys(app),
};
}

229
extensions/codex/src/app-server/plugin-thread-config.test.ts
View file

@ -7,7 +7,7 @@ import {
CODEX_PLUGINS_WORKSPACE_MARKETPLACE_NAME,
} from "./config.js";
import {
resolveOwnedAppReadOnlyToolConfigKeys,
resolveOwnedAppApprovalOverrideKeys,
resolveRecoverableCodexPluginConfigKeys,
} from "./plugin-inventory.js";
import { CodexPluginMetadataCache } from "./plugin-metadata-cache.js";
@ -28,7 +28,34 @@ describe("Codex plugin thread config", () => {
defaultCodexAppInventoryCache.clear();
});
it("does not classify keys shared with writable tools as read-only", () => {
it("keeps approval checks conservative when tool metadata is absent", () => {
expect(resolveOwnedAppApprovalOverrideKeys(appInfo("linear", true))).toStrictEqual({});
expect(
resolveOwnedAppApprovalOverrideKeys({ ...appInfo("linear", true), toolSummaries: [] }),
).toStrictEqual({ approvalOverrideToolConfigKeys: [] });
});
it("retains disabled writable tools in the approval boundary", () => {
expect(
resolveOwnedAppApprovalOverrideKeys({
...appInfo("linear", true),
toolSummaries: [
{
name: "save_issue",
title: "Save issue",
description: "Create or update an issue.",
isEnabled: false,
disabledReason: "App policy",
isReadOnly: false,
},
],
}),
).toStrictEqual({
approvalOverrideToolConfigKeys: ["Save issue", "linear_save_issue", "save_issue"],
});
});
it("preserves writable approval checks for keys shared with read-only tools", () => {
const app: v2.AppInfo = {
...appInfo("linear", true),
toolSummaries: [
@ -51,8 +78,8 @@ describe("Codex plugin thread config", () => {
],
};
expect(resolveOwnedAppReadOnlyToolConfigKeys(app)).toStrictEqual({
readOnlyToolConfigKeys: ["Fetch", "fetch"],
expect(resolveOwnedAppApprovalOverrideKeys(app)).toStrictEqual({
approvalOverrideToolConfigKeys: ["Save issue", "linear_fetch", "linear_linear_fetch"],
});
});
@ -700,100 +727,122 @@ describe("Codex plugin thread config", () => {
expect(configPatch).not.toHaveProperty("approvals_reviewer");
});
it("keeps ask policy apps when managed approval overrides cover only read-only tools", async () => {
const appCache = new CodexAppInventoryCache();
const linearApp: v2.AppInfo = {
...appInfo("linear", true),
toolSummaries: [
{
name: "fetch",
title: "Fetch",
description: "Fetch a Linear issue.",
isEnabled: true,
disabledReason: null,
isReadOnly: true,
},
{
name: "save_issue",
title: "linear/save_issue",
description: "Create or update a Linear issue.",
isEnabled: true,
disabledReason: null,
isReadOnly: false,
},
],
};
await appCache.refreshNow({
key: "runtime",
nowMs: 0,
request: async (method, params) => codexAppInventoryResponse(method, [linearApp], params),
});
const request = vi.fn(async (method: string, params?: unknown) => {
if (method === "plugin/installed" || method === "plugin/list") {
return pluginList([pluginSummary("linear", { installed: true, enabled: true })]);
}
if (method === "plugin/read") {
return pluginDetail("linear", [appSummary("linear")], ["linear"]);
}
if (method === "config/read") {
expect(params).toEqual({ includeLayers: true, cwd: "/repo/project" });
return {
config: {
apps: {
linear: {
tools: {
linear_fetch: { approval_mode: "approve" },
it.each(
[false, true].flatMap((allowAllPlugins) =>
[
{ name: "read-only", tools: { linear_fetch: { approval_mode: "approve" } } },
{ name: "cleared", tools: { linear_save_issue: { approval_mode: null } } },
{ name: "retired", tools: { linear_retired_tool: { approval_mode: "approve" } } },
].map(({ name, tools }) => ({ name, tools, allowAllPlugins })),
),
)(
"keeps ask policy apps with $name overrides (account-wide: $allowAllPlugins)",
async ({ tools, allowAllPlugins }) => {
const appCache = new CodexAppInventoryCache();
const linearApp: v2.AppInfo = {
...appInfo("linear", true),
toolSummaries: [
{
name: "fetch",
title: "Fetch",
description: "Fetch a Linear issue.",
isEnabled: true,
disabledReason: null,
isReadOnly: true,
},
{
name: "save_issue",
title: "linear/save_issue",
description: "Create or update a Linear issue.",
isEnabled: true,
disabledReason: null,
isReadOnly: false,
},
],
};
await appCache.refreshNow({
key: "runtime",
nowMs: 0,
request: async (method, params) => codexAppInventoryResponse(method, [linearApp], params),
});
const request = vi.fn(async (method: string, params?: unknown) => {
if (method === "app/installed" || method === "app/read") {
return codexAppInventoryResponse(
method,
[linearApp],
// SAFETY: the dispatcher supplies the narrowed inventory method's parameters.
params as CodexAppServerRequestParams<typeof method>,
);
}
if (method === "plugin/installed" || method === "plugin/list") {
return pluginList([pluginSummary("linear", { installed: true, enabled: true })]);
}
if (method === "plugin/read") {
return pluginDetail("linear", [appSummary("linear")], ["linear"]);
}
if (method === "config/read") {
expect(params).toEqual({ includeLayers: true, cwd: "/repo/project" });
return {
config: {
apps: {
linear: {
// Managed defaults can outlive a tool or remain after a local
// null/delete. Neither state may make the whole app disappear.
tools,
},
},
},
},
layers: [],
};
}
throw new Error(`unexpected request ${method}`);
});
layers: [],
};
}
throw new Error(`unexpected request ${method}`);
});
const config = await buildCodexPluginThreadConfig({
pluginConfig: {
codexPlugins: {
enabled: true,
allow_destructive_actions: "ask",
plugins: {
linear: {
marketplaceName: CODEX_PLUGINS_MARKETPLACE_NAME,
pluginName: "linear",
},
const config = await buildCodexPluginThreadConfig({
pluginConfig: {
codexPlugins: {
enabled: true,
allow_all_plugins: allowAllPlugins,
allow_destructive_actions: "ask",
plugins: allowAllPlugins
? {}
: {
linear: {
marketplaceName: CODEX_PLUGINS_MARKETPLACE_NAME,
pluginName: "linear",
},
},
},
},
},
appCache,
appCacheKey: "runtime",
configCwd: "/repo/project",
nowMs: 1,
request,
});
appCache,
appCacheKey: "runtime",
configCwd: "/repo/project",
nowMs: 1,
request,
});
expect(config.configPatch).toEqual({
apps: {
_default: {
enabled: false,
destructive_enabled: false,
open_world_enabled: false,
expect(config.configPatch).toEqual({
apps: {
_default: {
enabled: false,
destructive_enabled: false,
open_world_enabled: false,
},
linear: {
enabled: true,
approvals_reviewer: "user",
destructive_enabled: true,
open_world_enabled: true,
default_tools_approval_mode: "auto",
},
},
linear: {
enabled: true,
approvals_reviewer: "user",
destructive_enabled: true,
open_world_enabled: true,
default_tools_approval_mode: "auto",
},
},
});
expect(config.provisionalAppIds).toEqual(["linear"]);
expect(config.diagnostics).toStrictEqual([]);
expect(request.mock.calls.filter(([method]) => method === "config/read")).toHaveLength(1);
expect(request.mock.calls.map(([method]) => method)).not.toContain("config/batchWrite");
});
});
expect(config.provisionalAppIds).toEqual(["linear"]);
expect(config.diagnostics).toStrictEqual([]);
expect(request.mock.calls.filter(([method]) => method === "config/read")).toHaveLength(1);
expect(request.mock.calls.map(([method]) => method)).not.toContain("config/batchWrite");
},
);
it("omits ask policy apps when cwd effective approval overrides remain after cleanup", async () => {
const appCache = new CodexAppInventoryCache();

15
extensions/codex/src/app-server/plugin-thread-config.ts
View file

@ -114,7 +114,9 @@ type BuildCodexPluginThreadConfigParams = {
nowMs?: number;
};
const CODEX_PLUGIN_THREAD_CONFIG_INPUT_FINGERPRINT_VERSION = 3;
// Admission changes must rebuild existing bindings too, or an app omitted by
// an older rule stays missing even after the gateway has been upgraded.
const CODEX_PLUGIN_THREAD_CONFIG_INPUT_FINGERPRINT_VERSION = 4;
const CODEX_PLUGIN_THREAD_CONFIG_FINGERPRINT_VERSION = 2;
/** Returns true when plugin config exists and thread config may need app patches. */
@ -674,15 +676,18 @@ function readPersistedAppToolApprovalOverrideNames(
if (!isJsonObject(tools)) {
return [];
}
const keys = app.approvalOverrideToolConfigKeys;
return Object.entries(tools)
.filter(([toolName]) => !app.readOnlyToolConfigKeys?.includes(toolName))
.filter(([, value]) => hasPersistedToolApprovalOverride(value))
.map(([toolName]) => toolName)
.flatMap(([name, value]) =>
(!keys || keys.includes(name)) && hasPersistedToolApprovalOverride(value) ? [name] : [],
)
.toSorted();
}
function hasPersistedToolApprovalOverride(value: JsonValue): boolean {
return isJsonObject(value) && value.approval_mode !== undefined;
// Codex serializes an unset optional approval mode as null. Treating null as
// an override turns a successful cleanup into an app-wide admission failure.
return isJsonObject(value) && value.approval_mode !== undefined && value.approval_mode !== null;
}
function quoteConfigKeyPathSegment(segment: string): string {

114
extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts
View file

@ -307,7 +307,7 @@ function createDraftStreamStub() {
await editFinal();
return true;
}),
messageId: () => "171234.567",
messageId: (): string | undefined => "171234.567",
channelId: () => "C123",
};
}
@ -4122,6 +4122,118 @@ describe("dispatchPreparedSlackMessage preview fallback", () => {
await requireCapturedItemEventHandler()({ progressText: "hidden progress" });
});
it.each([undefined, "compact"] as const)(
"buffers the first notifying preamble but streams later edits (style=%s)",
async (style) => {
const checkpoint = vi.fn();
let postedMessageId: string | undefined;
const draftStream = {
...createDraftStreamStub(),
messageId: () => postedMessageId,
};
draftStream.flush.mockImplementation(async () => {
if (draftStream.update.mock.calls.length > 0) {
postedMessageId = "171234.567";
}
});
createSlackDraftStreamMock.mockReturnValueOnce(draftStream);
finalizeSlackPreviewEditMock.mockResolvedValueOnce(undefined);
mockedSlackStreamingMode = "progress";
mockedDispatchSequence = [{ kind: "final", payload: { text: FINAL_REPLY_TEXT } }];
mockedReplyOptionEvents = [
{ kind: "item", itemKind: "preamble", itemId: "p1", phase: "update", progressText: "I" },
{
kind: "checkpoint",
run: async () => {
// Even a timer/flush must not post the first token: Slack freezes
// its push notification at creation, then edits do not re-notify.
checkpoint();
await draftStream.flush();
expect(draftStream.update).not.toHaveBeenCalled();
expect(postedMessageId).toBeUndefined();
},
},
{
kind: "item",
itemKind: "preamble",
itemId: "p1",
phase: "update",
progressText: "I will check the result.",
},
{
kind: "checkpoint",
run: async () => {
checkpoint();
expect(draftStream.update).not.toHaveBeenCalled();
},
},
{
kind: "item",
itemKind: "preamble",
itemId: "p1",
phase: "end",
progressText: "I will check the result.",
},
{
kind: "checkpoint",
run: async () => {
checkpoint();
expect(postedMessageId).toBe("171234.567");
expect(draftUpdateTexts(draftStream)).toEqual(["_I will check the result._"]);
},
},
{
kind: "item",
itemKind: "preamble",
itemId: "p2",
phase: "update",
progressText: "The result",
},
{
kind: "checkpoint",
run: async () => {
checkpoint();
expectLastDraftUpdateText(draftStream, "_The result_");
},
},
{
kind: "item",
itemKind: "preamble",
itemId: "p2",
phase: "end",
progressText: "The result is ready.",
},
];
await dispatchPreparedSlackMessage(
createPreparedSlackMessage({
accountConfig: {
streaming: {
mode: "progress",
progress: {
style,
label: false,
commentary: true,
toolProgress: false,
maxLines: 1,
},
},
},
}),
);
// Assert the intermediate observations ran; the final text alone cannot
// prove that Slack never received a first-token notification.
expect(checkpoint).toHaveBeenCalledTimes(4);
expectLastDraftUpdateText(draftStream, "_The result is ready._");
expectMockCallArgFields(finalizeSlackPreviewEditMock, 0, "same-message final edit", {
messageId: "171234.567",
text: FINAL_REPLY_TEXT,
});
expect(deliverRepliesMock).not.toHaveBeenCalled();
},
);
it("keeps only the latest Slack commentary when tool progress is disabled", async () => {
const draftStream = createDraftStreamStub();
createSlackDraftStreamMock.mockReturnValueOnce(draftStream);

11
extensions/slack/src/monitor/message-handler/dispatch.ts
View file

@ -482,6 +482,17 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
return await progress.progressDraft.pushToolEvent(payload);
},
onItemEvent: async (payload) => {
// Slack freezes notification text on the first post. Keep incomplete
// preambles out of the compositor until a message actually exists;
// later edits may stream. A timer or tool event must not flush "I".
if (
payload.kind === "preamble" &&
(payload.phase === "start" || payload.phase === "update") &&
!draftStream?.messageId() &&
!delivery.streamSession?.delivered
) {
return false;
}
if (progress.isProgressMode && payload.kind === "preamble") {
if (progress.shouldYieldDraftProgress()) {
return false;

28
src/agents/cli-runner/execute-events.tool-result-args.test.ts
View file

@ -69,6 +69,34 @@ function collectToolEvents(runId: string): {
}
describe("cli tool result events", () => {
it("emits complete CLI commentary as a completed preamble", () => {
const runId = "run-commentary-complete";
const handlers = createCliEventHandlers({
context: buildContext(runId),
toolTracking: buildToolTracking(),
getRunState: () => ({ failed: false, error: undefined }),
});
const events: AgentEventRuntimePayload[] = [];
const dispose = onAgentEvent((event) => {
if (event.runId === runId && event.stream === "item") {
events.push(event);
}
});
try {
// The JSONL parser has already accumulated this whole pre-tool segment.
// An update-only event would leave first-notification buffering waiting forever.
handlers.emitCliCommentaryText("Let me check that for you.");
expect(events).toMatchObject([
{
stream: "item",
data: { kind: "preamble", phase: "end", progressText: "Let me check that for you." },
},
]);
} finally {
dispose();
}
});
it("keeps correlated result args without adding them to display results", () => {
const runId = "run-tool-result-args";
const handlers = createCliEventHandlers({

4
src/agents/cli-runner/execute-events.ts
View file

@ -309,7 +309,9 @@ export function createCliEventHandlers(params: {
data: {
kind: "preamble",
itemId: `commentary-${runParams.runId}-${commentaryCounter}`,
phase: "update",
// The JSONL parser flushes a complete pre-tool text segment here.
// Mark its boundary so channels can safely create their first notification.
phase: "end",
title: "commentary",
status: "running",
progressText: applyPluginTextReplacements(

108
src/agents/embedded-agent-runner/run/incomplete-turn-recovery.test.ts
View file

@ -8,6 +8,7 @@ import {
resolveReasoningOnlyRetryInstruction,
shouldTreatEmptyAssistantReplyAsSilent,
} from "./incomplete-turn-recovery.js";
import { resolveIncompleteTurnPayloadText } from "./incomplete-turn-resolution.js";
const EMPTY_RESPONSE_RETRY_INSTRUCTION =
"The previous attempt did not produce a user-visible answer. Continue from the current state and produce the visible answer now. Do not restart from scratch.";
@ -31,6 +32,113 @@ function emptyAttempt(assistant = emptyAssistant()) {
}
describe("incomplete-turn recovery policy", () => {
it.each(["required", "optional"] as const)(
"keeps async-owned work out of completed silence (reply=%s)",
(terminalReplyExpectation) => {
const assistant = emptyAssistant({ content: [{ type: "text", text: "NO_REPLY" }] });
const attempt = makeEmbeddedRunnerAttempt({
assistantTexts: ["NO_REPLY"],
lastAssistant: assistant,
currentAttemptAssistant: assistant,
toolMetas: [{ toolName: "image_generate", asyncStarted: true, replaySafe: false }],
replayMetadata: { hadPotentialSideEffects: true, replaySafe: false },
currentAttemptReplayMetadata: { hadPotentialSideEffects: true, replaySafe: false },
});
const state = { payloadCount: 0, aborted: false, timedOut: false, attempt };
// Silence must not steal completion ownership from background work. Nor
// may it trigger a replay or a spurious warning while that owner continues.
expect(
shouldTreatEmptyAssistantReplyAsSilent({
...state,
allowEmptyAssistantReplyAsSilent: true,
onlyExplicitSilentReply: true,
terminalReplyExpectation,
}),
).toBe(false);
expect(resolveEmptyResponseRetryInstruction(state)).toBeNull();
expect(resolveIncompleteTurnPayloadText({ ...state, externalAbort: false })).toBeNull();
},
);
it.each([
{
name: "a completed reaction",
aborted: false,
timedOut: false,
yielded: false,
error: false,
silent: true,
},
{
name: "a failed reaction",
aborted: false,
timedOut: false,
yielded: false,
error: true,
silent: false,
},
{
name: "an aborted turn",
aborted: true,
timedOut: false,
yielded: false,
error: false,
silent: false,
},
{
name: "a timed-out turn",
aborted: false,
timedOut: true,
yielded: false,
error: false,
silent: false,
},
{
name: "pending work",
aborted: false,
timedOut: false,
yielded: true,
error: false,
silent: false,
},
])(
"classifies explicit silence after $name without replaying tools",
({ aborted, timedOut, yielded, error, silent }) => {
const assistant = emptyAssistant({ content: [{ type: "text", text: "NO_REPLY" }] });
const attempt = makeEmbeddedRunnerAttempt({
assistantTexts: ["NO_REPLY"],
lastAssistant: assistant,
currentAttemptAssistant: assistant,
toolMetas: [{ toolName: "message", meta: "react", replaySafe: false, isError: error }],
replayMetadata: { hadPotentialSideEffects: true, replaySafe: false },
currentAttemptReplayMetadata: { hadPotentialSideEffects: true, replaySafe: false },
...(yielded ? { yieldDetected: true } : {}),
...(error ? { lastToolError: { toolName: "message", error: "reaction failed" } } : {}),
});
// A user-triggered turn can intentionally end with only a reaction. Do not
// conflate permission to stay silent with permission to replay that effect.
expect(
shouldTreatEmptyAssistantReplyAsSilent({
allowEmptyAssistantReplyAsSilent: true,
terminalReplyExpectation: "required",
onlyExplicitSilentReply: true,
payloadCount: 0,
aborted,
timedOut,
attempt,
}),
).toBe(silent);
expect(
resolveEmptyResponseRetryInstruction({
payloadCount: 0,
aborted,
timedOut,
attempt,
}),
).toBeNull();
},
);
it.each([
{
name: "zero-token Anthropic stop",

9
src/agents/embedded-agent-runner/run/incomplete-turn-recovery.ts
View file

@ -89,7 +89,7 @@ function shouldSkipNonVisibleTurnRetry(params: {
aborted: boolean;
timedOut: boolean;
attempt: IncompleteTurnAttempt;
/** Reply-optional silent classification tolerates committed side effects; retries never can. */
/** Silent classification can tolerate completed effects, never unfinished work or replay. */
tolerateSideEffects?: boolean;
}): boolean {
return Boolean(
@ -101,6 +101,7 @@ function shouldSkipNonVisibleTurnRetry(params: {
params.attempt.didSendDeterministicApprovalPrompt ||
params.attempt.lastToolError ||
hasAcceptedSessionSpawn(params.attempt.acceptedSessionSpawns) ||
hasAsyncActivity(params.attempt.toolMetas) ||
(params.tolerateSideEffects !== true && params.attempt.replayMetadata.hadPotentialSideEffects),
);
}
@ -115,8 +116,10 @@ export function shouldTreatEmptyAssistantReplyAsSilent(params: {
timedOut: boolean;
attempt: IncompleteTurnAttempt;
}): boolean {
// Optional runs owe no reply. An explicit silent reply also closes a
// successful side-effecting tool turn; replaying it could repeat the effect.
// NO_REPLY is an authored outcome, not missing output: a successful reaction
// can be the entire reply. Agents: classify it before the side-effect retry
// guard, or it becomes a false missing-summary warning (or a repeated tool).
// Actual failures, aborts and pending work still pass through the guards below.
const terminalReplyOptional = params.terminalReplyExpectation === "optional";
const assistant = resolveCurrentAttemptAssistant(params.attempt);
const explicitSilentReply =

2
src/agents/embedded-agent-subscribe.handlers.messages.lifecycle.ts
View file

@ -87,7 +87,7 @@ export function handleMessageEnd(
rawText: coerceChatContentText(extractEmbeddedAssistantText(assistantMessage)),
rawThinking: extractAssistantThinking(assistantMessage),
}));
emitAssistantCommentaryStreamData(ctx, assistantMessage);
emitAssistantCommentaryStreamData(ctx, assistantMessage, true);
// Commentary-tagged tool turns can still carry durable reasoning under /reasoning on.
const suppressedTrimmedReasoning = ctx.state.includeReasoning
? extractAssistantThinking(assistantMessage).trim()

4
src/agents/embedded-agent-subscribe.handlers.messages.stream.ts
View file

@ -212,6 +212,7 @@ export function scopeAssistantMessageToStreamBlock(
export function emitAssistantCommentaryStreamData(
ctx: EmbeddedAgentSubscribeContext,
message: AssistantMessage,
finalMessage = false,
) {
const isResponsesCommentary = isResponsesApiAssistantMessage(message);
const { lastAssistantStreamContentIndex: index, lastAssistantStreamItemId: itemId } = ctx.state;
@ -220,7 +221,7 @@ export function emitAssistantCommentaryStreamData(
? scopeAssistantMessageToStreamBlock(message, index, itemId)
: message;
const text = extractAssistantCommentaryText(commentaryMessage);
if (text && (!isResponsesCommentary || ctx.state.deltaBuffer !== text)) {
if (text && (finalMessage || !isResponsesCommentary || ctx.state.deltaBuffer !== text)) {
// Generic commentary must carry the identity the phase tagger generated so
// the Control UI can key the live row to the persisted fallback row; without
// it every generic segment is unkeyed and survives as a duplicate.
@ -234,6 +235,7 @@ export function emitAssistantCommentaryStreamData(
phase: "commentary",
itemId: commentaryItemId,
}),
{ finalMessage },
);
}
}

63
src/agents/embedded-agent-subscribe.handlers.messages.update-stream-items.test.ts
View file

@ -10,6 +10,7 @@ import {
createOpenAiResponsesPartial,
createOpenAiResponsesTextEvent as createTextUpdateEvent,
} from "./embedded-agent-subscribe.openai-responses.test-helpers.js";
import { createReplyDelivery } from "./embedded-agent-subscribe.reply-delivery.js";
describe("handleMessageUpdate text signatures", () => {
it("emits the full incrementally extracted reasoning value on every delta", async () => {
@ -293,9 +294,12 @@ describe("handleMessageUpdate text signatures", () => {
"openclaw-openai-responses-transport",
"openclaw-openai-chatgpt-responses-transport",
"openclaw-azure-openai-responses-transport",
])("streams %s commentary bytes exactly once across start, deltas, and end", async (api) => {
])("streams %s commentary with one complete-preamble boundary", async (api) => {
const onAgentEvent = vi.fn();
const context = createMessageUpdateContext({ onAgentEvent });
// Exercise the real projection/deduplication owner. A raw callback mock
// mistakes completion metadata for another assistant text message.
context.emitAssistantStreamData = createReplyDelivery(context).emitAssistantStreamData;
const createPartial = (text: string) => ({
...createOpenAiResponsesPartial({
text,
@ -323,24 +327,34 @@ describe("handleMessageUpdate text signatures", () => {
message: finalPartial,
});
expect(onAgentEvent.mock.calls.map(([event]) => event)).toMatchObject([
expect(onAgentEvent.mock.calls.map(([event]) => event)).toEqual([
{
stream: "assistant",
stream: "item",
data: {
text: "Work",
delta: "",
replace: true,
phase: "commentary",
kind: "preamble",
title: "Preamble",
progressText: "Work",
phase: "update",
itemId: "item-commentary",
},
},
{
stream: "assistant",
stream: "item",
data: {
text: "Working...",
delta: "",
replace: true,
phase: "commentary",
kind: "preamble",
title: "Preamble",
progressText: "Working...",
phase: "update",
itemId: "item-commentary",
},
},
{
stream: "item",
data: {
kind: "preamble",
title: "Preamble",
progressText: "Working...",
phase: "end",
itemId: "item-commentary",
},
},
@ -352,6 +366,7 @@ describe("handleMessageUpdate text signatures", () => {
it("keeps same-index commentary snapshot extensions on the original live item key", async () => {
const onAgentEvent = vi.fn();
const context = createMessageUpdateContext({ onAgentEvent });
context.emitAssistantStreamData = createReplyDelivery(context).emitAssistantStreamData;
const createPartial = (text: string, id: string) =>
createOpenAiResponsesPartial({
text,
@ -374,24 +389,26 @@ describe("handleMessageUpdate text signatures", () => {
}
await endMessage(context, { message: extendedPartial });
expect(onAgentEvent.mock.calls.map(([event]) => event)).toMatchObject([
// Both snapshots finish the same logical item. The later message_end must
// not publish its already-observed completion again.
expect(onAgentEvent.mock.calls.map(([event]) => event)).toEqual([
{
stream: "assistant",
stream: "item",
data: {
text: "Working",
delta: "",
replace: true,
phase: "commentary",
kind: "preamble",
title: "Preamble",
progressText: "Working",
phase: "end",
itemId: "item-1",
},
},
{
stream: "assistant",
stream: "item",
data: {
text: "Working now",
delta: "",
replace: true,
phase: "commentary",
kind: "preamble",
title: "Preamble",
progressText: "Working now",
phase: "end",
itemId: "item-1",
},
},

4
src/agents/embedded-agent-subscribe.handlers.messages.update.ts
View file

@ -251,7 +251,7 @@ export function handleMessageUpdate(
? ctx.state.deltaBuffer
: coerceChatContentText(extractAssistantCommentaryText(streamAssistant));
const commentaryData =
commentaryText && (chunk || !hadResponsesCommentaryText)
commentaryText && (chunk || !hadResponsesCommentaryText || evtType === "text_end")
? buildAssistantStreamData({
text: commentaryText,
replace: true,
@ -260,7 +260,7 @@ export function handleMessageUpdate(
})
: undefined;
if (commentaryData) {
ctx.emitAssistantStreamData(commentaryData);
ctx.emitAssistantStreamData(commentaryData, { finalMessage: evtType === "text_end" });
}
return undefined;
}

1
src/agents/embedded-agent-subscribe.handlers.types.ts
View file

@ -70,6 +70,7 @@ type AssistantStreamDelivery = {
data: AssistantStreamData;
eventData?: AssistantStreamData;
emitPartialReply: boolean;
finalMessage?: boolean;
};
/** Incremental tag and Markdown parsing state, owned by one stream lane. */

17
src/agents/embedded-agent-subscribe.reply-delivery.ts
View file

@ -52,13 +52,17 @@ export function createReplyDelivery({ params, state, log }: ReplyDeliveryParams)
const itemId = eventData?.itemId ?? "";
const progressText =
eventData?.phase === "commentary" ? eventData.text.replace(/\s+/g, " ").trim() : "";
const preamblePhase = delivery.finalMessage ? "end" : "update";
// Completion must survive an identical last delta: first-notification
// consumers wait for this boundary, not a timer or a repeated text snapshot.
const commentarySignature = `${preamblePhase}\0${progressText}`;
const event = progressText
? {
stream: "item" as const,
data: {
kind: "preamble",
title: "Preamble",
phase: "update",
phase: preamblePhase,
progressText,
...(itemId ? { itemId } : {}),
},
@ -68,10 +72,10 @@ export function createReplyDelivery({ params, state, log }: ReplyDeliveryParams)
: { stream: "assistant" as const, data: eventData };
if (
event &&
(event.stream !== "item" || lastEmittedCommentaryByItem.get(itemId) !== progressText)
(event.stream !== "item" || lastEmittedCommentaryByItem.get(itemId) !== commentarySignature)
) {
if (event.stream === "item") {
lastEmittedCommentaryByItem.set(itemId, progressText);
lastEmittedCommentaryByItem.set(itemId, commentarySignature);
}
emitAgentEvent({ runId: params.runId, ...event });
if (params.onAgentEvent) {
@ -151,7 +155,12 @@ export function createReplyDelivery({ params, state, log }: ReplyDeliveryParams)
}
// Project before deferral while these message/block indices are current.
// Channel partials retain their block-scoped payload; the bus gets a whole message.
const delivery = { data, eventData, emitPartialReply: options?.emitPartialReply === true };
const delivery = {
data,
eventData,
emitPartialReply: options?.emitPartialReply === true,
finalMessage: options?.finalMessage === true,
};
if (!eventData && !delivery.emitPartialReply) {
return;
}

42
src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.reasoning-delivery.test.ts
View file

@ -194,6 +194,17 @@ describe("subscribeEmbeddedAgentSession", () => {
partial: commentaryMessage,
},
});
emit({
type: "message_update",
message: commentaryMessage,
assistantMessageEvent: {
type: "text_end",
contentIndex: 0,
content: "Checking files",
partial: commentaryMessage,
},
});
emit({ type: "message_end", message: commentaryMessage });
await subscription.waitForPendingEvents();
expect(onAgentEvent.mock.calls.map(([event]) => event)).toMatchObject([
@ -215,6 +226,15 @@ describe("subscribeEmbeddedAgentSession", () => {
progressText: "Checking files",
},
},
{
stream: "item",
data: {
kind: "preamble",
itemId: "item-commentary",
phase: "end",
progressText: "Checking files",
},
},
]);
expect(onBlockReply).not.toHaveBeenCalled();
expect(onPartialReply).not.toHaveBeenCalled();
@ -299,6 +319,19 @@ describe("subscribeEmbeddedAgentSession", () => {
stream: "item",
data: { kind: "preamble", itemId: "second", progressText: scenario.secondText },
},
...(scenario.eventType === "text_delta"
? [
{
stream: "item",
data: {
kind: "preamble",
itemId: "second",
phase: "end",
progressText: scenario.secondText,
},
},
]
: []),
]);
expect(onBlockReply).not.toHaveBeenCalled();
expect(onPartialReply).not.toHaveBeenCalled();
@ -336,6 +369,15 @@ describe("subscribeEmbeddedAgentSession", () => {
progressText: "First. Second.",
},
},
{
stream: "item",
data: {
kind: "preamble",
title: "Preamble",
phase: "end",
progressText: "First. Second.",
},
},
]);
});

9
test/scripts/write-plugin-sdk-entry-dts.test.ts
View file

@ -150,7 +150,8 @@ describe("write-plugin-sdk-entry-dts", () => {
expect(
(privateQa.stdout + privateQa.stderr).match(/\[tsdown-build\] invocation \d\/2 finished/gu),
).toHaveLength(2);
expectOutputs(root, qa, Object.keys(treeHashes(path.join(root, "dist"))));
const priorOutputs = treeHashes(path.join(root, "dist"));
expectOutputs(root, qa, Object.keys(priorOutputs));
expectStagingClean(root);
writeDeclarations("after");
@ -195,10 +196,12 @@ describe("write-plugin-sdk-entry-dts", () => {
for (const [relative, content] of Object.entries(preserved)) {
writeRelocated(relative, content);
}
// Seed only unowned history; current cache outputs must come from the restore below.
for (const file of Object.keys(first).filter(
// The QA build can add shared chunks after the production snapshot. Seed
// only unowned history; current cache outputs must come from the restore.
for (const file of Object.keys(priorOutputs).filter(
(entry) => !entry.startsWith("plugin-sdk/") && !cachedDistFiles.has(entry),
)) {
expect(first[file]).toBe(priorOutputs[file]);
writeRelocated(`dist/${file}`, fs.readFileSync(path.join(root, "dist", file), "utf8"));
}
writeRelocated("dist/plugin-sdk/obsolete.d.ts", "obsolete restored declaration");