src/auto-reply/reply/agent-runner-execute.ts

import crypto from "node:crypto";
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import type { OpenClawConfig } from "../../config/config.js";
import type { SessionEntry } from "../../config/sessions.js";
import { withBeforeAgentReplyObserver } from "../../plugins/before-agent-reply.js";
import { getGatewayContextResolver } from "../../plugins/runtime/gateway-request-scope.js";
import { readPendingUserTurnTranscriptAdmission } from "../../sessions/user-turn-transcript-admission.js";
import { setReplyPayloadMetadata } from "../reply-payload.js";
import type { OriginatingChannelType } from "../templating.js";
import { SILENT_REPLY_TOKEN } from "../tokens.js";
import type { ReplyPayload } from "../types.js";
import {
  resolveReplyRunDeliveryContext,
  resolveSourceReplyPolicy,
  type RunReplyAgentParams,
} from "./agent-runner-core.js";
import { executeAgentTurn } from "./agent-runner-execution.js";
import { markPostCompactionModelFailurePayload } from "./agent-runner-failure-reply.js";
import { runMemoryFlushIfNeeded, runSessionCompactionIfNeeded } from "./agent-runner-memory.js";
import { accountAgentTurnCompaction } from "./agent-runner-result-accounting.js";
import { finalizeReplyAgentRun } from "./agent-runner-result.js";
import { buildThreadingToolContext } from "./agent-runner-utils.js";
import type { BlockReplyPipeline } from "./block-reply-pipeline.js";
import type { CompactionNoticePhase } from "./compaction-notice.js";
import { createFollowupRunner } from "./followup-runner.js";
import {
  buildRecoverablePendingFinalDeliveryText,
  normalizePendingFinalDeliveryPayloads,
} from "./pending-final-delivery.js";
import type { FollowupRun } from "./queue.js";
import type { ReplyMediaContext } from "./reply-media-paths.js";
import { isReplyOperationSuperseded } from "./reply-operation-abort.js";
import { recordReplyOperationAgentTurn } from "./reply-operation-run-state.js";
import type { ReplyOperation } from "./reply-run-registry.js";
import { replyRunRegistry } from "./reply-run-registry.js";
import { resolveReplyToMode } from "./reply-threading.js";
import { createReplyRestartRecoveryClaimController } from "./restart-recovery-claim.js";
import { resolveRoutedDeliveryThreadId } from "./routed-delivery-thread.js";
import type { TypingSignaler } from "./typing-mode.js";
type ExecutePreparedReplyAgentRunInput = Pick<
  RunReplyAgentParams,
  | "blockReplyChunking"
  | "blockStreamingEnabled"
  | "commandBody"
  | "defaultModel"
  | "followupRun"
  | "opts"
  | "queueKey"
  | "replyThreadingOverride"
  | "resolvedBlockStreamingBreak"
  | "resolvedQueue"
  | "resolvedVerboseLevel"
  | "runtimePolicySessionKey"
  | "sessionCtx"
  | "sessionKey"
  | "shouldInjectGroupIntro"
  | "storePath"
  | "toolProgressDetail"
  | "transcriptCommandBody"
  | "typing"
  | "typingMode"
> & {
  activeSessionStore: Record<string, SessionEntry> | undefined;
  admitUserTurn: ReturnType<typeof createReplyRestartRecoveryClaimController>["admitUserTurn"];
  applyReplyToMode: (payload: ReplyPayload) => ReplyPayload;
  beginBeforeAgentReply: ReturnType<
    typeof createReplyRestartRecoveryClaimController
  >["beginBeforeAgentReply"];
  blockReplyPipeline: BlockReplyPipeline | null;
  cfg: OpenClawConfig;
  checkpointBeforeAgentReply: ReturnType<
    typeof createReplyRestartRecoveryClaimController
  >["checkpointBeforeAgentReply"];
  resolveVisibleReplyDelivery: () => Promise<boolean>;
  getActiveIsNewSession: () => boolean;
  getActiveSessionEntry: () => SessionEntry | undefined;
  isHeartbeat: boolean;
  isRestartRecoveryArmed: () => boolean;
  pendingToolTasks: Set<Promise<void>>;
  replyMediaContext: ReplyMediaContext;
  replyOperation: ReplyOperation;
  replyRouteThreadId: ReturnType<typeof resolveRoutedDeliveryThreadId>;
  replyToChannel: OriginatingChannelType | undefined;
  replyToMode: ReturnType<typeof resolveReplyToMode>;
  resetSessionAfterRoleOrderingConflict: (reason: string) => Promise<boolean>;
  returnWithQueuedFollowupDrain: <T>(value: T) => T;
  runFollowupTurn: (queued: FollowupRun) => Promise<void>;
  sendDirectCompactionNotice: ((phase: CompactionNoticePhase) => Promise<void>) | undefined;
  setRunFollowupTurn: (runner: (queued: FollowupRun) => Promise<void>) => void;
  setActiveSessionEntry: (entry: SessionEntry | undefined) => void;
  shouldEmitToolOutput: () => boolean;
  shouldEmitToolResult: () => boolean;
  traceAgentPhase: <T>(name: string, run: () => Promise<T> | T) => Promise<T>;
  turnAdoptionLifecycle: NonNullable<RunReplyAgentParams["opts"]>["turnAdoptionLifecycle"];
  typingSignals: TypingSignaler;
};

