src/auto-reply/reply/agent-runner-direct-runtime-config.test.ts

// Tests direct runtime config overrides passed into agent runner execution.
import { writeFile } from "node:fs/promises";
import { join } from "node:path";
import { beforeEach, describe, expect, it, vi } from "vitest";
import { OAuthRefreshFailureError } from "../../agents/auth-profiles/oauth-refresh-failure.js";
import { FailoverError } from "../../agents/failover-error.js";
import { SessionManager } from "../../agents/sessions/session-manager.js";
import {
  appendTranscriptMessage,
  loadSessionEntry,
  replaceSessionEntry,
} from "../../config/sessions/session-accessor.js";
import type { SessionEntry } from "../../config/sessions/types.js";
import {
  clearMemoryPluginState,
  registerMemoryCapability,
} from "../../plugins/memory-state.test-fixtures.js";
import { withTestDir } from "../../test-helpers/temp-dir.js";
import { withReplyDispatcher } from "../dispatch-dispatcher.js";
import { getReplyPayloadMetadata } from "../reply-payload.js";
import type { TemplateContext } from "../templating.js";
import { SILENT_REPLY_TOKEN } from "../tokens.js";
import type { ReplyPayload } from "../types.js";
import { createTestFollowupRun, withTestModelContextTokens } from "./agent-runner.test-fixtures.js";
import type { QueueSettings } from "./queue.js";
import { createReplyDispatcher } from "./reply-dispatcher.js";
import type { ReplyDispatchKind } from "./reply-dispatcher.types.js";
import {
  REPLY_OPERATION_RUN_STATE,
  resolveReplyOperationAgentTurn,
  type ReplyOperationRunState,
} from "./reply-operation-run-state.js";
import type { ReplyOperation } from "./reply-run-registry.js";
import { createMockReplyOperation, createMockTypingController } from "./test-helpers.js";

const freshCfg = { runtimeFresh: true };
const staleCfg = {
  runtimeFresh: false,
  skills: {
    entries: {
      whisper: {
        apiKey: { source: "env" as const, provider: "default", id: "OPENAI_API_KEY" },
      },
    },
  },
};
const sentinelError = new Error("stop-after-preflight");

const resolveQueuedReplyExecutionConfigMock = vi.fn();
const resolveReplyToModeMock = vi.fn();
const createReplyToModeFilterForChannelMock = vi.fn();
const createReplyMediaContextMock = vi.fn();
const createReplyMediaPathNormalizerMock = vi.fn();
const runSessionCompactionIfNeededMock = vi.fn();
const runMemoryFlushIfNeededMock = vi.fn();
const executeAgentTurnMock = vi.fn();
const resetReplyRunSessionMock = vi.fn();
const enqueueFollowupRunMock = vi.fn();

vi.mock("./agent-runner-utils.js", async () => {
  const actual =
    await vi.importActual<typeof import("./agent-runner-utils.js")>("./agent-runner-utils.js");
  return {
    ...actual,
    resolveQueuedReplyExecutionConfig: (...args: unknown[]) =>
      resolveQueuedReplyExecutionConfigMock(...args),
  };
});

vi.mock("./reply-threading.js", async () => {
  const actual =
    await vi.importActual<typeof import("./reply-threading.js")>("./reply-threading.js");
  return {
    ...actual,
    resolveReplyToMode: (...args: unknown[]) => resolveReplyToModeMock(...args),
    createReplyToModeFilterForChannel: (...args: unknown[]) =>
      createReplyToModeFilterForChannelMock(...args),
  };
});

vi.mock("./reply-media-paths.js", () => ({
  createReplyMediaContext: (...args: unknown[]) => {
    createReplyMediaContextMock(...args);
    return {
      normalizePayload: createReplyMediaPathNormalizerMock(...args),
    };
  },
  createReplyMediaPathNormalizer: (...args: unknown[]) =>
    createReplyMediaPathNormalizerMock(...args),
}));

vi.mock("./agent-runner-memory.js", () => ({
  runSessionCompactionIfNeeded: (...args: unknown[]) => runSessionCompactionIfNeededMock(...args),
  runMemoryFlushIfNeeded: (...args: unknown[]) => runMemoryFlushIfNeededMock(...args),
}));

