Fix PR 75776 rebase regressions

AIVenture · AIVenture · commit dfd776ca598c · 2026-05-21T23:22:50.000-06:00
diff --git a/src/auto-reply/reply/dispatch-from-config.acp-abort.test.ts b/src/auto-reply/reply/dispatch-from-config.acp-abort.test.ts
@@ -30,6 +30,7 @@ let dispatchReplyFromConfig: typeof import("./dispatch-from-config.js").dispatch
 let tryDispatchAcpReplyHook: typeof import("../../plugin-sdk/acp-runtime.js").tryDispatchAcpReplyHook;
 let resetInboundDedupe: typeof import("./inbound-dedupe.js").resetInboundDedupe;
 let replyRunRegistry: typeof import("./reply-run-registry.js").replyRunRegistry;
+let replyRunRegistryTesting: typeof import("./reply-run-registry.js").__testing;
 let getActiveReplyRunCount: typeof import("./reply-run-registry.js").getActiveReplyRunCount;
 let createReplyOperation: typeof import("./reply-run-registry.js").createReplyOperation;
 
@@ -155,11 +156,16 @@ describe("dispatchReplyFromConfig ACP abort", () => {
     ({ dispatchReplyFromConfig } = await import("./dispatch-from-config.js"));
     ({ tryDispatchAcpReplyHook } = await import("../../plugin-sdk/acp-runtime.js"));
     ({ resetInboundDedupe } = await import("./inbound-dedupe.js"));
-    ({ replyRunRegistry, getActiveReplyRunCount, createReplyOperation } =
-      await import("./reply-run-registry.js"));
+    ({
+      replyRunRegistry,
+      __testing: replyRunRegistryTesting,
+      getActiveReplyRunCount,
+      createReplyOperation,
+    } = await import("./reply-run-registry.js"));
   });
 
   beforeEach(() => {
+    replyRunRegistryTesting.resetReplyRunRegistry();
     setDiscordTestRegistry();
     resetInboundDedupe();
     acpManagerRuntimeMocks.getAcpSessionManager.mockReset();
diff --git a/src/auto-reply/reply/get-reply-run.media-only.test.ts b/src/auto-reply/reply/get-reply-run.media-only.test.ts
@@ -120,6 +120,7 @@ vi.mock("./session-updates.runtime.js", () => ({
 }));
 
 vi.mock("./session-system-events.js", () => ({
+  drainFormattedSystemEventBlock: vi.fn().mockResolvedValue(undefined),
   drainFormattedSystemEvents: vi.fn().mockResolvedValue(undefined),
 }));
 
@@ -130,7 +131,7 @@ vi.mock("./typing-mode.js", () => ({
 let runPreparedReply: typeof import("./get-reply-run.js").runPreparedReply;
 let runReplyAgent: typeof import("./agent-runner.runtime.js").runReplyAgent;
 let routeReply: typeof import("./route-reply.runtime.js").routeReply;
-let drainFormattedSystemEvents: typeof import("./session-system-events.js").drainFormattedSystemEvents;
+let drainFormattedSystemEventBlock: typeof import("./session-system-events.js").drainFormattedSystemEventBlock;
 let resolveTypingMode: typeof import("./typing-mode.js").resolveTypingMode;
 let buildDirectChatContext: typeof import("./groups.js").buildDirectChatContext;
 let buildGroupChatContext: typeof import("./groups.js").buildGroupChatContext;
@@ -277,7 +278,7 @@ describe("runPreparedReply media-only handling", () => {
     ({ runPreparedReply } = await import("./get-reply-run.js"));
     ({ runReplyAgent } = await import("./agent-runner.runtime.js"));
     ({ routeReply } = await import("./route-reply.runtime.js"));
-    ({ drainFormattedSystemEvents } = await import("./session-system-events.js"));
+    ({ drainFormattedSystemEventBlock } = await import("./session-system-events.js"));
     ({ resolveTypingMode } = await import("./typing-mode.js"));
     ({ buildDirectChatContext, buildGroupChatContext } = await import("./groups.js"));
     ({ buildInboundUserContextPrefix, resolveInboundUserContextPromptJoiner } =
@@ -1517,9 +1518,15 @@ describe("runPreparedReply media-only handling", () => {
   it("re-drains system events after waiting behind an active run", async () => {
     const queueSettings = await import("./queue/settings-runtime.js");
     vi.mocked(queueSettings.resolveQueueSettings).mockReturnValueOnce({ mode: "interrupt" });
-    vi.mocked(drainFormattedSystemEvents)
-      .mockResolvedValueOnce("System: [t] Initial event.")
-      .mockResolvedValueOnce("System: [t] Post-compaction context.");
+    vi.mocked(drainFormattedSystemEventBlock)
+      .mockResolvedValueOnce({
+        text: "System: [t] Initial event.",
+        forceSenderIsOwnerFalse: false,
+      })
+      .mockResolvedValueOnce({
+        text: "System: [t] Post-compaction context.",
+        forceSenderIsOwnerFalse: false,
+      });
 
     const previousRun = createReplyOperation({
       sessionId: "session-events-after-wait",
@@ -2224,7 +2231,10 @@ describe("runPreparedReply media-only handling", () => {
   });
 
   it("routes queued system events into user prompt text, not system prompt context", async () => {
-    vi.mocked(drainFormattedSystemEvents).mockResolvedValueOnce("System: [t] Model switched.");
+    vi.mocked(drainFormattedSystemEventBlock).mockResolvedValueOnce({
+      text: "System: [t] Model switched.",
+      forceSenderIsOwnerFalse: false,
+    });
 
     await runPreparedReply(baseParams());
 
@@ -2304,7 +2314,10 @@ describe("runPreparedReply media-only handling", () => {
   });
 
   it("keeps sender ownership when drained system events are present", async () => {
-    vi.mocked(drainFormattedSystemEvents).mockResolvedValueOnce("System: [t] Trusted event.");
+    vi.mocked(drainFormattedSystemEventBlock).mockResolvedValueOnce({
+      text: "System: [t] Trusted event.",
+      forceSenderIsOwnerFalse: false,
+    });
     const params = ownerParams();
 
     await runPreparedReply(params);
@@ -2314,9 +2327,10 @@ describe("runPreparedReply media-only handling", () => {
   });
 
   it("does not downgrade sender ownership when event text contains the untrusted marker", async () => {
-    vi.mocked(drainFormattedSystemEvents).mockResolvedValueOnce(
-      "System: [t] Relay text mentions System (untrusted): but event is trusted.",
-    );
+    vi.mocked(drainFormattedSystemEventBlock).mockResolvedValueOnce({
+      text: "System: [t] Relay text mentions System (untrusted): but event is trusted.",
+      forceSenderIsOwnerFalse: false,
+    });
     const params = ownerParams();
 
     await runPreparedReply(params);
@@ -2329,7 +2343,10 @@ describe("runPreparedReply media-only handling", () => {
     // drainFormattedSystemEvents returns the events block; the caller prepends it.
     // The hint must be extracted from the user body BEFORE prepending, so "System:"
     // does not shadow the low|medium|high shorthand.
-    vi.mocked(drainFormattedSystemEvents).mockResolvedValueOnce("System: [t] Node connected.");
+    vi.mocked(drainFormattedSystemEventBlock).mockResolvedValueOnce({
+      text: "System: [t] Node connected.",
+      forceSenderIsOwnerFalse: false,
+    });
 
     await runPreparedReply(
       baseParams({
@@ -2352,7 +2369,10 @@ describe("runPreparedReply media-only handling", () => {
   it("carries system events into followupRun.prompt for deferred turns", async () => {
     // drainFormattedSystemEvents returns the events block; the caller prepends it to
     // effectiveBaseBody for the queue path so deferred turns see events.
-    vi.mocked(drainFormattedSystemEvents).mockResolvedValueOnce("System: [t] Node connected.");
+    vi.mocked(drainFormattedSystemEventBlock).mockResolvedValueOnce({
+      text: "System: [t] Node connected.",
+      forceSenderIsOwnerFalse: false,
+    });
 
     await runPreparedReply(baseParams());
 
@@ -2363,7 +2383,7 @@ describe("runPreparedReply media-only handling", () => {
   it("does not strip think-hint token from deferred queue body", async () => {
     // In steer mode the inferred thinkLevel is never consumed, so the first token
     // must not be stripped from the queue/steer body (followupRun.prompt).
-    vi.mocked(drainFormattedSystemEvents).mockResolvedValueOnce(undefined);
+    vi.mocked(drainFormattedSystemEventBlock).mockResolvedValueOnce(undefined);
 
     await runPreparedReply(
       baseParams({
diff --git a/src/auto-reply/reply/get-reply-run.ts b/src/auto-reply/reply/get-reply-run.ts
@@ -81,7 +81,7 @@ import { resolveRoutedDeliveryThreadId } from "./routed-delivery-thread.js";
 import { resolveRuntimePolicySessionKey } from "./runtime-policy-session-key.js";
 import { resolveBareSessionResetPromptState } from "./session-reset-prompt.js";
 import { resolveBareResetBootstrapFileAccess } from "./session-reset-prompt.js";
-import { drainFormattedSystemEvents } from "./session-system-events.js";
+import { drainFormattedSystemEventBlock } from "./session-system-events.js";
 import { buildSessionStartupContextPrelude, shouldApplyStartupContext } from "./startup-context.js";
 import { resolveStoredModelOverride } from "./stored-model-override.js";
 import { resolveTypingMode } from "./typing-mode.js";
@@ -724,21 +724,25 @@ export async function runPreparedReply(
       ? `[Thread starter - for context]\n${threadStarterBody}`
       : undefined;
   const drainedSystemEventBlocks: string[] = [];
+  let forceSenderIsOwnerFalseFromSystemEvents = false;
   const rebuildPromptBodies = async (): Promise<{
     prefixedCommandBody: string;
     queuedBody: string;
     transcriptCommandBody: string;
     currentInboundContext?: typeof promptEnvelopeBase.currentInboundContext;
   }> => {
     if (!useFastReplyRuntime) {
-      const eventsBlock = await drainFormattedSystemEvents({
+      const eventsBlock = await drainFormattedSystemEventBlock({
         cfg,
         sessionKey,
         isMainSession,
         isNewSession,
       });
       if (eventsBlock) {
-        drainedSystemEventBlocks.push(eventsBlock);
+        drainedSystemEventBlocks.push(eventsBlock.text);
+        if (eventsBlock.forceSenderIsOwnerFalse) {
+          forceSenderIsOwnerFalseFromSystemEvents = true;
+        }
       }
     }
     return buildReplyPromptEnvelope({
@@ -1123,11 +1127,10 @@ export async function runPreparedReply(
       senderName: normalizeOptionalString(sessionCtx.SenderName),
       senderUsername: normalizeOptionalString(sessionCtx.SenderUsername),
       senderE164: normalizeOptionalString(sessionCtx.SenderE164),
-      // Queued system events are prompt content in the same trusted session;
-      // they do not rewrite the sender identity used by command/action auth.
-      senderIsOwner: command.senderIsOwner,
+      senderIsOwner: forceSenderIsOwnerFalseFromSystemEvents ? false : command.senderIsOwner,
       traceAuthorized:
-        command.senderIsOwner || (ctx.GatewayClientScopes ?? []).includes("operator.admin"),
+        (forceSenderIsOwnerFalseFromSystemEvents ? false : command.senderIsOwner) ||
+        (ctx.GatewayClientScopes ?? []).includes("operator.admin"),
       sessionFile: preparedSessionState.sessionFile,
       workspaceDir,
       config: cfg,
diff --git a/src/auto-reply/reply/memory-flush.ts b/src/auto-reply/reply/memory-flush.ts
@@ -4,7 +4,8 @@ import { parseNonNegativeByteSize } from "../../config/byte-size.js";
 import { resolveFreshSessionTotalTokens, type SessionEntry } from "../../config/sessions.js";
 import type { OpenClawConfig } from "../../config/types.openclaw.js";
 
-const PREFLIGHT_COMPACTION_MAX_THRESHOLD_TOKENS = 65_000;
+const PREFLIGHT_COMPACTION_PROACTIVE_THRESHOLD_TOKENS = 65_000;
+const PREFLIGHT_COMPACTION_PROACTIVE_CONTEXT_WINDOW_MAX_TOKENS = 400_000;
 
 export function resolveMemoryFlushContextWindowTokens(params: {
   modelId?: string;
@@ -57,10 +58,13 @@ export function resolvePreflightCompactionThresholdTokens(params: {
 }): number {
   const contextWindow = Math.max(1, Math.floor(params.contextWindowTokens));
   const configuredThreshold = resolveConfiguredCompactionThresholdTokens(params);
-  const proactiveThreshold = Math.max(
-    1,
-    Math.min(PREFLIGHT_COMPACTION_MAX_THRESHOLD_TOKENS, Math.floor(contextWindow * 0.25)),
-  );
+  if (
+    typeof configuredThreshold === "number" &&
+    contextWindow > PREFLIGHT_COMPACTION_PROACTIVE_CONTEXT_WINDOW_MAX_TOKENS
+  ) {
+    return configuredThreshold;
+  }
+  const proactiveThreshold = Math.max(1, PREFLIGHT_COMPACTION_PROACTIVE_THRESHOLD_TOKENS);
   return Math.min(configuredThreshold ?? proactiveThreshold, proactiveThreshold);
 }
 
diff --git a/src/auto-reply/reply/session-system-events.ts b/src/auto-reply/reply/session-system-events.ts
@@ -21,6 +21,11 @@ function selectGenericSystemEvents(events: readonly SystemEvent[]): SystemEvent[
   return events.filter((event) => !isExecCompletionEvent(event.text));
 }
 
+export type FormattedSystemEventBlock = {
+  text: string;
+  forceSenderIsOwnerFalse: boolean;
+};
+
 function compactSystemEvent(line: string): string | null {
   const trimmed = line.trim();
   if (!trimmed) {
@@ -83,15 +88,16 @@ function formatSystemEventTimestamp(ts: number, cfg: OpenClawConfig) {
   );
 }
 
-/** Drain queued system events, format as `System:` lines, return the block text (or undefined). */
-export async function drainFormattedSystemEvents(params: {
+/** Drain queued system events, format as `System:` lines, return the block with authority metadata. */
+export async function drainFormattedSystemEventBlock(params: {
   cfg: OpenClawConfig;
   sessionKey: string;
   isMainSession: boolean;
   isNewSession: boolean;
-}): Promise<string | undefined> {
+}): Promise<FormattedSystemEventBlock | undefined> {
   const summaryLines: string[] = [];
   const systemLines: string[] = [];
+  let forceSenderIsOwnerFalse = false;
   // Exec completions have a dedicated heartbeat prompt; leave those entries queued
   // so the heartbeat path can consume and deliver them.
   const queued = consumeSelectedSystemEventEntries(
@@ -103,6 +109,9 @@ export async function drainFormattedSystemEvents(params: {
     if (!compacted) {
       continue;
     }
+    if (event.forceSenderIsOwnerFalse === true) {
+      forceSenderIsOwnerFalse = true;
+    }
     const timestamp = `[${formatSystemEventTimestamp(event.ts, params.cfg)}]`;
     let index = 0;
     for (const subline of compacted.split("\n")) {
@@ -126,7 +135,21 @@ export async function drainFormattedSystemEvents(params: {
 
   // Each sub-line gets its own prefix so continuation lines can't be mistaken
   // for regular user content.
-  return summaryLines.length > 0
-    ? [...summaryLines, ...systemLines].join("\n")
-    : systemLines.join("\n");
+  return {
+    text:
+      summaryLines.length > 0
+        ? [...summaryLines, ...systemLines].join("\n")
+        : systemLines.join("\n"),
+    forceSenderIsOwnerFalse,
+  };
+}
+
+/** Drain queued system events, format as `System:` lines, return the block text (or undefined). */
+export async function drainFormattedSystemEvents(params: {
+  cfg: OpenClawConfig;
+  sessionKey: string;
+  isMainSession: boolean;
+  isNewSession: boolean;
+}): Promise<string | undefined> {
+  return (await drainFormattedSystemEventBlock(params))?.text;
 }
diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts
@@ -2140,7 +2140,7 @@ export const chatHandlers: GatewayRequestHandlers = {
       runIds: res.aborted ? [runId] : [],
     });
   },
-  "chat.send": async ({ params, respond, context, client }) => {
+  "chat.send": async ({ params, respond, context, client, isWebchatConnect }) => {
     if (!validateChatSendParams(params)) {
       respond(
         false,
@@ -2514,12 +2514,14 @@ export const chatHandlers: GatewayRequestHandlers = {
         status: "started" as const,
       };
       respond(true, ackPayload, undefined, { runId: clientRunId });
-      broadcastWebchatPreflightAcknowledgement({
-        context,
-        runId: clientRunId,
-        sessionKey,
-        message: parsedMessage,
-      });
+      if (client?.connect && isWebchatConnect(client.connect)) {
+        broadcastWebchatPreflightAcknowledgement({
+          context,
+          runId: clientRunId,
+          sessionKey,
+          message: parsedMessage,
+        });
+      }
       const persistedImagesPromise = persistChatSendImages({
         images: parsedImages,
         imageOrder,
diff --git a/test/vitest/vitest.auto-reply-reply.config.ts b/test/vitest/vitest.auto-reply-reply.config.ts
@@ -2,14 +2,27 @@ import { createScopedVitestConfig } from "./vitest.scoped-config.ts";
 import { autoReplyReplySubtreeTestInclude } from "./vitest.test-shards.mjs";
 
 export function createAutoReplyReplyVitestConfig(env?: Record<string, string | undefined>) {
-  return createScopedVitestConfig([...autoReplyReplySubtreeTestInclude], {
+  const config = createScopedVitestConfig([...autoReplyReplySubtreeTestInclude], {
     dir: "src/auto-reply",
     env,
     name: "auto-reply-reply",
     sequence: {
       groupOrder: 1,
     },
   });
+  // This shard uses the non-isolated runner and shared reply dispatch/abort
+  // registries. Keep files serialized so ACP abort tests cannot race other
+  // dispatch tests that intentionally mutate the same singleton state.
+  config.test = {
+    ...config.test,
+    maxWorkers: 1,
+    fileParallelism: false,
+    sequence: {
+      ...config.test?.sequence,
+      groupOrder: 1,
+    },
+  };
+  return config;
 }
 
 export default createAutoReplyReplyVitestConfig();
