src/auto-reply/reply/agent-runner-cli-dispatch.ts

// Builds CLI runtime dispatch inputs for agent runner executions.
import { isRecord } from "@openclaw/normalization-core/record-coerce";
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { runCliAgent } from "../../agents/cli-runner.js";
import type { RunCliAgentParams } from "../../agents/cli-runner/types.js";
import type { MediaImageLayout } from "../../agents/embedded-agent-runner/run/prompt-image-metadata.js";
import { extractToolResultText } from "../../agents/embedded-agent-tool-results.js";
import type { EmbeddedAgentRunResult } from "../../agents/embedded-agent.js";
import {
  DEFAULT_FAST_MODE_AUTO_ON_SECONDS,
  formatFastModeAutoProgressText,
  resolveFastModeForElapsed,
  type FastModeAutoProgressState,
} from "../../agents/fast-mode.js";
import {
  isAgentRunRestartAbortReason,
  resolveAgentRunAbortLifecycleFields,
  resolveAgentRunErrorLifecycleFields,
} from "../../agents/run-termination.js";
import { inferToolMetaFromArgsCore, isCommandBearingToolCall } from "../../agents/tool-display.js";
import { normalizeAgentPlanSteps } from "../../channels/streaming.js";
import type { AgentEventPayload } from "../../infra/agent-events.js";
import { emitAgentEvent, withAgentRunLifecycleGeneration } from "../../infra/agent-events.js";
import { isAgentPlanProgressToolName } from "../../session-cards/progress-card-channel-summary.js";
import { FAST_MODE_AUTO_PROGRESS_KIND, type ReplyPayload } from "../reply-payload.js";
import { formatToolAggregate } from "../tool-meta.js";
import type { GetReplyOptions } from "../types.js";
import {
  type AgentEventDeliveryStartOrder,
  createAgentEventBridge,
  createAgentEventDeliveryStartOrder,
} from "./agent-event-bridge.js";
import { resolveAgentLifecycleTerminalMetadata } from "./agent-lifecycle-terminal.js";

type RunCliAgentInternalParams = RunCliAgentParams & {
  mediaImageLayout?: MediaImageLayout;
};

type AgentEventBridge = {
  unsubscribe: () => void;
  drain: () => Promise<void>;
};

async function stopAgentEventBridges(bridges: readonly AgentEventBridge[]): Promise<void> {
  for (const bridge of bridges) {
    bridge.unsubscribe();
  }
  for (const bridge of bridges) {
    await bridge.drain();
  }
}

function createAssistantTextBridge(params: {
  runId: string;
  suppressed?: boolean;
  deliver?: (text: string) => Promise<boolean | void>;
  startOrder?: AgentEventDeliveryStartOrder;
}) {
  let lastText: string | undefined;
  return createAgentEventBridge({
    runId: params.runId,
    suppressed: params.suppressed,
    deliver: params.deliver,
    startOrder: params.startOrder,
    read: (evt) => {
      if (evt.stream !== "assistant") {
        return undefined;
      }
      const text = typeof evt.data.text === "string" ? evt.data.text : undefined;
      if (text === undefined || text === lastText) {
        return undefined;
      }
      lastText = text;
      return text;
    },
  });
}

type ReasoningTextPayload = {
  text: string;
  isReasoningSnapshot?: boolean;
};

type ReasoningProgressPayload = {
  progressTokens: number;
};

export function createCliReasoningStreamBridge(
  onReasoningStream: GetReplyOptions["onReasoningStream"] | undefined,
): ((payload: ReasoningTextPayload) => Promise<void>) | undefined {
  if (!onReasoningStream) {
    return undefined;
  }
  return async ({ text, isReasoningSnapshot }) => {
    await onReasoningStream({
      text,
      ...(isReasoningSnapshot ? { isReasoningSnapshot } : {}),
      requiresReasoningProgressOptIn: true,
    });
  };
}

