ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-24 19:35:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(auto-reply): apply prepared delivery before streaming

Browse Source
This commit is contained in:
joshavant
2026-08-10 19:09:20 -05:00
committed by Josh Avant
parent 1d1a7fa499
commit 375df347c4
4 changed files with 76 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/auto-reply/reply/agent-runner-embedded-candidate.ts
+14 -13
View File
@@ -44,6 +44,7 @@ import type { FollowupRun } from "./queue.js";
import { isReplyOperationRestartAbort } from "./reply-operation-abort.js";
import { markReplyOperationGlobalLaneWaitProgress } from "./reply-run-registry.js";
import {
bindPreparedHarnessSourceReplyDeliveryMode,
readPreparedHarnessSourceReplyDeliveryMode,
type SourceReplyDeliveryRuntimeOptions,
} from "./source-reply-delivery-runtime.js";
@@ -193,6 +194,17 @@ export async function runEmbeddedFallbackCandidate(params: {
}),
});
params.onLifecycleBackstop(lifecycleBackstop);
const unbindPreparedHarnessSourceReplyDeliveryMode = bindPreparedHarnessSourceReplyDeliveryMode(
params.candidateRun,
(mode) => {
params.candidateRun.sourceReplyDeliveryMode = mode;
turn.followupRun.run.sourceReplyDeliveryMode = mode;
if (turn.opts) {
turn.opts.sourceReplyDeliveryMode = mode;
}
sourceReplyDeliveryRuntimeOptions?.onSourceReplyDeliveryModeResolved?.(mode);
},
);
try {
// Profiler milestone. Exposes pre-dispatch delay without normal-path logging.
params.timing.logMilestoneIfSlow({
@@ -202,7 +214,7 @@ export async function runEmbeddedFallbackCandidate(params: {
milestone: "before_embedded_run",
});
let eventHandler: ReturnType<typeof createAgentRunEventHandler> | undefined;
const embeddedRun = params.timing.measure("embedded_run", () =>
const result = await params.timing.measure("embedded_run", () =>
runEmbeddedAgent({
preparedRunAdmission: params.preparedRunAdmission,
...embeddedContext,
@@ -428,18 +440,6 @@ export async function runEmbeddedFallbackCandidate(params: {
: undefined,
}),
);
const result = await embeddedRun.finally(() => {
const mode = readPreparedHarnessSourceReplyDeliveryMode(params.candidateRun);
if (!mode) {
return;
}
params.candidateRun.sourceReplyDeliveryMode = mode;
turn.followupRun.run.sourceReplyDeliveryMode = mode;
if (turn.opts) {
turn.opts.sourceReplyDeliveryMode = mode;
}
sourceReplyDeliveryRuntimeOptions?.onSourceReplyDeliveryModeResolved?.(mode);
});
const resultCompactionCount = Math.max(0, result.meta?.agentMeta?.compactionCount ?? 0);
attemptCompactionCount = Math.max(attemptCompactionCount, resultCompactionCount);
return {
@@ -449,6 +449,7 @@ export async function runEmbeddedFallbackCandidate(params: {
),
};
} finally {
unbindPreparedHarnessSourceReplyDeliveryMode();
params.onCompactionCount(attemptCompactionCount);
revokeMessageActionTurnCapability(messageActionTurnCapability);
}
src/auto-reply/reply/dispatch-from-config.prepare-execution.ts
+8 -13
View File
@@ -32,20 +32,17 @@ export async function prepareDispatchExecution(state: ChooseDispatchRouteReadySt
noteCommentaryProgress,
params,
sendPayloadAsync,
sendPolicyDenied,
sessionKey,
shouldEmitVerboseProgress,
shouldRouteToOriginating,
shouldSendToolSummaries,
shouldSendVerboseProgressMessages,
suppressAutomaticSourceDelivery,
suppressDelivery,
turnLedger,
} = state;
// When automatic source delivery is suppressed, still let the agent process
// the inbound message (context, memory, tool calls) but suppress automatic
// outbound source delivery.
if (suppressDelivery) {
if (state.suppressDelivery) {
logVerbose(
`Delivery suppressed by ${state.deliverySuppressionReason} for session ${state.sessionStoreEntry.sessionKey ?? sessionKey ?? "unknown"} — agent will still process the message`,
);
@@ -146,8 +143,8 @@ export async function prepareDispatchExecution(state: ChooseDispatchRouteReadySt
systemEvent: shouldRouteToOriginating,
});
const shouldSuppressProgressDelivery = () =>
sendPolicyDenied ||
(suppressDelivery && !state.shouldDeliverVerboseProgressDespiteSourceSuppression());
state.sendPolicyDenied ||
(state.suppressDelivery && !state.shouldDeliverVerboseProgressDespiteSourceSuppression());
const hasVisibleRegularVerboseToolProgress = () =>
shouldEmitVerboseProgress() &&
!state.shouldEmitFullVerboseProgress() &&
@@ -217,9 +214,9 @@ export async function prepareDispatchExecution(state: ChooseDispatchRouteReadySt
return false;
}
return (
!suppressAutomaticSourceDelivery ||
!state.suppressAutomaticSourceDelivery ||
(allowSuppressedSourceProgressCallbacks &&
!sendPolicyDenied &&
!state.sendPolicyDenied &&
options?.forwardWhenSourceDeliverySuppressed === true)
);
};
@@ -310,13 +307,11 @@ export async function prepareDispatchExecution(state: ChooseDispatchRouteReadySt
forwardWhenSourceDeliverySuppressed: true,
requiresToolSummaryVisibility: true,
} as const;
const canForwardItemEvents =
Boolean(params.replyOptions?.onItemEvent) &&
shouldForwardProgressCallback(itemEventForwardingOptions);
const canForwardItemEvents = Boolean(params.replyOptions?.onItemEvent);
const canForwardSuppressedSourceItemEvents =
suppressAutomaticSourceDelivery &&
allowSuppressedSourceProgressCallbacks &&
canForwardItemEvents;
!state.sendPolicyDenied &&
Boolean(params.replyOptions?.onItemEvent);
const shouldDeliverDurableCommentaryProgress = (
payload: Parameters<NonNullable<GetReplyOptions["onItemEvent"]>>[0],
) =>
src/auto-reply/reply/prepared-harness-source-delivery.integration.test.ts
+33 -10
View File
@@ -39,18 +39,24 @@ describe("prepared harness source delivery", () => {
it.each([
{
name: "delivers one final when preparation changes tool ownership to automatic",
name: "delivers one streamed answer when preparation changes tool ownership to automatic",
failsCliPrimary: true,
preliminaryVisibleReplies: "message_tool" as const,
preparedVisibleReplies: "automatic" as const,
expectedTransitions: ["message_tool_only", "automatic"],
expectedDeliveries: 1,
expectedPartials: 1,
expectedFinals: 1,
},
{
name: "sends nothing when preparation retains tool ownership",
name: "suppresses live output when preparation changes automatic ownership to tool",
failsCliPrimary: false,
preliminaryVisibleReplies: "automatic" as const,
preparedVisibleReplies: "message_tool" as const,
expectedTransitions: ["message_tool_only", "message_tool_only"],
expectedTransitions: ["message_tool_only"],
expectedDeliveries: 0,
expectedPartials: 0,
expectedFinals: 0,
},
])("$name", async (testCase) => {
const { runEmbeddedAgent, registerPreparedAgentHarness } =
@@ -62,10 +68,15 @@ describe("prepared harness source delivery", () => {
providerOverride: "openai",
modelOverride: "gpt-5.4",
});
const emittedStreamingCallbacks: string[] = [];
mockedBuildEmbeddedRunPayloads.mockReturnValue([{ text: "Short fallback final" }]);
mockedRunEmbeddedAttempt.mockResolvedValue(
makeAttemptResult({ assistantTexts: ["Short fallback final"] }),
);
mockedRunEmbeddedAttempt.mockImplementation(async (attemptParams) => {
emittedStreamingCallbacks.push("partial");
await attemptParams.onPartialReply?.({ text: "Short fallback final" });
emittedStreamingCallbacks.push("block");
await attemptParams.onBlockReply?.({ text: "Short fallback final" });
return makeAttemptResult({ assistantTexts: ["Short fallback final"] });
});
useOpenAIPlatformAuthFixture();
let embeddedError: unknown;
let embeddedParams: unknown;
@@ -84,7 +95,9 @@ describe("prepared harness source delivery", () => {
runnerState.runCliAgentMock.mockRejectedValueOnce(new Error("cli failed"));
runnerState.runWithModelFallbackMock.mockImplementationOnce(
async (params: FallbackRunnerParams) => {
await params.run("anthropic", "primary").catch(() => undefined);
if (testCase.failsCliPrimary) {
await params.run("anthropic", "primary").catch(() => undefined);
}
return {
result: await params.run("custom", "plugin-fallback"),
provider: "custom",
@@ -115,7 +128,13 @@ describe("prepared harness source delivery", () => {
provider === "openai"
? { supported: true, priority: 200 }
: { supported: false, reason: "prepared OpenAI route only" },
runAttempt: vi.fn(async () => ({}) as never),
runAttempt: vi.fn(async (attemptParams) => {
emittedStreamingCallbacks.push("partial");
await attemptParams.onPartialReply?.({ text: "Short fallback final" });
emittedStreamingCallbacks.push("block");
await attemptParams.onBlockReply?.({ text: "Short fallback final" });
return makeAttemptResult({ assistantTexts: ["Short fallback final"] });
}),
});
}
sessionStoreMocks.currentEntry = {
@@ -156,7 +175,7 @@ describe("prepared harness source delivery", () => {
opts: runtimeOpts,
typingSignals: createMockTypingSignaler(),
blockReplyPipeline: null,
blockStreamingEnabled: false,
blockStreamingEnabled: true,
resolvedBlockStreamingBreak: "message_end",
applyReplyToMode: (payload) => payload,
shouldEmitToolResult: () => true,
@@ -183,6 +202,7 @@ describe("prepared harness source delivery", () => {
return execution.runResult.payloads[0] satisfies ReplyPayload;
});
const deliver = vi.fn(async () => {});
const onPartialReply = vi.fn(async () => {});
const dispatcher = createReplyDispatcher({ deliver });
const result = await dispatchReplyFromConfig({
@@ -190,6 +210,7 @@ describe("prepared harness source delivery", () => {
cfg: emptyConfig,
dispatcher,
replyResolver,
replyOptions: { onPartialReply },
});
await settleReplyDispatcher({ dispatcher });
@@ -197,7 +218,9 @@ describe("prepared harness source delivery", () => {
{ prompt: "hello" },
expect.any(Object),
);
expect(emittedStreamingCallbacks).toEqual(["partial", "block"]);
expect(modeTransitions).toEqual(testCase.expectedTransitions);
expect(onPartialReply).toHaveBeenCalledTimes(testCase.expectedPartials);
expect(result.queuedFinal).toBe(testCase.expectedDeliveries === 1);
if (testCase.expectedDeliveries === 1) {
expect(result.sourceReplyDeliveryMode).toBeUndefined();
@@ -212,7 +235,7 @@ describe("prepared harness source delivery", () => {
expect(dispatcher.getQueuedCounts()).toEqual({
tool: 0,
block: 0,
final: testCase.expectedDeliveries,
final: testCase.expectedFinals,
});
expect(dispatcher.getFailedCounts()).toEqual({ tool: 0, block: 0, final: 0 });
});
src/auto-reply/reply/source-reply-delivery-runtime.ts
+21 -2
View File
@@ -7,14 +7,16 @@ export type SourceReplyDeliveryRuntimeOptions = {
onSourceReplyDeliveryModeResolved?: (mode: SourceReplyDeliveryMode) => void;
};
// Enumerable symbol metadata follows queue-owned run spreads without widening its public type.
// Losing it would let a failed candidate's runtime default govern the fallback winner.
// The shared enumerable binding follows queue/run spreads without widening their public types.
// Its listener moves prepared ownership before live callbacks; copying only the mode would leave
// pre-settlement source delivery on the preliminary policy.
const sourceReplyDeliveryModeOriginKey: unique symbol = Symbol.for(
"openclaw.source-reply-delivery-runtime",
);
type SourceReplyDeliveryRuntimeBinding = {
origin?: SourceReplyDeliveryModeOrigin;
preparedHarnessMode?: SourceReplyDeliveryMode;
preparedHarnessModeListener?: (mode: SourceReplyDeliveryMode) => void;
};
type SourceReplyDeliveryModeOwner = {
[sourceReplyDeliveryModeOriginKey]?: SourceReplyDeliveryRuntimeBinding;
@@ -55,9 +57,26 @@ export function publishPreparedHarnessSourceReplyDeliveryMode(
const binding = readSourceReplyDeliveryRuntimeBinding(owner);
if (binding?.origin === "runtime_default") {
binding.preparedHarnessMode = mode;
binding.preparedHarnessModeListener?.(mode);
}
}
export function bindPreparedHarnessSourceReplyDeliveryMode(
owner: object,
listener: (mode: SourceReplyDeliveryMode) => void,
): () => void {
const binding = readSourceReplyDeliveryRuntimeBinding(owner);
if (binding?.origin !== "runtime_default") {
return () => {};
}
binding.preparedHarnessModeListener = listener;
return () => {
if (binding.preparedHarnessModeListener === listener) {
binding.preparedHarnessModeListener = undefined;
}
};
}
export function readPreparedHarnessSourceReplyDeliveryMode(
owner: object,
): SourceReplyDeliveryMode | undefined {