openclaw/src/commands/agent.ts

import crypto from "node:crypto";
import {
  resolveAgentDir,
  resolveAgentWorkspaceDir,
} from "../agents/agent-scope.js";
import { ensureAuthProfileStore } from "../agents/auth-profiles.js";
import { runCliAgent } from "../agents/cli-runner.js";
import { getCliSessionId, setCliSessionId } from "../agents/cli-session.js";
import { lookupContextTokens } from "../agents/context.js";
import {
  DEFAULT_CONTEXT_TOKENS,
  DEFAULT_MODEL,
  DEFAULT_PROVIDER,
} from "../agents/defaults.js";
import { loadModelCatalog } from "../agents/model-catalog.js";
import { runWithModelFallback } from "../agents/model-fallback.js";
import {
  buildAllowedModelSet,
  isCliProvider,
  modelKey,
  resolveConfiguredModelRef,
  resolveThinkingDefault,
} from "../agents/model-selection.js";
import { runEmbeddedPiAgent } from "../agents/pi-embedded.js";
import { buildWorkspaceSkillSnapshot } from "../agents/skills.js";
import { resolveAgentTimeoutMs } from "../agents/timeout.js";
import { hasNonzeroUsage } from "../agents/usage.js";
import { ensureAgentWorkspace } from "../agents/workspace.js";
import type { MsgContext } from "../auto-reply/templating.js";
import {
  normalizeThinkLevel,
  normalizeVerboseLevel,
  type ThinkLevel,
  type VerboseLevel,
} from "../auto-reply/thinking.js";
import {
  type CliDeps,
  createDefaultDeps,
  createOutboundSendDeps,
} from "../cli/deps.js";
import { type ClawdbotConfig, loadConfig } from "../config/config.js";
import {
  DEFAULT_IDLE_MINUTES,
  loadSessionStore,
  resolveAgentIdFromSessionKey,
  resolveSessionFilePath,
  resolveSessionKey,
  resolveStorePath,
  type SessionEntry,
  saveSessionStore,
} from "../config/sessions.js";
import {
  emitAgentEvent,
  registerAgentRunContext,
} from "../infra/agent-events.js";
import { deliverOutboundPayloads } from "../infra/outbound/deliver.js";
import { buildOutboundResultEnvelope } from "../infra/outbound/envelope.js";
import {
  formatOutboundPayloadLog,
  type NormalizedOutboundPayload,
  normalizeOutboundPayloads,
  normalizeOutboundPayloadsForJson,
} from "../infra/outbound/payloads.js";
import { resolveOutboundTarget } from "../infra/outbound/targets.js";
import {
  getProviderPlugin,
  normalizeProviderId,
} from "../providers/plugins/index.js";
import type { ProviderOutboundTargetMode } from "../providers/plugins/types.js";
import { DEFAULT_CHAT_PROVIDER } from "../providers/registry.js";
import { normalizeMainKey } from "../routing/session-key.js";
import { defaultRuntime, type RuntimeEnv } from "../runtime.js";
import { applyVerboseOverride } from "../sessions/level-overrides.js";
import { resolveSendPolicy } from "../sessions/send-policy.js";
import {
  isInternalMessageProvider,
  resolveGatewayMessageProvider,
  resolveMessageProvider,
} from "../utils/message-provider.js";

/** Image content block for Claude API multimodal messages. */
type ImageContent = {
  type: "image";
  data: string;
  mimeType: string;
};

type AgentCommandOpts = {
  message: string;
  /** Optional image attachments for multimodal messages. */
  images?: ImageContent[];
  to?: string;
  sessionId?: string;
  sessionKey?: string;
  thinking?: string;
  thinkingOnce?: string;
  verbose?: string;
  json?: boolean;
  timeout?: string;
  deliver?: boolean;
  /** Message provider context (webchat|voicewake|whatsapp|...). */
  messageProvider?: string;
  provider?: string; // delivery provider (whatsapp|telegram|...)
  deliveryTargetMode?: ProviderOutboundTargetMode;
  bestEffortDeliver?: boolean;
  abortSignal?: AbortSignal;
  lane?: string;
  runId?: string;
  extraSystemPrompt?: string;
};