function createReasoningTextBridge(params: {
  runId: string;
  suppressed?: boolean;
  deliver?: (payload: ReasoningTextPayload) => Promise<void>;
  startOrder?: AgentEventDeliveryStartOrder;
}) {
  let lastText: string | undefined;
  return createAgentEventBridge({
    runId: params.runId,
    suppressed: params.suppressed,
    deliver: params.deliver,
    startOrder: params.startOrder,
    read: (evt) => {
      if (evt.stream !== "thinking") {
        return undefined;
      }
      const text = typeof evt.data.text === "string" ? evt.data.text : undefined;
      if (text === undefined || text === lastText) {
        return undefined;
      }
      lastText = text;
      return {
        text,
        ...(evt.data.isReasoningSnapshot === true ? { isReasoningSnapshot: true } : {}),
      };
    },
  });
}

function createReasoningProgressBridge(params: {
  runId: string;
  suppressed?: boolean;
  deliver?: (payload: ReasoningProgressPayload) => Promise<void>;
  startOrder?: AgentEventDeliveryStartOrder;
}) {
  let lastProgressTokens: number | undefined;
  return createAgentEventBridge({
    runId: params.runId,
    suppressed: params.suppressed,
    deliver: params.deliver,
    startOrder: params.startOrder,
    read: (evt) => {
      if (evt.stream !== "thinking") {
        return undefined;
      }
      const progressTokens = evt.data.progressTokens;
      if (
        typeof progressTokens !== "number" ||
        !Number.isFinite(progressTokens) ||
        progressTokens <= 0 ||
        progressTokens === lastProgressTokens
      ) {
        return undefined;
      }
      lastProgressTokens = progressTokens;
      return { progressTokens };
    },
  });
}

type CommentaryTextPayload = {
  text: string;
  itemId?: string;
};

function readCommentaryTextPayload(evt: AgentEventPayload): CommentaryTextPayload | undefined {
  if (evt.stream !== "item" || evt.data.kind !== "preamble") {
    return undefined;
  }
  const text = typeof evt.data.progressText === "string" ? evt.data.progressText.trim() : "";
  if (!text) {
    return undefined;
  }
  return {
    text,
    ...(typeof evt.data.itemId === "string" ? { itemId: evt.data.itemId } : {}),
  };
}

type CliToolEventPayload = {
  name: string | undefined;
  phase: "start" | "update" | "result";
  args: Record<string, unknown> | undefined;
  toolCallId?: string;
  isError?: boolean;
  result?: unknown;
};

export function keepCliSessionBindingOnlyWhenReused(params: {
  result: EmbeddedAgentRunResult;
  existingSessionId?: string;
  onDroppedReplacement?: () => void;
}): EmbeddedAgentRunResult {
  const existingSessionId = normalizeOptionalString(params.existingSessionId);
  const agentMeta = params.result.meta.agentMeta;
  const returnedSessionId = normalizeOptionalString(agentMeta?.cliSessionBinding?.sessionId);
  const shouldClearStoredSession = agentMeta?.clearCliSessionBinding === true;
  if (
    agentMeta === undefined ||
    (!shouldClearStoredSession && existingSessionId === undefined) ||
    returnedSessionId === existingSessionId
  ) {
    return params.result;
  }
  if (returnedSessionId || shouldClearStoredSession) {
    params.onDroppedReplacement?.();
  }
  return {
    ...params.result,
    meta: {
      ...params.result.meta,
      agentMeta: {
        ...agentMeta,
        sessionId: "",
        cliSessionBinding: undefined,
        clearCliSessionBinding: undefined,
      },
    },
  };
}

function createToolEventBridge(params: {
  runId: string;
  suppressed?: boolean;
  deliver?: (payload: CliToolEventPayload) => Promise<void>;
  startOrder?: AgentEventDeliveryStartOrder;
}) {
  return createAgentEventBridge({
    runId: params.runId,
    suppressed: params.suppressed,
    deliver: params.deliver,
    startOrder: params.startOrder,
    read: (evt) => {
      if (evt.stream !== "tool") {
        return undefined;
      }
      const phaseValue = evt.data.phase;
      if (phaseValue !== "start" && phaseValue !== "update" && phaseValue !== "result") {
        return undefined;
      }
      const phase: CliToolEventPayload["phase"] =
        phaseValue === "start" ? "start" : phaseValue === "update" ? "update" : "result";
      return {
        name: typeof evt.data.name === "string" ? evt.data.name : undefined,
        phase,
        args: isRecord(evt.data.args) ? evt.data.args : undefined,
        toolCallId: typeof evt.data.toolCallId === "string" ? evt.data.toolCallId : undefined,
        ...(phase === "result"
          ? {
              isError: evt.data.isError === true,
              result: evt.data.result,
            }
          : {}),
      };
    },
  });
}

