src/agents/command/acp-execution.ts

import { createLazyAcpElicitationHandler } from "../../auto-reply/reply/acp-elicitation-handler-lazy.js";
import { resolveInlineAgentImageAttachments } from "../../auto-reply/reply/agent-turn-attachments.js";
import { recordAgentRunTerminalOutcome } from "../../channels/turn/agent-run-terminal-outcome.js";
import type { CliDeps } from "../../cli/deps.types.js";
import type { SessionEntry } from "../../config/sessions/types.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { assertAgentRunLifecycleGenerationCurrent } from "../../infra/agent-events.js";
import { registerAgentRunContext } from "../../infra/agent-run-registry.js";
import { formatErrorMessage } from "../../infra/errors.js";
import {
  getInstallationTarget,
  LOCAL_INSTALLATION_TARGET_UNSUPPORTED,
} from "../../infra/installation-target-context.js";
import { createSubsystemLogger } from "../../logging/subsystem.js";
import { normalizeAgentId, resolveAgentIdFromSessionKey } from "../../routing/session-key.js";
import type { RuntimeEnv } from "../../runtime.js";
import {
  getAdmittedRunDelegatedAuthority,
  type PreparedAgentRunAdmission,
} from "../admitted-run-context.js";
import {
  buildAgentRunTerminalOutcomeFromLifecycleEvent,
  classifyAgentRunTerminalOutcome,
} from "../agent-run-terminal-outcome.js";
import { prepareInternalSessionEffectsSession } from "../internal-session-effects.js";
import type { AgentRunSessionTarget } from "../run-session-target.js";
import { isAgentRunRestartAbortReason } from "../run-termination.js";
import { applyAgentRunAbortMetadata } from "./lifecycle.js";
import type { PreparedAgentCommandExecution } from "./prepare.js";
import {
  loadAcpPolicyRuntime,
  loadAcpRuntimeErrorsRuntime,
  loadAcpSessionIdentifiersRuntime,
  loadAttemptExecutionRuntime,
  loadDeliveryRuntime,
} from "./runtime-loaders.js";
import { resolveInternalSessionEffectsSource } from "./session-helpers.js";
import type { AgentCommandOpts } from "./types.js";

const log = createSubsystemLogger("agents/agent-command");

type AcpReadyResolution = Extract<
  NonNullable<PreparedAgentCommandExecution["acpResolution"]>,
  { kind: "ready" }
>;