vi.mock("./agent-runner-execution.js", async () => {
  const actual = await vi.importActual<typeof import("./agent-runner-execution.js")>(
    "./agent-runner-execution.js",
  );
  return {
    ...actual,
    executeAgentTurn: (...args: unknown[]) => executeAgentTurnMock(...args),
  };
});

vi.mock("./agent-runner-session-reset.js", async () => {
  const actual = await vi.importActual<typeof import("./agent-runner-session-reset.js")>(
    "./agent-runner-session-reset.js",
  );
  return {
    ...actual,
    resetReplyRunSession: (...args: unknown[]) => resetReplyRunSessionMock(...args),
  };
});

vi.mock("./queue.js", async () => {
  const actual = await vi.importActual<typeof import("./queue.js")>("./queue.js");
  return {
    ...actual,
    enqueueFollowupRun: (...args: unknown[]) => enqueueFollowupRunMock(...args),
  };
});

const { runReplyAgent } = await import("./agent-runner.js");

function createTelegramSessionCtx(): TemplateContext {
  return {
    Provider: "telegram",
    OriginatingChannel: "telegram",
    OriginatingTo: "12345",
    AccountId: "default",
    ChatType: "dm",
    MessageSid: "msg-1",
  } as unknown as TemplateContext;
}

type TestReplyOperation = ReplyOperation & {
  setPhase: ReturnType<typeof vi.fn<ReplyOperation["setPhase"]>>;
};

function createReplyOperation(): TestReplyOperation {
  const { replyOperation } = createMockReplyOperation({ key: "test", sessionId: "session-1" });
  return Object.assign(replyOperation, {
    phase: "queued" as const,
    setPhase: vi.fn<ReplyOperation["setPhase"]>(),
    hasOwnedSessionId: vi.fn(() => false),
    captureOwnedSessionIds: vi.fn(() => new Set()),
  });
}

function createDirectRuntimeReplyParams({
  shouldFollowup,
  isActive,
}: {
  shouldFollowup: boolean;
  isActive: boolean;
}) {
  const followupRun = createTestFollowupRun({
    sessionId: "session-1",
    sessionKey: "agent:main:telegram:default:direct:test",
    messageProvider: "telegram",
    config: staleCfg,
    provider: "openai",
    model: "gpt-5.4",
  });
  const resolvedQueue = { mode: "interrupt" } as QueueSettings;
  const replyParams: Parameters<typeof runReplyAgent>[0] = {
    commandBody: "hello",
    followupRun,
    queueKey: "main",
    resolvedQueue,
    shouldSteer: false,
    shouldFollowup,
    isActive,
    typing: createMockTypingController(),
    sessionCtx: createTelegramSessionCtx(),
    defaultModel: "openai/gpt-5.4",
    resolvedVerboseLevel: "off",
    isNewSession: false,
    blockStreamingEnabled: false,
    resolvedBlockStreamingBreak: "message_end",
    shouldInjectGroupIntro: false,
    typingMode: "instant",
  };

  return { followupRun, resolvedQueue, replyParams };
}

function requireResolveQueuedReplyExecutionConfigCall(index = 0) {
  const call = resolveQueuedReplyExecutionConfigMock.mock.calls[index] as
    | [
        unknown,
        {
          originatingChannel?: string;
          messageProvider?: string;
        },
      ]
    | undefined;
  if (!call) {
    throw new Error(`resolveQueuedReplyExecutionConfig call ${index} missing`);
  }
  return call;
}

type MockCallSource = {
  mock: {
    calls: unknown[][];
  };
};

function requireMaintenanceCall(mock: MockCallSource, name: string, index = 0) {
  const call = mock.mock.calls[index]?.[0] as
    | {
        cfg?: unknown;
        followupRun?: unknown;
        sessionKey?: string;
        runtimePolicySessionKey?: string;
      }
    | undefined;
  if (!call) {
    throw new Error(`${name} call ${index} missing`);
  }
  return call;
}

type PreflightParams = Parameters<
  typeof import("./agent-runner-memory.js").runSessionCompactionIfNeeded
>[0];