type SessionResolution = {
  sessionId: string;
  sessionKey?: string;
  sessionEntry?: SessionEntry;
  sessionStore?: Record<string, SessionEntry>;
  storePath: string;
  isNewSession: boolean;
  persistedThinking?: ThinkLevel;
  persistedVerbose?: VerboseLevel;
};

function resolveSession(opts: {
  cfg: ClawdbotConfig;
  to?: string;
  sessionId?: string;
  sessionKey?: string;
}): SessionResolution {
  const sessionCfg = opts.cfg.session;
  const scope = sessionCfg?.scope ?? "per-sender";
  const mainKey = normalizeMainKey(sessionCfg?.mainKey);
  const idleMinutes = Math.max(
    sessionCfg?.idleMinutes ?? DEFAULT_IDLE_MINUTES,
    1,
  );
  const idleMs = idleMinutes * 60_000;
  const explicitSessionKey = opts.sessionKey?.trim();
  const storeAgentId = resolveAgentIdFromSessionKey(explicitSessionKey);
  const storePath = resolveStorePath(sessionCfg?.store, {
    agentId: storeAgentId,
  });
  const sessionStore = loadSessionStore(storePath);
  const now = Date.now();

  const ctx: MsgContext | undefined = opts.to?.trim()
    ? { From: opts.to }
    : undefined;
  let sessionKey: string | undefined =
    explicitSessionKey ??
    (ctx ? resolveSessionKey(scope, ctx, mainKey) : undefined);
  let sessionEntry = sessionKey ? sessionStore[sessionKey] : undefined;

  // If a session id was provided, prefer to re-use its entry (by id) even when no key was derived.
  if (
    !explicitSessionKey &&
    opts.sessionId &&
    (!sessionEntry || sessionEntry.sessionId !== opts.sessionId)
  ) {
    const foundKey = Object.keys(sessionStore).find(
      (key) => sessionStore[key]?.sessionId === opts.sessionId,
    );
    if (foundKey) {
      sessionKey = sessionKey ?? foundKey;
      sessionEntry = sessionStore[foundKey];
    }
  }

  const fresh = sessionEntry && sessionEntry.updatedAt >= now - idleMs;
  const sessionId =
    opts.sessionId?.trim() ||
    (fresh ? sessionEntry?.sessionId : undefined) ||
    crypto.randomUUID();
  const isNewSession = !fresh && !opts.sessionId;

  const persistedThinking =
    fresh && sessionEntry?.thinkingLevel
      ? normalizeThinkLevel(sessionEntry.thinkingLevel)
      : undefined;
  const persistedVerbose =
    fresh && sessionEntry?.verboseLevel
      ? normalizeVerboseLevel(sessionEntry.verboseLevel)
      : undefined;

  return {
    sessionId,
    sessionKey,
    sessionEntry,
    sessionStore,
    storePath,
    isNewSession,
    persistedThinking,
    persistedVerbose,
  };
}