export async function runAcpAgentCommand(params: {
  preparedRunAdmission: PreparedAgentRunAdmission;
  cfg: OpenClawConfig;
  deps: CliDeps;
  runtime: RuntimeEnv;
  opts: AgentCommandOpts;
  outboundSession: PreparedAgentCommandExecution["outboundSession"];
  sessionEntry?: SessionEntry;
  sessionStore?: Record<string, SessionEntry>;
  body: string;
  transcriptBody: string;
  suppressVisibleSessionEffects: boolean;
  provenance: "agent" | "human" | "system";
  sessionAgentId: string;
  sessionId: string;
  sessionKey: string;
  storePath: string;
  workspaceDir: string;
  runId: string;
  lifecycleGeneration: string;
  acpManager: PreparedAgentCommandExecution["acpManager"];
  acpResolution: AcpReadyResolution;
  trackInternalModelRunTarget: (target: AgentRunSessionTarget | undefined) => void;
}) {
  if (getInstallationTarget()) {
    throw new Error(LOCAL_INSTALLATION_TARGET_UNSUPPORTED);
  }
  const attemptExecutionRuntime = await loadAttemptExecutionRuntime();
  const acpToolTracker = attemptExecutionRuntime.createAcpToolLifecycleTracker();
  const startedAt = Date.now();
  registerAgentRunContext(params.runId, {
    sessionKey: params.sessionKey,
    sessionId: params.sessionId,
    agentId: params.sessionAgentId,
    lifecycleGeneration: params.lifecycleGeneration,
    projectSessionActive: !params.suppressVisibleSessionEffects,
    ...(params.suppressVisibleSessionEffects ? { isControlUiVisible: false } : {}),
  });
  attemptExecutionRuntime.emitAcpLifecycleStart({
    runId: params.runId,
    startedAt,
    agentId: params.sessionAgentId,
    lifecycleGeneration: params.lifecycleGeneration,
  });

  const visibleTextAccumulator = attemptExecutionRuntime.createAcpVisibleTextAccumulator();
  let stopReason: string | undefined;
  let resultStatus: "completed" | "cancelled" | undefined;
  let terminalOutcome: "blocked" | undefined;
  try {
    const {
      resolveAcpAgentPolicyError,
      resolveAcpDispatchPolicyError,
      resolveAcpExplicitTurnPolicyError,
    } = await loadAcpPolicyRuntime();
    const turnPolicyError =
      params.opts.acpTurnSource === "manual_spawn"
        ? resolveAcpExplicitTurnPolicyError(params.cfg)
        : resolveAcpDispatchPolicyError(params.cfg);
    if (turnPolicyError) {
      terminalOutcome = "blocked";
      throw turnPolicyError;
    }
    const acpAgent = normalizeAgentId(
      params.acpResolution.meta.agent || resolveAgentIdFromSessionKey(params.sessionKey),
    );
    const agentPolicyError = resolveAcpAgentPolicyError(params.cfg, acpAgent);
    if (agentPolicyError) {
      terminalOutcome = "blocked";
      throw agentPolicyError;
    }

    const acpImageAttachments = resolveInlineAgentImageAttachments(params.opts.images);
    assertAgentRunLifecycleGenerationCurrent(params.lifecycleGeneration);
    const admittedRunContext = await params.preparedRunAdmission.admit("acp");
    const isElicitationActive = () => {
      if (
        params.opts.abortSignal?.aborted === true ||
        getAdmittedRunDelegatedAuthority(admittedRunContext) === undefined
      ) {
        return false;
      }
      try {
        assertAgentRunLifecycleGenerationCurrent(params.lifecycleGeneration);
        return true;
      } catch {
        return false;
      }
    };
    const onElicitation = createLazyAcpElicitationHandler({
      sourceSessionKey: params.opts.inputProvenance?.sourceSessionKey ?? params.sessionKey,
      targetSessionKey: params.sessionKey,
      outerRequestId: params.runId,
      agentId: params.sessionAgentId,
      runId: params.runId,
      delivery: {
        deliver: async (_kind, payload) => {
          if (!isElicitationActive()) {
            throw new Error("ACP input request is no longer active.");
          }
          if (payload.text) {
            attemptExecutionRuntime.emitAcpRuntimeEvent({
              runId: params.runId,
              toolTracker: acpToolTracker,
              sessionKey: params.sessionKey,
              agentId: params.sessionAgentId,
              abortSignal: params.opts.abortSignal,
              event: { type: "status", text: payload.text, tag: "elicitation" },
            });
          }
          return true;
        },
      },
      isActive: isElicitationActive,
    });
    await params.acpManager.runTurn({
      admittedRunContext,
      cfg: params.cfg,
      sessionKey: params.sessionKey,
      agentId: params.sessionAgentId,
      provenance: params.provenance,
      text: params.body,
      attachments: acpImageAttachments.length > 0 ? acpImageAttachments : undefined,
      mode: "prompt",
      requestId: params.runId,
      signal: params.opts.abortSignal,
      onElicitation,
      onBeforePrompt: async () => {
        const recorder = params.opts.userTurnTranscriptRecorder;
        if (recorder && !recorder.hasPersisted() && !(await recorder.persistApproved())) {
          throw new Error("ACP input could not enter the session transcript");
        }
        params.opts.onExecutionStarted?.();
      },
      onLifecycle: (event) => {
        if (event.type === "prompt_submitted") {
          attemptExecutionRuntime.emitAcpPromptSubmitted({
            runId: params.runId,
            sessionKey: params.sessionKey,
            at: event.at,
          });
        }
      },
      onEvent: (event) => {
        if (event.type !== "text_delta") {
          attemptExecutionRuntime.emitAcpRuntimeEvent({
            runId: params.runId,
            toolTracker: acpToolTracker,
            sessionKey: params.sessionKey,
            agentId: params.sessionAgentId,
            abortSignal: params.opts.abortSignal,
            event,
          });
        }
        if (event.type === "done") {
          stopReason = event.stopReason;
          resultStatus = event.status;
          return;
        }
        if (
          event.type !== "text_delta" ||
          (event.stream && event.stream !== "output") ||
          !event.text
        ) {
          return;
        }
        const visibleUpdate = visibleTextAccumulator.consume(event.text);
        if (visibleUpdate) {
          attemptExecutionRuntime.emitAcpAssistantDelta({
            runId: params.runId,
            text: visibleUpdate.text,
            delta: visibleUpdate.delta,
          });
        }
      },
    });
    if (isAgentRunRestartAbortReason(params.opts.abortSignal?.reason)) {
      throw params.opts.abortSignal?.reason;
    }
  } catch (error) {
    const { toAcpRuntimeError } = await loadAcpRuntimeErrorsRuntime();
    const acpError = toAcpRuntimeError({
      error,
      fallbackCode: "ACP_TURN_FAILED",
      fallbackMessage: "ACP turn failed before completion.",
    });
    attemptExecutionRuntime.emitAcpLifecycleError({
      runId: params.runId,
      toolTracker: acpToolTracker,
      error: acpError,
      sessionKey: params.sessionKey,
      agentId: params.sessionAgentId,
      lifecycleGeneration: params.lifecycleGeneration,
      abortSignal: params.opts.abortSignal,
      ...(terminalOutcome ? { terminalOutcome } : {}),
    });
    throw acpError;
  }

  // Execution is settled before persistence; later delivery cancellation remains live below.
  const endFields = attemptExecutionRuntime.resolveAcpLifecycleEndFields(
    params.opts.abortSignal,
    stopReason,
    resultStatus,
  );
  const finalTextRaw = visibleTextAccumulator.finalizeRaw();
  const finalText = visibleTextAccumulator.finalize();
  const terminalReply = visibleTextAccumulator.finalizeReplySnapshot();
  let sessionEntry = params.sessionEntry;
  try {
    const { resolveAcpSessionCwd } = await loadAcpSessionIdentifiersRuntime();
    const internalSource = params.suppressVisibleSessionEffects
      ? resolveInternalSessionEffectsSource({
          agentId: params.sessionAgentId,
          sessionId: params.sessionId,
          sessionKey: params.sessionKey,
          storePath: params.storePath,
        })
      : undefined;
    const internalTarget = params.suppressVisibleSessionEffects
      ? await prepareInternalSessionEffectsSession({
          agentId: params.sessionAgentId,
          cwd: resolveAcpSessionCwd(params.acpResolution.meta) ?? params.workspaceDir,
          runId: params.runId,
          source: internalSource,
          storePath: params.storePath,
        })
      : undefined;
    params.trackInternalModelRunTarget(internalTarget);
    const transcriptResult = await attemptExecutionRuntime.persistAcpTurnTranscript({
      body: params.body,
      transcriptBody: params.transcriptBody,
      userTurnTranscriptRecorder: params.opts.userTurnTranscriptRecorder,
      ...(!params.opts.userTurnTranscriptRecorder &&
      params.opts.suppressPromptPersistence !== true &&
      params.opts.transcriptMedia?.length
        ? {
            userInput: {
              text: params.transcriptBody,
              media: params.opts.transcriptMedia,
            },
          }
        : {}),
      finalText: finalTextRaw,
      terminalOutcome: buildAgentRunTerminalOutcomeFromLifecycleEvent({
        phase: "end",
        data: endFields,
      }),
      sessionId: internalTarget?.sessionId ?? params.sessionId,
      sessionKey: internalTarget?.sessionKey ?? params.sessionKey,
      sessionEntry: internalTarget?.sessionEntry ?? sessionEntry,
      sessionStore: params.suppressVisibleSessionEffects ? undefined : params.sessionStore,
      storePath: internalTarget?.storePath ?? params.storePath,
      sessionAgentId: internalTarget?.agentId ?? params.sessionAgentId,
      threadId: params.opts.threadId,
      sessionCwd: resolveAcpSessionCwd(params.acpResolution.meta) ?? params.workspaceDir,
      config: params.cfg,
    });
    if (!internalTarget) {
      sessionEntry = transcriptResult.sessionEntry;
    }
  } catch (error) {
    log.warn(
      `ACP transcript persistence failed for ${params.sessionKey}: ${formatErrorMessage(error)}`,
    );
  }
  const restartAbortReason = params.opts.abortSignal?.reason;
  if (isAgentRunRestartAbortReason(restartAbortReason)) {
    attemptExecutionRuntime.emitAcpLifecycleError({
      runId: params.runId,
      toolTracker: acpToolTracker,
      error: restartAbortReason,
      sessionKey: params.sessionKey,
      agentId: params.sessionAgentId,
      lifecycleGeneration: params.lifecycleGeneration,
      abortSignal: params.opts.abortSignal,
    });
    throw restartAbortReason;
  }
  attemptExecutionRuntime.emitAcpLifecycleEnd({
    runId: params.runId,
    toolTracker: acpToolTracker,
    agentId: params.sessionAgentId,
    lifecycleGeneration: params.lifecycleGeneration,
    endFields,
    terminalReply,
  });

  const result = applyAgentRunAbortMetadata(
    attemptExecutionRuntime.buildAcpResult({
      payloadText: finalText,
      terminalReply,
      startedAt,
      stopReason,
      resultStatus,
      abortSignal: params.opts.abortSignal,
    }),
    params.opts.abortSignal,
  );
  const { deliverAgentCommandResult } = await loadDeliveryRuntime();
  const deliveryResult = await deliverAgentCommandResult({
    cfg: params.cfg,
    deps: params.deps,
    runtime: params.runtime,
    opts: params.opts,
    outboundSession: params.outboundSession,
    sessionEntry,
    result,
    payloads: result.payloads,
    assertDeliveryCurrent: () =>
      assertAgentRunLifecycleGenerationCurrent(params.lifecycleGeneration),
  });
  // Use the owner's status and current signal: delivery may outlive the result snapshot.
  const outcome = buildAgentRunTerminalOutcomeFromLifecycleEvent({
    phase: "end",
    data: { status: resultStatus, stopReason },
    abortSignal: params.opts.abortSignal,
  });
  return recordAgentRunTerminalOutcome(
    deliveryResult,
    classifyAgentRunTerminalOutcome(outcome) === "success" ? "completed" : "failed",
  );
}