/**
 * Tracks CLI tool start/result events and renders the same durable tool
 * summaries the embedded runner emits: a formatToolAggregate line per result
 * (args-derived meta captured at start), plus the output block under full
 * verbose. Keeps CLI runs at tool-summary parity with embedded runs.
 */
export function createCliToolSummaryTracker(params: {
  detailMode?: "explain" | "raw";
  commandDetailsVisible: boolean;
  shouldEmitToolResult: () => boolean;
  shouldEmitToolOutput: () => boolean;
  deliver: (payload: { text: string; isError?: boolean }) => Promise<void> | void;
}) {
  const toolByCallId = new Map<string, { name: string; meta?: string; commandBearing: boolean }>();
  return {
    noteToolEvent: async (payload: CliToolEventPayload): Promise<boolean> => {
      if (payload.phase === "start") {
        if (payload.toolCallId && payload.name) {
          toolByCallId.set(payload.toolCallId, {
            name: payload.name,
            meta: inferToolMetaFromArgsCore(payload.name, payload.args, {
              detailMode: params.detailMode ?? "explain",
            }),
            commandBearing: isCommandBearingToolCall(payload.name, payload.args),
          });
        }
        return false;
      }
      if (payload.phase !== "result") {
        return false;
      }
      const storedTool = payload.toolCallId ? toolByCallId.get(payload.toolCallId) : undefined;
      const toolName = payload.name ?? storedTool?.name;
      const meta =
        params.commandDetailsVisible || !storedTool?.commandBearing ? storedTool?.meta : undefined;
      if (payload.toolCallId) {
        toolByCallId.delete(payload.toolCallId);
      }
      if (payload.isError !== true && isAgentPlanProgressToolName(toolName)) {
        return false;
      }
      if (!params.shouldEmitToolResult()) {
        return storedTool?.commandBearing === true;
      }
      const aggregate = formatToolAggregate(toolName, meta ? [meta] : undefined, {
        markdown: true,
      });
      let text = aggregate;
      if (params.shouldEmitToolOutput()) {
        const output = extractToolResultText(payload.result)?.trim();
        if (output) {
          text = `${aggregate}\n\`\`\`txt\n${output}\n\`\`\``;
        }
      }
      if (!text.trim()) {
        return storedTool?.commandBearing === true;
      }
      await params.deliver({ text, ...(payload.isError === true ? { isError: true } : {}) });
      return storedTool?.commandBearing === true;
    },
  };
}

function createCommentaryEventBridge(params: {
  runId: string;
  suppressed?: boolean;
  deliver?: (payload: CommentaryTextPayload) => Promise<void>;
  startOrder?: AgentEventDeliveryStartOrder;
}) {
  return createAgentEventBridge({
    runId: params.runId,
    suppressed: params.suppressed,
    deliver: params.deliver,
    startOrder: params.startOrder,
    read: readCommentaryTextPayload,
  });
}

function createPlanUpdateBridge(params: {
  runId: string;
  suppressed?: boolean;
  deliver?: GetReplyOptions["onPlanUpdate"];
  startOrder?: AgentEventDeliveryStartOrder;
}) {
  const deliver = params.deliver;
  return createAgentEventBridge({
    runId: params.runId,
    suppressed: params.suppressed,
    // GetReplyOptions callbacks may return void; the bridge awaits promises.
    deliver: deliver
      ? async (payload: Parameters<NonNullable<GetReplyOptions["onPlanUpdate"]>>[0]) => {
          await deliver(payload);
        }
      : undefined,
    startOrder: params.startOrder,
    read: (evt) => {
      if (evt.stream !== "plan") {
        return undefined;
      }
      return {
        phase: normalizeOptionalString(evt.data.phase),
        title: normalizeOptionalString(evt.data.title),
        explanation: normalizeOptionalString(evt.data.explanation),
        ...(evt.data.explanationFormat === "plain" ? { explanationFormat: "plain" as const } : {}),
        steps: normalizeAgentPlanSteps(evt.data.steps),
        source: normalizeOptionalString(evt.data.source),
      };
    },
  });
}