export async function agentCommand(
  opts: AgentCommandOpts,
  runtime: RuntimeEnv = defaultRuntime,
  deps: CliDeps = createDefaultDeps(),
) {
  const body = (opts.message ?? "").trim();
  if (!body) throw new Error("Message (--message) is required");
  if (!opts.to && !opts.sessionId && !opts.sessionKey) {
    throw new Error("Pass --to <E.164> or --session-id to choose a session");
  }

  const cfg = loadConfig();
  const agentCfg = cfg.agents?.defaults;
  const sessionAgentId = resolveAgentIdFromSessionKey(opts.sessionKey?.trim());
  const workspaceDirRaw = resolveAgentWorkspaceDir(cfg, sessionAgentId);
  const agentDir = resolveAgentDir(cfg, sessionAgentId);
  const workspace = await ensureAgentWorkspace({
    dir: workspaceDirRaw,
    ensureBootstrapFiles: !agentCfg?.skipBootstrap,
  });
  const workspaceDir = workspace.dir;

  const thinkOverride = normalizeThinkLevel(opts.thinking);
  const thinkOnce = normalizeThinkLevel(opts.thinkingOnce);
  if (opts.thinking && !thinkOverride) {
    throw new Error(
      "Invalid thinking level. Use one of: off, minimal, low, medium, high.",
    );
  }
  if (opts.thinkingOnce && !thinkOnce) {
    throw new Error(
      "Invalid one-shot thinking level. Use one of: off, minimal, low, medium, high.",
    );
  }

  const verboseOverride = normalizeVerboseLevel(opts.verbose);
  if (opts.verbose && !verboseOverride) {
    throw new Error('Invalid verbose level. Use "on" or "off".');
  }

  const timeoutSecondsRaw =
    opts.timeout !== undefined
      ? Number.parseInt(String(opts.timeout), 10)
      : undefined;
  if (
    timeoutSecondsRaw !== undefined &&
    (Number.isNaN(timeoutSecondsRaw) || timeoutSecondsRaw <= 0)
  ) {
    throw new Error("--timeout must be a positive integer (seconds)");
  }
  const timeoutMs = resolveAgentTimeoutMs({
    cfg,
    overrideSeconds: timeoutSecondsRaw,
  });

  const sessionResolution = resolveSession({
    cfg,
    to: opts.to,
    sessionId: opts.sessionId,
    sessionKey: opts.sessionKey,
  });

  const {
    sessionId,
    sessionKey,
    sessionEntry: resolvedSessionEntry,
    sessionStore,
    storePath,
    isNewSession,
    persistedThinking,
    persistedVerbose,
  } = sessionResolution;
  let sessionEntry = resolvedSessionEntry;
  const runId = opts.runId?.trim() || sessionId;

  if (opts.deliver === true) {
    const sendPolicy = resolveSendPolicy({
      cfg,
      entry: sessionEntry,
      sessionKey,
      provider: sessionEntry?.provider,
      chatType: sessionEntry?.chatType,
    });
    if (sendPolicy === "deny") {
      throw new Error("send blocked by session policy");
    }
  }

  let resolvedThinkLevel =
    thinkOnce ??
    thinkOverride ??
    persistedThinking ??
    (agentCfg?.thinkingDefault as ThinkLevel | undefined);
  const resolvedVerboseLevel =
    verboseOverride ??
    persistedVerbose ??
    (agentCfg?.verboseDefault as VerboseLevel | undefined);

  if (sessionKey) {
    registerAgentRunContext(runId, {
      sessionKey,
      verboseLevel: resolvedVerboseLevel,
    });
  }

  const needsSkillsSnapshot = isNewSession || !sessionEntry?.skillsSnapshot;
  const skillsSnapshot = needsSkillsSnapshot
    ? buildWorkspaceSkillSnapshot(workspaceDir, { config: cfg })
    : sessionEntry?.skillsSnapshot;

  if (skillsSnapshot && sessionStore && sessionKey && needsSkillsSnapshot) {
    const current = sessionEntry ?? {
      sessionId,
      updatedAt: Date.now(),
    };
    const next: SessionEntry = {
      ...current,
      sessionId,
      updatedAt: Date.now(),
      skillsSnapshot,
    };
    sessionStore[sessionKey] = next;
    await saveSessionStore(storePath, sessionStore);
    sessionEntry = next;
  }

  // Persist explicit /command overrides to the session store when we have a key.
  if (sessionStore && sessionKey) {
    const entry = sessionStore[sessionKey] ??
      sessionEntry ?? { sessionId, updatedAt: Date.now() };
    const next: SessionEntry = { ...entry, sessionId, updatedAt: Date.now() };
    if (thinkOverride) {
      if (thinkOverride === "off") delete next.thinkingLevel;
      else next.thinkingLevel = thinkOverride;
    }
    applyVerboseOverride(next, verboseOverride);
    sessionStore[sessionKey] = next;
    await saveSessionStore(storePath, sessionStore);
  }

  const { provider: defaultProvider, model: defaultModel } =
    resolveConfiguredModelRef({
      cfg,
      defaultProvider: DEFAULT_PROVIDER,
      defaultModel: DEFAULT_MODEL,
    });
  let provider = defaultProvider;
  let model = defaultModel;
  const hasAllowlist =
    agentCfg?.models && Object.keys(agentCfg.models).length > 0;
  const hasStoredOverride = Boolean(
    sessionEntry?.modelOverride || sessionEntry?.providerOverride,
  );
  const needsModelCatalog = hasAllowlist || hasStoredOverride;
  let allowedModelKeys = new Set<string>();
  let allowedModelCatalog: Awaited<ReturnType<typeof loadModelCatalog>> = [];
  let modelCatalog: Awaited<ReturnType<typeof loadModelCatalog>> | null = null;

  if (needsModelCatalog) {
    modelCatalog = await loadModelCatalog({ config: cfg });
    const allowed = buildAllowedModelSet({
      cfg,
      catalog: modelCatalog,
      defaultProvider,
      defaultModel,
    });
    allowedModelKeys = allowed.allowedKeys;
    allowedModelCatalog = allowed.allowedCatalog;
  }

  if (sessionEntry && sessionStore && sessionKey && hasStoredOverride) {
    const overrideProvider =
      sessionEntry.providerOverride?.trim() || defaultProvider;
    const overrideModel = sessionEntry.modelOverride?.trim();
    if (overrideModel) {
      const key = modelKey(overrideProvider, overrideModel);
      if (
        !isCliProvider(overrideProvider, cfg) &&
        allowedModelKeys.size > 0 &&
        !allowedModelKeys.has(key)
      ) {
        delete sessionEntry.providerOverride;
        delete sessionEntry.modelOverride;
        sessionEntry.updatedAt = Date.now();
        sessionStore[sessionKey] = sessionEntry;
        await saveSessionStore(storePath, sessionStore);
      }
    }
  }

  const storedProviderOverride = sessionEntry?.providerOverride?.trim();
  const storedModelOverride = sessionEntry?.modelOverride?.trim();
  if (storedModelOverride) {
    const candidateProvider = storedProviderOverride || defaultProvider;
    const key = modelKey(candidateProvider, storedModelOverride);
    if (
      isCliProvider(candidateProvider, cfg) ||
      allowedModelKeys.size === 0 ||
      allowedModelKeys.has(key)
    ) {
      provider = candidateProvider;
      model = storedModelOverride;
    }
  }
  if (sessionEntry?.authProfileOverride) {
    const store = ensureAuthProfileStore();
    const profile = store.profiles[sessionEntry.authProfileOverride];
    if (!profile || profile.provider !== provider) {
      delete sessionEntry.authProfileOverride;
      sessionEntry.updatedAt = Date.now();
      if (sessionStore && sessionKey) {
        sessionStore[sessionKey] = sessionEntry;
        await saveSessionStore(storePath, sessionStore);
      }
    }
  }

  if (!resolvedThinkLevel) {
    let catalogForThinking = modelCatalog ?? allowedModelCatalog;
    if (!catalogForThinking || catalogForThinking.length === 0) {
      modelCatalog = await loadModelCatalog({ config: cfg });
      catalogForThinking = modelCatalog;
    }
    resolvedThinkLevel = resolveThinkingDefault({
      cfg,
      provider,
      model,
      catalog: catalogForThinking,
    });
  }
  const sessionFile = resolveSessionFilePath(sessionId, sessionEntry, {
    agentId: sessionAgentId,
  });

  const startedAt = Date.now();
  let lifecycleEnded = false;

  let result: Awaited<ReturnType<typeof runEmbeddedPiAgent>>;
  let fallbackProvider = provider;
  let fallbackModel = model;
  try {
    const messageProvider = resolveMessageProvider(
      opts.messageProvider,
      opts.provider,
    );
    const fallbackResult = await runWithModelFallback({
      cfg,
      provider,
      model,
      run: (providerOverride, modelOverride) => {
        if (isCliProvider(providerOverride, cfg)) {
          const cliSessionId = getCliSessionId(sessionEntry, providerOverride);
          return runCliAgent({
            sessionId,
            sessionKey,
            sessionFile,
            workspaceDir,
            config: cfg,
            prompt: body,
            provider: providerOverride,
            model: modelOverride,
            thinkLevel: resolvedThinkLevel,
            timeoutMs,
            runId,
            extraSystemPrompt: opts.extraSystemPrompt,
            cliSessionId,
            images: opts.images,
          });
        }
        return runEmbeddedPiAgent({
          sessionId,
          sessionKey,
          messageProvider,
          sessionFile,
          workspaceDir,
          config: cfg,
          skillsSnapshot,
          prompt: body,
          images: opts.images,
          provider: providerOverride,
          model: modelOverride,
          authProfileId: sessionEntry?.authProfileOverride,
          thinkLevel: resolvedThinkLevel,
          verboseLevel: resolvedVerboseLevel,
          timeoutMs,
          runId,
          lane: opts.lane,
          abortSignal: opts.abortSignal,
          extraSystemPrompt: opts.extraSystemPrompt,
          agentDir,
          onAgentEvent: (evt) => {
            if (
              evt.stream === "lifecycle" &&
              typeof evt.data?.phase === "string" &&
              (evt.data.phase === "end" || evt.data.phase === "error")
            ) {
              lifecycleEnded = true;
            }
            emitAgentEvent({
              runId,
              stream: evt.stream,
              data: evt.data,
            });
          },
        });
      },
    });
    result = fallbackResult.result;
    fallbackProvider = fallbackResult.provider;
    fallbackModel = fallbackResult.model;
    if (!lifecycleEnded) {
      emitAgentEvent({
        runId,
        stream: "lifecycle",
        data: {
          phase: "end",
          startedAt,
          endedAt: Date.now(),
          aborted: result.meta.aborted ?? false,
        },
      });
    }
  } catch (err) {
    if (!lifecycleEnded) {
      emitAgentEvent({
        runId,
        stream: "lifecycle",
        data: {
          phase: "error",
          startedAt,
          endedAt: Date.now(),
          error: String(err),
        },
      });
    }
    throw err;
  }

  // Update token+model fields in the session store.
  if (sessionStore && sessionKey) {
    const usage = result.meta.agentMeta?.usage;
    const modelUsed = result.meta.agentMeta?.model ?? fallbackModel ?? model;
    const providerUsed =
      result.meta.agentMeta?.provider ?? fallbackProvider ?? provider;
    const contextTokens =
      agentCfg?.contextTokens ??
      lookupContextTokens(modelUsed) ??
      DEFAULT_CONTEXT_TOKENS;

    const entry = sessionStore[sessionKey] ?? {
      sessionId,
      updatedAt: Date.now(),
    };
    const next: SessionEntry = {
      ...entry,
      sessionId,
      updatedAt: Date.now(),
      modelProvider: providerUsed,
      model: modelUsed,
      contextTokens,
    };
    if (isCliProvider(providerUsed, cfg)) {
      const cliSessionId = result.meta.agentMeta?.sessionId?.trim();
      if (cliSessionId) setCliSessionId(next, providerUsed, cliSessionId);
    }
    next.abortedLastRun = result.meta.aborted ?? false;
    if (hasNonzeroUsage(usage)) {
      const input = usage.input ?? 0;
      const output = usage.output ?? 0;
      const promptTokens =
        input + (usage.cacheRead ?? 0) + (usage.cacheWrite ?? 0);
      next.inputTokens = input;
      next.outputTokens = output;
      next.totalTokens =
        promptTokens > 0 ? promptTokens : (usage.total ?? input);
    }
    sessionStore[sessionKey] = next;
    await saveSessionStore(storePath, sessionStore);
  }

  const payloads = result.payloads ?? [];
  const deliver = opts.deliver === true;
  const bestEffortDeliver = opts.bestEffortDeliver === true;
  const deliveryProvider =
    resolveGatewayMessageProvider(opts.provider) ?? DEFAULT_CHAT_PROVIDER;
  // Provider docking: delivery providers are resolved via plugin registry.
  const deliveryPlugin = !isInternalMessageProvider(deliveryProvider)
    ? getProviderPlugin(
        normalizeProviderId(deliveryProvider) ?? deliveryProvider,
      )
    : undefined;

  const logDeliveryError = (err: unknown) => {
    const message = `Delivery failed (${deliveryProvider}${deliveryTarget ? ` to ${deliveryTarget}` : ""}): ${String(err)}`;
    runtime.error?.(message);
    if (!runtime.error) runtime.log(message);
  };

  const isDeliveryProviderKnown =
    isInternalMessageProvider(deliveryProvider) || Boolean(deliveryPlugin);

  const targetMode: ProviderOutboundTargetMode =
    opts.deliveryTargetMode ?? (opts.to ? "explicit" : "implicit");
  const resolvedTarget =
    deliver && isDeliveryProviderKnown && deliveryProvider
      ? resolveOutboundTarget({
          provider: deliveryProvider,
          to: opts.to,
          cfg,
          accountId:
            targetMode === "implicit" ? sessionEntry?.lastAccountId : undefined,
          mode: targetMode,
        })
      : null;
  const deliveryTarget = resolvedTarget?.ok ? resolvedTarget.to : undefined;

  if (deliver) {
    if (!isDeliveryProviderKnown) {
      const err = new Error(`Unknown provider: ${deliveryProvider}`);
      if (!bestEffortDeliver) throw err;
      logDeliveryError(err);
    } else if (resolvedTarget && !resolvedTarget.ok) {
      if (!bestEffortDeliver) throw resolvedTarget.error;
      logDeliveryError(resolvedTarget.error);
    }
  }

  const normalizedPayloads = normalizeOutboundPayloadsForJson(payloads);
  if (opts.json) {
    runtime.log(
      JSON.stringify(
        buildOutboundResultEnvelope({
          payloads: normalizedPayloads,
          meta: result.meta,
        }),
        null,
        2,
      ),
    );
    if (!deliver) {
      return { payloads: normalizedPayloads, meta: result.meta };
    }
  }

  if (payloads.length === 0) {
    runtime.log("No reply from agent.");
    return { payloads: [], meta: result.meta };
  }

  const deliveryPayloads = normalizeOutboundPayloads(payloads);
  const logPayload = (payload: NormalizedOutboundPayload) => {
    if (opts.json) return;
    const output = formatOutboundPayloadLog(payload);
    if (output) runtime.log(output);
  };
  if (!deliver) {
    for (const payload of deliveryPayloads) {
      logPayload(payload);
    }
  }
  if (
    deliver &&
    deliveryProvider &&
    !isInternalMessageProvider(deliveryProvider)
  ) {
    if (deliveryTarget) {
      await deliverOutboundPayloads({
        cfg,
        provider: deliveryProvider,
        to: deliveryTarget,
        payloads: deliveryPayloads,
        bestEffort: bestEffortDeliver,
        onError: (err) => logDeliveryError(err),
        onPayload: logPayload,
        deps: createOutboundSendDeps(deps, cfg),
      });
    }
  }

  return { payloads: normalizedPayloads, meta: result.meta };
}