function markPostCompactionFailureResult(
  result: ReplyPayload | ReplyPayload[] | undefined,
  postCompactionModelFailure: true | undefined,
): ReplyPayload | ReplyPayload[] | undefined {
  if (Array.isArray(result)) {
    return result.map((payload) =>
      markPostCompactionModelFailurePayload(postCompactionModelFailure, payload),
    );
  }
  return result
    ? markPostCompactionModelFailurePayload(postCompactionModelFailure, result)
    : result;
}

export async function executePreparedReplyAgentRun(
  context: ExecutePreparedReplyAgentRunInput,
): Promise<ReplyPayload | ReplyPayload[] | undefined> {
  const {
    activeSessionStore,
    admitUserTurn: admitUserTurnWithRecovery,
    applyReplyToMode,
    beginBeforeAgentReply: beginBeforeAgentReplyWithRecovery,
    blockReplyChunking,
    blockReplyPipeline,
    blockStreamingEnabled,
    cfg,
    checkpointBeforeAgentReply: checkpointBeforeAgentReplyWithRecovery,
    commandBody,
    defaultModel,
    followupRun,
    getActiveIsNewSession,
    getActiveSessionEntry,
    isHeartbeat,
    isRestartRecoveryArmed,
    opts,
    pendingToolTasks,
    queueKey,
    replyMediaContext,
    replyOperation,
    replyRouteThreadId,
    replyThreadingOverride,
    replyToChannel,
    replyToMode,
    resetSessionAfterRoleOrderingConflict,
    resolvedBlockStreamingBreak,
    resolvedQueue,
    resolvedVerboseLevel,
    returnWithQueuedFollowupDrain,
    runtimePolicySessionKey,
    sendDirectCompactionNotice,
    sessionCtx,
    sessionKey,
    setActiveSessionEntry,
    setRunFollowupTurn,
    shouldEmitToolOutput,
    shouldEmitToolResult,
    shouldInjectGroupIntro,
    storePath,
    toolProgressDetail,
    traceAgentPhase,
    transcriptCommandBody,
    turnAdoptionLifecycle,
    typing,
    typingMode,
    typingSignals,
  } = context;
  let activeSessionEntry = getActiveSessionEntry();
  const admitUserTurn = async (
    ...args: Parameters<typeof admitUserTurnWithRecovery>
  ): ReturnType<typeof admitUserTurnWithRecovery> => {
    const result = await admitUserTurnWithRecovery(...args);
    activeSessionEntry = getActiveSessionEntry();
    return result;
  };
  const beginBeforeAgentReply = async (
    ...args: Parameters<typeof beginBeforeAgentReplyWithRecovery>
  ): ReturnType<typeof beginBeforeAgentReplyWithRecovery> => {
    const result = await beginBeforeAgentReplyWithRecovery(...args);
    activeSessionEntry = getActiveSessionEntry();
    return result;
  };
  const checkpointBeforeAgentReply = async (
    ...args: Parameters<typeof checkpointBeforeAgentReplyWithRecovery>
  ): ReturnType<typeof checkpointBeforeAgentReplyWithRecovery> => {
    const result = await checkpointBeforeAgentReplyWithRecovery(...args);
    activeSessionEntry = getActiveSessionEntry();
    return result;
  };

  await typingSignals.signalRunStart();

  const preflightAdmission = readPendingUserTurnTranscriptAdmission(
    followupRun.userTurnTranscriptRecorder,
  );
  const checkpointMemory = async (entry: SessionEntry) => {
    const flushed = await traceAgentPhase("reply.memory_flush", () =>
      runMemoryFlushIfNeeded({
        preflightAdmission,
        cfg,
        followupRun,
        promptForEstimate: followupRun.prompt,
        opts,
        defaultModel,
        resolvedVerboseLevel,
        sessionEntry: entry,
        sessionStore: activeSessionStore,
        sessionKey,
        runtimePolicySessionKey,
        storePath,
        isHeartbeat,
        replyOperation,
      }),
    );
    setActiveSessionEntry(flushed.sessionEntry);
    replyOperation.abortSignal.throwIfAborted();
    if (flushed.outcome === "exhausted") {
      await sendDirectCompactionNotice?.("memory_flush_degraded");
    }
    return flushed.sessionEntry;
  };

  const prePreflightCompactionCount = activeSessionEntry?.compactionCount ?? 0;
  activeSessionEntry = await traceAgentPhase("reply.preflight_compaction", () =>
    runSessionCompactionIfNeeded({
      pendingUserEntryId: preflightAdmission?.entryId,
      cfg,
      followupRun,
      promptForEstimate: followupRun.prompt,
      defaultModel,
      sessionEntry: activeSessionEntry,
      sessionStore: activeSessionStore,
      sessionKey,
      runtimePolicySessionKey,
      storePath,
      isHeartbeat,
      abortSignal: replyOperation.abortSignal,
      beforeCompaction: checkpointMemory,
      onCompactionStart: () => replyOperation.setPhase("preflight_compacting"),
      onSessionIdChanged: (sessionId) => replyOperation.updateSessionId(sessionId),
      onCompactionNotice: sendDirectCompactionNotice,
    }),
  );
  setActiveSessionEntry(activeSessionEntry);
  const preflightCompactionApplied =
    (activeSessionEntry?.compactionCount ?? 0) > prePreflightCompactionCount;

  const runFollowupTurn = createFollowupRunner({
    resolveGatewayContext: getGatewayContextResolver(replyOperation),
    opts,
    typing,
    typingMode,
    sessionEntry: activeSessionEntry,
    sessionStore: activeSessionStore,
    sessionKey,
    storePath,
    defaultModel,
    toolProgressDetail,
  });
  setRunFollowupTurn(runFollowupTurn);

  replyOperation.setPhase("running");
  const runStartedAt = Date.now();
  const userTurnAdmission = await admitUserTurn(followupRun.userTurnTranscriptRecorder);
  if (userTurnAdmission === "duplicate-source") {
    return returnWithQueuedFollowupDrain(undefined);
  }
  // Adoption marks run start and must never be spool-replayed (would re-run tools).
  // Suppressed delivery persists only the user transcript; crashed suppressed runs die
  // silently. Deliverable turns atomically persist transcript plus recovery ownership.
  await turnAdoptionLifecycle?.onAdopted();
  const runOutcome = await withBeforeAgentReplyObserver(
    {
      beforeDispatch: async () => {
        return await beginBeforeAgentReply();
      },
      afterDispatch: async (hookResult) => {
        if (!hookResult?.handled) {
          await checkpointBeforeAgentReply({ state: undefined });
          return hookResult;
        }
        const hookReply = hookResult.reply ?? { text: SILENT_REPLY_TOKEN };
        const hookFinalDeliveryText = buildRecoverablePendingFinalDeliveryText([hookReply]);
        const normalizedHookReplies = normalizePendingFinalDeliveryPayloads([hookReply]);
        let hookCheckpoint: Parameters<typeof checkpointBeforeAgentReply>[0] = {
          state: normalizedHookReplies.length === 0 ? "handled-silent" : "pending",
        };
        if (sessionKey && storePath && normalizedHookReplies.length > 0) {
          const sourceReplyPolicy = resolveSourceReplyPolicy({
            cfg,
            sessionCtx,
            sessionEntry: activeSessionEntry,
            sessionKey,
            runtimePolicySessionKey,
            opts,
          });
          if (!sourceReplyPolicy.suppressDelivery) {
            const pendingFinalDeliveryIntentId = crypto.randomUUID();
            const pendingFinalDeliveryDeliveryId = crypto.randomUUID();
            setReplyPayloadMetadata(hookReply, {
              pendingFinalDeliveryCompletion: {
                deliveryId: pendingFinalDeliveryDeliveryId,
                intentId: pendingFinalDeliveryIntentId,
                ...(activeSessionEntry?.restartRecoveryDeliveryRunId
                  ? { recoveryRunId: activeSessionEntry.restartRecoveryDeliveryRunId }
                  : {}),
                sessionId: replyOperation.sessionId,
                sessionKey,
                storePath,
              },
            });
            hookCheckpoint = {
              state: "handled-reply",
              pendingFinalDelivery: {
                text: hookFinalDeliveryText ?? "",
                intentId: pendingFinalDeliveryIntentId,
                deliveries: [{ id: pendingFinalDeliveryDeliveryId, state: "prepared" }],
                context: resolveReplyRunDeliveryContext({
                  cfg,
                  sessionCtx,
                  sessionEntry: activeSessionEntry,
                  sessionKey,
                  runtimePolicySessionKey,
                  opts,
                }),
              },
            };
          } else {
            // dispatch-from-config owns source visibility for every returned payload.
            // This checkpoint records that recovery owes no delivery; the outer gate drops the reply.
            hookCheckpoint = { state: "handled-silent" };
          }
        }
        await checkpointBeforeAgentReply(hookCheckpoint);
        return { ...hookResult, reply: hookReply };
      },
    },
    () =>
      traceAgentPhase("reply.run_agent_turn", () =>
        executeAgentTurn({
          commandBody,
          transcriptCommandBody,
          followupRun,
          sessionCtx,
          replyThreading: replyThreadingOverride ?? sessionCtx.ReplyThreading,
          replyOperation,
          opts,
          resolveVisibleReplyDelivery: context.resolveVisibleReplyDelivery,
          typingSignals,
          blockReplyPipeline,
          blockStreamingEnabled,
          blockReplyChunking,
          resolvedBlockStreamingBreak,
          applyReplyToMode,
          shouldEmitToolResult,
          shouldEmitToolOutput,
          pendingToolTasks,
          resetSessionAfterRoleOrderingConflict,
          isHeartbeat,
          sessionKey,
          runtimePolicySessionKey,
          getActiveSessionEntry,
          activeSessionStore,
          storePath,
          resolvedVerboseLevel,
          toolProgressDetail,
          replyMediaContext,
          isRestartRecoveryArmed,
        }),
      ),
  );
  const operationSuperseded = isReplyOperationSuperseded(replyOperation);
  recordReplyOperationAgentTurn(
    followupRun.replyOperationRunStates,
    replyOperation,
    runOutcome.outcome,
  );
  activeSessionEntry = getActiveSessionEntry();
  const activeIsNewSession = getActiveIsNewSession();

  if (runOutcome.outcome.kind !== "settled") {
    // Only captured facts cross cancellation; no successor adoption, hooks, or reply work.
    await accountAgentTurnCompaction({
      compaction: runOutcome.outcome.compaction,
      sessionStore: activeSessionStore,
      replyOperation,
    });
  }
  if (operationSuperseded) {
    return { text: SILENT_REPLY_TOKEN };
  }
  if (runOutcome.outcome.kind !== "settled") {
    if (runOutcome.outcome.kind === "rejected" && !replyOperation.result) {
      replyOperation.fail("run_failed", new Error("reply operation exited with final payload"));
    }
    return returnWithQueuedFollowupDrain(
      runOutcome.outcome.kind === "rejected"
        ? markPostCompactionModelFailurePayload(
            runOutcome.outcome.postCompactionModelFailure,
            runOutcome.outcome.payload,
          )
        : { text: SILENT_REPLY_TOKEN },
    );
  }

  const result = await finalizeReplyAgentRun({
    activeIsNewSession,
    activeSessionEntry,
    activeSessionStore,
    blockReplyPipeline,
    blockStreamingEnabled,
    cfg,
    commandBody,
    defaultModel,
    followupRun,
    isHeartbeat,
    opts,
    pendingToolTasks,
    preflightCompactionApplied,
    queueKey,
    replyMediaContext,
    replyOperation,
    replyRouteThreadId,
    replyThreadingOverride,
    replyToChannel,
    replyToMode,
    resolvedBlockStreamingBreak,
    resolvedQueue,
    resolvedVerboseLevel,
    returnWithQueuedFollowupDrain,
    runFollowupTurn,
    execution: runOutcome.outcome,
    runId: runOutcome.runId,
    runStartedAt,
    runtimePolicySessionKey,
    sessionCtx,
    sessionKey,
    shouldInjectGroupIntro,
    storePath,
    typingSignals,
  });
  return markPostCompactionFailureResult(result, runOutcome.outcome.postCompactionModelFailure);
}