function createToolBoundaryBridge(params: {
  runId: string;
  suppressed?: boolean;
  deliver?: () => Promise<void>;
}) {
  return createAgentEventBridge({
    runId: params.runId,
    suppressed: params.suppressed,
    deliver: params.deliver,
    read: (evt) => {
      if (evt.stream !== "tool") {
        return undefined;
      }
      const phase = typeof evt.data.phase === "string" ? evt.data.phase : "";
      return ["completed", "end", "error", "result"].includes(phase) ? true : undefined;
    },
  });
}

type RunCliAgentWithLifecycleParams = {
  runId: string;
  lifecycleGeneration?: string;
  provider: string;
  runParams: RunCliAgentInternalParams;
  startedAt?: number;
  emitLifecycleTerminal?: boolean;
  onAgentRunStart?: () => void;
  suppressAssistantBridge?: boolean;
  /**
   * Stamped before every delivered CLI progress event (assistant, reasoning,
   * tool, commentary, fast-mode). Callers wire this to the reply operation's
   * activity evidence; per-callback stamps at call sites drift and a missed
   * stamp lets stale-takeover reclaim a healthy run.
   */
  onActivity?: () => void;
  preserveProgressCallbackStartOrder?: boolean;
  onAssistantText?: (text: string) => Promise<boolean | void>;
  onReasoningText?: (payload: ReasoningTextPayload) => Promise<void>;
  onReasoningProgress?: (payload: ReasoningProgressPayload) => Promise<void>;
  onCompactionStart?: GetReplyOptions["onCompactionStart"];
  onCompactionEnd?: GetReplyOptions["onCompactionEnd"];
  onToolEvent?: (payload: CliToolEventPayload) => Promise<void>;
  onCommentaryText?: (payload: CommentaryTextPayload) => Promise<void>;
  onPlanUpdate?: GetReplyOptions["onPlanUpdate"];
  onFastModeAutoProgress?: (payload: ReplyPayload) => Promise<void>;
  onErrorBeforeLifecycle?: (err: unknown) => Promise<void>;
  transformResult?: (result: EmbeddedAgentRunResult) => EmbeddedAgentRunResult;
};

export function runCliAgentWithLifecycle(
  params: RunCliAgentWithLifecycleParams,
): Promise<EmbeddedAgentRunResult> {
  if (!params.lifecycleGeneration) {
    return runCliAgentWithLifecycleInternal(params);
  }
  return withAgentRunLifecycleGeneration(params.lifecycleGeneration, () =>
    runCliAgentWithLifecycleInternal(params),
  );
}