async function runRequiredCheckpoint(params: PreflightParams) {
  if (!params.beforeCompaction) {
    throw new Error("Expected the required preflight checkpoint");
  }
  return params.beforeCompaction(
    params.sessionEntry ?? { sessionId: params.followupRun.run.sessionId, updatedAt: 1 },
  );
}

describe("runReplyAgent runtime config", () => {
  beforeEach(() => {
    resolveQueuedReplyExecutionConfigMock.mockReset();
    resolveReplyToModeMock.mockReset();
    createReplyToModeFilterForChannelMock.mockReset();
    createReplyMediaContextMock.mockReset();
    createReplyMediaPathNormalizerMock.mockReset();
    runSessionCompactionIfNeededMock.mockReset();
    runMemoryFlushIfNeededMock.mockReset();
    executeAgentTurnMock.mockReset();
    resetReplyRunSessionMock.mockReset();
    enqueueFollowupRunMock.mockReset();

    resolveQueuedReplyExecutionConfigMock.mockResolvedValue(freshCfg);
    resolveReplyToModeMock.mockReturnValue("all");
    createReplyToModeFilterForChannelMock.mockReturnValue((payload: unknown) => payload);
    createReplyMediaPathNormalizerMock.mockReturnValue((payload: unknown) => payload);
    runSessionCompactionIfNeededMock.mockImplementation(async (params: PreflightParams) => {
      await runRequiredCheckpoint(params);
      throw sentinelError;
    });
    runMemoryFlushIfNeededMock.mockResolvedValue({ sessionEntry: undefined, outcome: "skipped" });
    executeAgentTurnMock.mockResolvedValue({
      runId: "runtime-config-test",
      outcome: { kind: "rejected", payload: { text: "main reply" } },
    });
    resetReplyRunSessionMock.mockResolvedValue(false);
  });

  it("resolves direct reply runs before early helpers read config", async () => {
    const { followupRun, replyParams } = createDirectRuntimeReplyParams({
      shouldFollowup: false,
      isActive: false,
    });

    await expect(runReplyAgent(replyParams)).rejects.toBe(sentinelError);

    expect(followupRun.run.config).toBe(freshCfg);
    expect(resolveQueuedReplyExecutionConfigMock).toHaveBeenCalledTimes(1);
    const [configArg, configContextArg] = requireResolveQueuedReplyExecutionConfigCall();
    expect(configArg).toBe(staleCfg);
    expect(configContextArg.originatingChannel).toBe("telegram");
    expect(configContextArg.messageProvider).toBe("telegram");
    expect(resolveReplyToModeMock).toHaveBeenCalledWith(freshCfg, "telegram", "default", "dm");
    expect(createReplyMediaContextMock).toHaveBeenCalledWith({
      cfg: freshCfg,
      agentId: "main",
      sessionKey: undefined,
      workspaceDir: followupRun.run.workspaceDir,
      messageProvider: "telegram",
      accountId: undefined,
      groupId: undefined,
      groupChannel: undefined,
      groupSpace: undefined,
      requesterSenderId: undefined,
      requesterSenderName: undefined,
      requesterSenderUsername: undefined,
      requesterSenderE164: undefined,
    });
    expect(runSessionCompactionIfNeededMock).toHaveBeenCalledTimes(1);
    expect(runMemoryFlushIfNeededMock).toHaveBeenCalledTimes(1);
    expect(runSessionCompactionIfNeededMock.mock.invocationCallOrder[0]).toBeLessThan(
      runMemoryFlushIfNeededMock.mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY,
    );
    const memoryCall = requireMaintenanceCall(runMemoryFlushIfNeededMock, "runMemoryFlushIfNeeded");
    expect(memoryCall.cfg).toBe(freshCfg);
    expect(memoryCall.followupRun).toBe(followupRun);
    const preflightCall = requireMaintenanceCall(
      runSessionCompactionIfNeededMock,
      "runSessionCompactionIfNeeded",
    );
    expect(preflightCall.cfg).toBe(freshCfg);
    expect(preflightCall.followupRun).toBe(followupRun);
  });

  it("passes the derived runtime-policy key to pre-run maintenance", async () => {
    const { followupRun, replyParams } = createDirectRuntimeReplyParams({
      shouldFollowup: false,
      isActive: false,
    });
    const runtimePolicySessionKey = "agent:main:telegram:default:direct:test";
    followupRun.run.sessionKey = "agent:main:main";
    followupRun.run.runtimePolicySessionKey = runtimePolicySessionKey;
    replyParams.sessionKey = "agent:main:main";
    replyParams.runtimePolicySessionKey = runtimePolicySessionKey;

    await expect(runReplyAgent(replyParams)).rejects.toBe(sentinelError);

    const preflightCall = requireMaintenanceCall(
      runSessionCompactionIfNeededMock,
      "runSessionCompactionIfNeeded",
    );
    expect(preflightCall.sessionKey).toBe("agent:main:main");
    expect(preflightCall.runtimePolicySessionKey).toBe(runtimePolicySessionKey);
    const memoryCall = requireMaintenanceCall(runMemoryFlushIfNeededMock, "runMemoryFlushIfNeeded");
    expect(memoryCall.sessionKey).toBe("agent:main:main");
    expect(memoryCall.runtimePolicySessionKey).toBe(runtimePolicySessionKey);
  });

  it("continues the main reply after a recorded memory-flush failure", async () => {
    const { replyParams } = createDirectRuntimeReplyParams({
      shouldFollowup: false,
      isActive: false,
    });
    const onBlockReply = vi.fn();
    const replyOperation = createReplyOperation();
    replyParams.opts = { sourceReplyDeliveryMode: "message_tool_only", onBlockReply };
    replyParams.replyOperation = replyOperation;
    resolveQueuedReplyExecutionConfigMock.mockResolvedValue({
      ...freshCfg,
      agents: { defaults: { compaction: { notifyUser: true } } },
    });
    runSessionCompactionIfNeededMock.mockImplementation(runRequiredCheckpoint);
    runMemoryFlushIfNeededMock.mockImplementation(
      async (params: {
        replyOperation: ReplyOperation;
        onVisibleErrorPayloads?: (payloads: Array<{ text?: string; isError?: boolean }>) => void;
      }) => {
        params.replyOperation.setPhase("memory_flushing");
        params.onVisibleErrorPayloads?.([
          { text: "⚠️ memory flush preparation failed", isError: true },
        ]);
        return { sessionEntry: undefined, outcome: "failed" };
      },
    );

    const result = await runReplyAgent(replyParams);

    expect(result).toEqual({ text: "main reply" });
    expect(onBlockReply).not.toHaveBeenCalled();
    expect(executeAgentTurnMock).toHaveBeenCalledOnce();
    expect(replyOperation.setPhase).toHaveBeenLastCalledWith("running");
  });

  it("preserves conversation without running byte-forced optional memory work before reply", async () => {
    const memory = await vi.importActual<typeof import("./agent-runner-memory.js")>(
      "./agent-runner-memory.js",
    );
    const reset = await vi.importActual<typeof import("./agent-runner-session-reset.js")>(
      "./agent-runner-session-reset.js",
    );
    runMemoryFlushIfNeededMock.mockImplementation(memory.runMemoryFlushIfNeeded);
    runSessionCompactionIfNeededMock.mockImplementation(memory.runSessionCompactionIfNeeded);
    resetReplyRunSessionMock.mockImplementation(reset.resetReplyRunSession);
    registerMemoryCapability("memory-core", {
      flushPlanResolver: () => ({
        softThresholdTokens: 4_000,
        forceFlushTranscriptBytes: 1,
        reserveTokensFloor: 20_000,
        prompt: "Save durable notes. NO_REPLY",
        systemPrompt: "Write memory to memory/notes.md.",
        relativePath: "memory/notes.md",
      }),
    });
    try {
      await withTestDir({ prefix: "openclaw-direct-runtime-" }, async (tempDir) => {
        const { replyParams, followupRun } = createDirectRuntimeReplyParams({
          shouldFollowup: false,
          isActive: false,
        });
        const sessionKey = "agent:main:telegram:default:direct:test";
        const sessionEntry: SessionEntry = {
          sessionId: "session-1",
          lifecycleRevision: "before-maintenance",
          updatedAt: 1,
          totalTokens: 100,
          totalTokensFresh: true,
          totalTokensVersion: 1,
          compactionCount: 4,
          memoryFlush: { kind: "failed", failureCount: 2 },
        };
        const sessionStore = { [sessionKey]: sessionEntry };
        const storePath = join(tempDir, "sessions.json");
        const scope = { agentId: "main", sessionId: sessionEntry.sessionId, sessionKey, storePath };
        await replaceSessionEntry(scope, sessionEntry);
        const olderMessage = { role: "user", content: "Remember the older marker: amber-orchid" };
        const messages = [
          olderMessage,
          { role: "assistant", content: "Noted." },
          ...Array.from({ length: 3 }, (_, index) => [
            { role: "user", content: `Recent question ${index}` },
            { role: "assistant", content: `Recent answer ${index}` },
          ]).flat(),
        ];
        for (const message of messages) {
          await appendTranscriptMessage(scope, { message });
        }
        // A file blocks the memory directory: optional maintenance fails without token pressure.
        await writeFile(join(tempDir, "memory"), "not a directory");
        followupRun.run.workspaceDir = tempDir;
        followupRun.run.provider = "anthropic";
        followupRun.run.model = "claude-sonnet-4-6";
        replyParams.sessionKey = sessionKey;
        replyParams.storePath = storePath;
        replyParams.sessionEntry = sessionEntry;
        replyParams.sessionStore = sessionStore;
        resolveQueuedReplyExecutionConfigMock.mockResolvedValue(
          withTestModelContextTokens({
            cfg: { agents: { defaults: { compaction: { notifyUser: true } } } },
            followupRun,
            defaultModel: replyParams.defaultModel,
            contextTokens: 100_000,
          }),
        );
        const onBlockReply = vi.fn();
        replyParams.opts = { onBlockReply };
        const replyOperation = createReplyOperation();
        replyParams.replyOperation = replyOperation;
        executeAgentTurnMock.mockImplementation(async () => {
          expect(SessionManager.open(scope).buildSessionContext().messages).toEqual(
            messages.map((message) => expect.objectContaining(message)),
          );
          return {
            runId: "runtime-config-test",
            outcome: { kind: "rejected", payload: { text: "main reply" } },
          };
        });

        const result = await runReplyAgent(replyParams);

        expect(result).toEqual({ text: "main reply" });
        expect(resetReplyRunSessionMock).not.toHaveBeenCalled();
        expect(followupRun.run.sessionId).toBe(sessionEntry.sessionId);
        expect(replyOperation.sessionId).toBe(sessionEntry.sessionId);
        expect(loadSessionEntry(scope)).toMatchObject({
          sessionId: sessionEntry.sessionId,
          lifecycleRevision: sessionEntry.lifecycleRevision,
          compactionCount: 4,
          totalTokens: 100,
          totalTokensFresh: true,
          memoryFlush: { kind: "failed", failureCount: 2 },
        });
        expect(sessionStore[sessionKey]).toMatchObject({
          lifecycleRevision: sessionEntry.lifecycleRevision,
          compactionCount: 4,
          totalTokens: 100,
        });
        expect(runMemoryFlushIfNeededMock).not.toHaveBeenCalled();
        expect(runSessionCompactionIfNeededMock).toHaveBeenCalledOnce();
        expect(onBlockReply).not.toHaveBeenCalled();
        expect(executeAgentTurnMock).toHaveBeenCalledOnce();
      });
    } finally {
      clearMemoryPluginState();
    }
  });

  it("keeps the compacted session when preflight recovers an exhausted memory flush", async () => {
    const { replyParams } = createDirectRuntimeReplyParams({
      shouldFollowup: false,
      isActive: false,
    });
    const sessionEntry = {
      sessionId: "session-1",
      updatedAt: 1,
      compactionCount: 4,
    };
    replyParams.sessionEntry = sessionEntry;
    runMemoryFlushIfNeededMock.mockResolvedValue({
      sessionEntry,
      outcome: "exhausted",
    });
    runSessionCompactionIfNeededMock.mockImplementation(async (params: PreflightParams) => {
      expect(params.sessionEntry?.sessionId).toBe("session-1");
      const checkpointed = await runRequiredCheckpoint(params);
      return { ...checkpointed, compactionCount: 5 };
    });

    await expect(runReplyAgent(replyParams)).resolves.toEqual({ text: "main reply" });

    expect(resetReplyRunSessionMock).not.toHaveBeenCalled();
    expect(executeAgentTurnMock).toHaveBeenCalledOnce();
  });

  it.each(["context_overflow", "auth profile mismatch"])(
    "surfaces required preflight failure (%s) after memory exhaustion without resetting",
    async (reason) => {
      const { replyParams } = createDirectRuntimeReplyParams({
        shouldFollowup: false,
        isActive: false,
      });
      runMemoryFlushIfNeededMock.mockResolvedValue({
        sessionEntry: { sessionId: "session-1", updatedAt: 1, compactionCount: 4 },
        outcome: "exhausted",
      });
      runSessionCompactionIfNeededMock.mockImplementation(async (params: PreflightParams) => {
        await runRequiredCheckpoint(params);
        throw new Error(`Preflight compaction required but failed: ${reason}`);
      });

      const result = await runReplyAgent(replyParams);

      if (!result || Array.isArray(result)) {
        throw new Error("expected a single preflight compaction failure reply payload");
      }
      expect(result.text).toContain("auto-compaction could not recover");
      expect(getReplyPayloadMetadata(result)?.deliverDespiteSourceReplySuppression).toBe(true);
      expect(resetReplyRunSessionMock).not.toHaveBeenCalled();
      expect(executeAgentTurnMock).not.toHaveBeenCalled();
    },
  );

  it.each(["abortByUser", "abortForRestart"] as const)(
    "records %s during memory flush as cancellation without starting the main turn",
    async (abortMethod) => {
      const { replyParams } = createDirectRuntimeReplyParams({
        shouldFollowup: false,
        isActive: false,
      });
      const runState: ReplyOperationRunState = {};
      replyParams.opts = { [REPLY_OPERATION_RUN_STATE]: runState };
      runSessionCompactionIfNeededMock.mockImplementation(runRequiredCheckpoint);
      runMemoryFlushIfNeededMock.mockImplementation(
        async (params: { replyOperation: ReplyOperation }) => {
          expect(params.replyOperation[abortMethod]()).toBe(true);
          return { sessionEntry: undefined, outcome: "failed" };
        },
      );

      const result = await runReplyAgent(replyParams);

      expect(result).toMatchObject({
        text:
          abortMethod === "abortByUser"
            ? SILENT_REPLY_TOKEN
            : "⚠️ Gateway is restarting. Please wait a few seconds and try again.",
      });
      expect(resolveReplyOperationAgentTurn(runState)).toBe("cancelled");
      expect(executeAgentTurnMock).not.toHaveBeenCalled();
    },
  );

  it.each(["user", "restart"] as const)(
    "records an aborted %s agent turn as cancellation",
    async (reason) => {
      const { replyParams } = createDirectRuntimeReplyParams({
        shouldFollowup: false,
        isActive: false,
      });
      const runState: ReplyOperationRunState = {};
      replyParams.opts = { [REPLY_OPERATION_RUN_STATE]: runState };
      runSessionCompactionIfNeededMock.mockResolvedValue(undefined);
      executeAgentTurnMock.mockResolvedValue({
        runId: "cancelled-runtime-config-test",
        outcome: { kind: "aborted", reason },
      });

      await runReplyAgent(replyParams);

      expect(resolveReplyOperationAgentTurn(runState)).toBe("cancelled");
    },
  );

  it("surfaces known pre-run Codex usage-limit failures instead of dropping the reply", async () => {
    const { replyParams } = createDirectRuntimeReplyParams({
      shouldFollowup: false,
      isActive: false,
    });
    const codexMessage =
      "You've reached your Codex subscription usage limit. Codex did not return a reset time for this limit. Run /codex account for current usage details.";
    runSessionCompactionIfNeededMock.mockRejectedValue(
      new FailoverError(codexMessage, {
        reason: "rate_limit",
        provider: "openai",
        model: "gpt-5.5",
      }),
    );
    runMemoryFlushIfNeededMock.mockResolvedValue({ sessionEntry: undefined, outcome: "skipped" });

    const result = await runReplyAgent(replyParams);

    if (!result || Array.isArray(result)) {
      throw new Error("expected a single usage-limit reply payload");
    }
    expect(result.text).toBe(`⚠️ ${codexMessage}`);
    const metadata = getReplyPayloadMetadata(result);
    expect(metadata?.deliverDespiteSourceReplySuppression).toBe(true);
  });

  it("delivers known pre-run OAuth refresh failures instead of dropping the reply", async () => {
    const { replyParams } = createDirectRuntimeReplyParams({
      shouldFollowup: false,
      isActive: false,
    });
    runSessionCompactionIfNeededMock.mockRejectedValue(
      new OAuthRefreshFailureError({
        provider: "openai",
        message: "refresh_token_invalidated",
      }),
    );
    const delivered = vi.fn<(payload: ReplyPayload, kind: ReplyDispatchKind) => void>();
    const dispatcher = createReplyDispatcher({
      deliver: async (payload, { kind }) => {
        delivered(payload, kind);
      },
    });

    const result = await withReplyDispatcher({
      dispatcher,
      run: async () => {
        const payload = await runReplyAgent(replyParams);
        if (!payload || Array.isArray(payload)) {
          throw new Error("expected a single pre-run failure reply payload");
        }
        dispatcher.sendFinalReply(payload);
        return payload;
      },
    });

    const metadata = getReplyPayloadMetadata(result);
    expect(metadata?.deliverDespiteSourceReplySuppression).toBe(true);
    expect(executeAgentTurnMock).not.toHaveBeenCalled();
    expect(delivered).toHaveBeenCalledExactlyOnceWith(
      expect.objectContaining({ text: expect.stringContaining("/login openai"), isError: true }),
      "final",
    );
    expect(delivered.mock.calls[0]?.[0].presentation).toEqual({
      blocks: [
        {
          type: "buttons",
          buttons: [
            {
              label: "Sign in",
              action: { type: "command", command: "/login openai" },
            },
          ],
        },
      ],
    });
  });

  it("surfaces preflight compaction failures before the agent starts", async () => {
    const { replyParams } = createDirectRuntimeReplyParams({
      shouldFollowup: false,
      isActive: false,
    });
    runSessionCompactionIfNeededMock.mockRejectedValue(
      new Error("Preflight compaction required but failed: auth profile mismatch"),
    );
    runMemoryFlushIfNeededMock.mockResolvedValue({ sessionEntry: undefined, outcome: "skipped" });

    const result = await runReplyAgent(replyParams);

    if (!result || Array.isArray(result)) {
      throw new Error("expected a single preflight compaction failure reply payload");
    }
    expect(result.text).toContain("Context is too large");
    expect(result.text).toContain("auto-compaction could not recover");
    expect(result.text).toContain("/compact");
    expect(result.text).toContain("/new");
    const metadata = getReplyPayloadMetadata(result);
    expect(metadata?.deliverDespiteSourceReplySuppression).toBe(true);
  });

  it("does not resolve secrets before the enqueue-followup queue path", async () => {
    const { followupRun, resolvedQueue, replyParams } = createDirectRuntimeReplyParams({
      shouldFollowup: true,
      isActive: true,
    });
    const runState: ReplyOperationRunState = {};
    replyParams.opts = { [REPLY_OPERATION_RUN_STATE]: runState };
    enqueueFollowupRunMock.mockReturnValueOnce(true);

    await expect(runReplyAgent(replyParams)).resolves.toBeUndefined();

    expect(runState.admission).toEqual({ status: "accepted", mode: "followup" });
    expect(resolveQueuedReplyExecutionConfigMock).not.toHaveBeenCalled();
    expect(enqueueFollowupRunMock).toHaveBeenCalledTimes(1);
    const enqueueCall = enqueueFollowupRunMock.mock.calls.at(0);
    expect(enqueueCall?.[0]).toBe("main");
    expect(enqueueCall?.[1]).toBe(followupRun);
    expect(enqueueCall?.[2]).toBe(resolvedQueue);
    expect(enqueueCall?.[3]).toBe("message-id");
    expect(typeof enqueueCall?.[4]).toBe("function");
    expect(enqueueCall?.[5]).toBe(false);
  });
});