export function createReplyAgentRestartRecoveryController(
  context: Pick<
    RunReplyAgentParams,
    "followupRun" | "opts" | "runtimePolicySessionKey" | "sessionCtx" | "sessionKey" | "storePath"
  > & {
    activeSessionStore: Record<string, SessionEntry> | undefined;
    cfg: OpenClawConfig;
    getActiveSessionEntry: () => SessionEntry | undefined;
    replyOperation: ReplyOperation;
    restartRecoverySourceTurnId: string | undefined;
    setActiveSessionEntry: (entry: SessionEntry) => void;
  },
) {
  const {
    activeSessionStore,
    cfg,
    followupRun,
    getActiveSessionEntry,
    opts,
    replyOperation,
    restartRecoverySourceTurnId,
    runtimePolicySessionKey,
    sessionCtx,
    sessionKey,
    setActiveSessionEntry,
    storePath,
  } = context;

  const restartRecoverySameChannelThreadRequired = restartRecoverySourceTurnId
    ? buildThreadingToolContext({
        sessionCtx,
        config: cfg,
        hasRepliedRef: undefined,
      }).sameChannelThreadRequired
    : undefined;
  const {
    admitUserTurn,
    beginBeforeAgentReply,
    checkpointBeforeAgentReply,
    clear: clearRestartRecoveryDeliveryClaim,
    isArmed: isRestartRecoveryArmed,
  } = createReplyRestartRecoveryClaimController({
    lifecycleGeneration: replyOperation.lifecycleGeneration,
    admissionRunId:
      normalizeOptionalString(sessionCtx.MessageSid) ??
      normalizeOptionalString(sessionCtx.MessageSidFull),
    getEntry: () =>
      sessionKey
        ? (activeSessionStore?.[sessionKey] ?? getActiveSessionEntry())
        : getActiveSessionEntry(),
    getSessionId: () => replyOperation.sessionId,
    isRestartAbort: () =>
      replyOperation.result?.kind === "aborted" &&
      replyOperation.result.code === "aborted_for_restart",
    resolveDeliveryContext: (entry) =>
      sessionKey
        ? resolveReplyRunDeliveryContext({
            cfg,
            sessionCtx,
            sessionEntry: entry,
            sessionKey,
            runtimePolicySessionKey,
            opts,
          })
        : undefined,
    requesterAccountId:
      followupRun.originatingAccountId ?? sessionCtx.AccountId ?? followupRun.run.agentAccountId,
    requesterSenderId: sessionCtx.SenderId,
    resolveUserTurnTarget: ({
      entry,
      sessionId,
      sessionKey: targetSessionKey,
      storePath: targetStorePath,
    }) => ({
      sessionId,
      sessionKey: targetSessionKey,
      sessionEntry: entry,
      ...(activeSessionStore ? { sessionStore: activeSessionStore } : {}),
      storePath: targetStorePath,
      agentId: followupRun.run.agentId,
      cwd: followupRun.run.workspaceDir,
      config: cfg,
    }),
    ...(sessionKey ? { sessionKey } : {}),
    setEntry: (entry) => {
      setActiveSessionEntry(entry);
      if (activeSessionStore && sessionKey) {
        activeSessionStore[sessionKey] = entry;
      }
    },
    sameChannelThreadRequired: restartRecoverySameChannelThreadRequired,
    sourceTurnId: restartRecoverySourceTurnId,
    sourceReplyDeliveryMode: sessionKey
      ? resolveSourceReplyPolicy({
          cfg,
          sessionCtx,
          sessionEntry: getActiveSessionEntry(),
          sessionKey,
          runtimePolicySessionKey,
          opts,
        }).sourceReplyDeliveryMode
      : opts?.sourceReplyDeliveryMode,
    ...(storePath ? { storePath } : {}),
  });
  const admitUserTurnWithSourceBinding: typeof admitUserTurn = async (...args) => {
    const result = await admitUserTurn(...args);
    if (result === "admitted" && restartRecoverySourceTurnId) {
      replyRunRegistry.bindSourceTurnId(replyOperation, restartRecoverySourceTurnId);
    }
    return result;
  };
  return {
    admitUserTurn: admitUserTurnWithSourceBinding,
    beginBeforeAgentReply,
    checkpointBeforeAgentReply,
    clear: clearRestartRecoveryDeliveryClaim,
    isArmed: isRestartRecoveryArmed,
  };
}