async function runCliAgentWithLifecycleInternal(
  params: RunCliAgentWithLifecycleParams,
): Promise<EmbeddedAgentRunResult> {
  const startedAt = params.startedAt ?? Date.now();
  const fastModeStartedAtMs = params.runParams.fastModeStartedAtMs ?? startedAt;
  const fastModeAutoOnSeconds =
    params.runParams.fastModeAutoOnSeconds ?? DEFAULT_FAST_MODE_AUTO_ON_SECONDS;
  const fastModeAutoProgressState: FastModeAutoProgressState = params.runParams
    .fastModeAutoProgressState ?? {
    offAnnounced: false,
    resetAnnounced: false,
  };
  const emitFastModeAutoProgress = async (payload: {
    enabled: boolean;
    elapsedSeconds: number;
    fastAutoOnSeconds?: number;
  }) => {
    const summary = formatFastModeAutoProgressText(payload);
    emitAgentEvent({
      runId: params.runId,
      stream: "item",
      data: {
        kind: "status",
        title: "Fast",
        phase: "update",
        summary,
      },
      ...(params.runParams.sessionKey ? { sessionKey: params.runParams.sessionKey } : {}),
    });
    try {
      await params.onFastModeAutoProgress?.({
        text: summary,
        channelData: { openclawProgressKind: FAST_MODE_AUTO_PROGRESS_KIND },
      });
    } catch {
      // Progress hints are best-effort; a channel failure must not fail the agent turn.
    }
  };
  const maybeAnnounceFastModeAutoOff = async () => {
    if (params.runParams.fastMode !== "auto" || fastModeAutoProgressState.offAnnounced) {
      return;
    }
    const next = resolveFastModeForElapsed({
      mode: "auto",
      startedAtMs: fastModeStartedAtMs,
      fastAutoOnSeconds: fastModeAutoOnSeconds,
    });
    if (next.enabled) {
      return;
    }
    fastModeAutoProgressState.offAnnounced = true;
    await emitFastModeAutoProgress(next);
  };
  const maybeEmitFastModeAutoReset = async () => {
    if (
      params.runParams.fastMode !== "auto" ||
      !fastModeAutoProgressState.offAnnounced ||
      fastModeAutoProgressState.resetAnnounced
    ) {
      return;
    }
    fastModeAutoProgressState.resetAnnounced = true;
    await emitFastModeAutoProgress({
      enabled: true,
      elapsedSeconds: 0,
      fastAutoOnSeconds: fastModeAutoOnSeconds,
    });
  };
  const emitLifecycleTerminal = params.emitLifecycleTerminal ?? true;
  params.onAgentRunStart?.();
  emitAgentEvent({
    runId: params.runId,
    ...(params.runParams.agentId ? { agentId: params.runParams.agentId } : {}),
    ...(params.runParams.sessionKey ? { sessionKey: params.runParams.sessionKey } : {}),
    ...(params.runParams.sessionId ? { sessionId: params.runParams.sessionId } : {}),
    ...(params.lifecycleGeneration ? { lifecycleGeneration: params.lifecycleGeneration } : {}),
    stream: "lifecycle",
    data: {
      phase: "start",
      startedAt,
    },
  });
  // One delivery-independent activity seam for every CLI agent event.
  // Suppressed (silentExpected) runs still emit real events and must keep
  // stamping, or a healthy silent stream looks stale to the takeover window.
  const activityBridge = params.onActivity
    ? createAgentEventBridge<Record<string, never>>({
        runId: params.runId,
        read: () => ({}),
        deliver: async () => {
          params.onActivity?.();
        },
      })
    : undefined;
  const progressStartOrder = params.preserveProgressCallbackStartOrder
    ? createAgentEventDeliveryStartOrder()
    : undefined;
  const assistantBridge = createAssistantTextBridge({
    runId: params.runId,
    suppressed: params.suppressAssistantBridge,
    deliver: params.onAssistantText,
    startOrder: progressStartOrder,
  });
  let finalReasoningText: string | undefined;
  const reasoningBridge = createReasoningTextBridge({
    runId: params.runId,
    suppressed: params.suppressAssistantBridge,
    startOrder: progressStartOrder,
    deliver: async (payload: ReasoningTextPayload) => {
      finalReasoningText = normalizeOptionalString(payload.text);
      await params.onReasoningText?.(payload);
    },
  });
  const reasoningProgressBridge = createReasoningProgressBridge({
    runId: params.runId,
    suppressed: params.suppressAssistantBridge,
    deliver: params.onReasoningProgress,
    startOrder: progressStartOrder,
  });
  const compactionBridge = createAgentEventBridge<
    { phase: "start" } | { completed: boolean; phase: "end" }
  >({
    runId: params.runId,
    suppressed: params.suppressAssistantBridge,
    startOrder: progressStartOrder,
    deliver: async (event) => {
      if (event.phase === "start") {
        await params.onCompactionStart?.();
      } else {
        await params.onCompactionEnd?.({ completed: event.completed });
      }
    },
    read: (evt) => {
      if (evt.stream !== "compaction") {
        return undefined;
      }
      if (evt.data.phase === "start") {
        return { phase: "start" };
      }
      return evt.data.phase === "end"
        ? { phase: "end", completed: evt.data.completed === true }
        : undefined;
    },
  });
  const toolBridge = createToolEventBridge({
    runId: params.runId,
    suppressed: params.suppressAssistantBridge,
    deliver: params.onToolEvent,
    startOrder: progressStartOrder,
  });
  const commentaryBridge = createCommentaryEventBridge({
    runId: params.runId,
    suppressed: params.suppressAssistantBridge,
    deliver: params.onCommentaryText,
    startOrder: progressStartOrder,
  });
  const planBridge = createPlanUpdateBridge({
    runId: params.runId,
    suppressed: params.suppressAssistantBridge,
    deliver: params.onPlanUpdate,
    startOrder: progressStartOrder,
  });
  const toolBoundaryBridge = createToolBoundaryBridge({
    runId: params.runId,
    suppressed: params.suppressAssistantBridge,
    deliver: maybeAnnounceFastModeAutoOff,
  });
  const bridges = [
    activityBridge,
    assistantBridge,
    reasoningBridge,
    reasoningProgressBridge,
    compactionBridge,
    toolBridge,
    commentaryBridge,
    planBridge,
    toolBoundaryBridge,
  ].filter((bridge): bridge is AgentEventBridge => bridge !== undefined);
  let lifecycleTerminalEmitted = false;
  try {
    const rawResult = await runCliAgent({
      ...params.runParams,
      emitCommentaryText: params.runParams.emitCommentaryText ?? Boolean(params.onCommentaryText),
    });
    const restartAbortReason = params.runParams.abortSignal?.reason;
    if (isAgentRunRestartAbortReason(restartAbortReason)) {
      throw restartAbortReason;
    }
    const result = params.transformResult?.(rawResult) ?? rawResult;
    await stopAgentEventBridges(bridges);

    const cliText = normalizeOptionalString(result.payloads?.[0]?.text);
    const durableReasoningText = normalizeOptionalString(finalReasoningText);
    const resultWithReasoning = durableReasoningText
      ? {
          ...result,
          payloads: [{ text: durableReasoningText, isReasoning: true }, ...(result.payloads ?? [])],
        }
      : result;
    if (cliText) {
      emitAgentEvent({
        runId: params.runId,
        stream: "assistant",
        data: { text: cliText },
      });
    }

    if (emitLifecycleTerminal) {
      emitAgentEvent({
        runId: params.runId,
        ...(params.runParams.agentId ? { agentId: params.runParams.agentId } : {}),
        ...(params.runParams.sessionKey ? { sessionKey: params.runParams.sessionKey } : {}),
        ...(params.runParams.sessionId ? { sessionId: params.runParams.sessionId } : {}),
        ...(params.lifecycleGeneration ? { lifecycleGeneration: params.lifecycleGeneration } : {}),
        stream: "lifecycle",
        data: {
          phase: "end",
          startedAt,
          endedAt: Date.now(),
          ...resolveAgentLifecycleTerminalMetadata(result.meta),
          ...resolveAgentRunAbortLifecycleFields(params.runParams.abortSignal),
        },
      });
      lifecycleTerminalEmitted = true;
    }
    return resultWithReasoning;
  } catch (err) {
    await stopAgentEventBridges(bridges);
    await params.onErrorBeforeLifecycle?.(err);
    if (emitLifecycleTerminal) {
      emitAgentEvent({
        runId: params.runId,
        ...(params.runParams.agentId ? { agentId: params.runParams.agentId } : {}),
        ...(params.runParams.sessionKey ? { sessionKey: params.runParams.sessionKey } : {}),
        ...(params.runParams.sessionId ? { sessionId: params.runParams.sessionId } : {}),
        ...(params.lifecycleGeneration ? { lifecycleGeneration: params.lifecycleGeneration } : {}),
        stream: "lifecycle",
        data: {
          phase: "error",
          startedAt,
          endedAt: Date.now(),
          error: String(err),
          ...resolveAgentRunErrorLifecycleFields(err, params.runParams.abortSignal),
        },
      });
      lifecycleTerminalEmitted = true;
    }
    throw err;
  } finally {
    for (const bridge of bridges) {
      bridge.unsubscribe();
    }
    if (params.runParams.isFinalFallbackAttempt !== false) {
      await maybeEmitFastModeAutoReset();
    }
    if (emitLifecycleTerminal && !lifecycleTerminalEmitted) {
      emitAgentEvent({
        runId: params.runId,
        ...(params.runParams.agentId ? { agentId: params.runParams.agentId } : {}),
        ...(params.runParams.sessionKey ? { sessionKey: params.runParams.sessionKey } : {}),
        ...(params.runParams.sessionId ? { sessionId: params.runParams.sessionId } : {}),
        ...(params.lifecycleGeneration ? { lifecycleGeneration: params.lifecycleGeneration } : {}),
        stream: "lifecycle",
        data: {
          phase: "error",
          startedAt,
          endedAt: Date.now(),
          error: "CLI run completed without lifecycle terminal event",
          ...resolveAgentRunAbortLifecycleFields(params.runParams.abortSignal),
        },
      });
    }
  }
}