src/acp/control-plane/manager.runtime-controls.ts

/** Applies runtime mode/config controls to live ACP backend sessions. */
import type {
  AcpRuntime,
  AcpRuntimeCapabilities,
  AcpRuntimeHandle,
  AcpRuntimeStatus,
} from "@openclaw/acp-core/runtime/types";
import { asNullableRecord } from "@openclaw/normalization-core/record-coerce";
import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce";
import {
  AcpRuntimeError,
  formatAcpErrorChain,
  toAcpRuntimeError,
  withAcpRuntimeErrorBoundary,
} from "../runtime/errors.js";
import type { CachedRuntimeState } from "./manager.runtime-handle-cache.js";
import { isAcpOwnerRepairRequired } from "./manager.runtime-owner.js";
import type { AcpSessionRuntimeOptions, SessionAcpMeta } from "./manager.types.js";
import { createUnsupportedControlError } from "./manager.utils.js";
import {
  buildRuntimeConfigOptionPairs,
  buildRuntimeControlSignature,
  isThinkingConfigKey,
  normalizeText,
  reconcileAcceptedRuntimeOptions,
  resolveRuntimeConfigOptionKey,
  resolveRuntimeOptionsFromMeta,
  runtimeOptionsEqual,
} from "./runtime-options.js";

const OPTIONAL_TIMEOUT_CONFIG_KEYS = new Set(["timeout", "timeout_seconds"]);
const ACP_CONFIG_REJECTION_CODE_RE = /-3260[23]/;
const CONFIG_OPTION_REJECTION_RE =
  /invalid params|unsupported|not supported|not implement|invalid value|unknown config option|unknown value|not a valid value|must be one of/;

function extractConfigOptionKeys(value: unknown): string[] {
  if (!Array.isArray(value)) {
    return [];
  }
  return value
    .map((entry) => {
      if (typeof entry === "string") {
        return normalizeText(entry);
      }
      const record = asNullableRecord(entry);
      return normalizeText(record?.id ?? record?.key);
    })
    .filter(Boolean) as string[];
}

function extractRuntimeStatusConfigOptionKeys(status: AcpRuntimeStatus | undefined): string[] {
  const details = asNullableRecord(status?.details);
  return [
    ...extractConfigOptionKeys(details?.configOptions),
    ...extractConfigOptionKeys(details?.config_options),
  ];
}

function isOptionalTimeoutConfigKey(key: string): boolean {
  return OPTIONAL_TIMEOUT_CONFIG_KEYS.has(normalizeLowercaseStringOrEmpty(key));
}

function isUnsupportedControlRejection(error: unknown): boolean {
  const errorCode = error && typeof error === "object" ? (error as { code?: unknown }).code : null;
  return errorCode === "ACP_BACKEND_UNSUPPORTED_CONTROL";
}

function describeConfigOptionRejection(error: unknown): string {
  const described = toAcpRuntimeError({
    error,
    fallbackCode: "ACP_TURN_FAILED",
    fallbackMessage: "",
  });
  return normalizeLowercaseStringOrEmpty(`${formatAcpErrorChain(error)} ${described.message}`);
}

function isUnsupportedOptionalTimeoutConfigRejection(key: string, error: unknown): boolean {
  if (!isOptionalTimeoutConfigKey(key)) {
    return false;
  }
  if (isUnsupportedControlRejection(error)) {
    return true;
  }
  const message =
    error instanceof Error ? error.message : typeof error === "string" ? error : String(error);
  const normalized = normalizeLowercaseStringOrEmpty(message);
  return (
    normalized.includes("session/set_config_option") &&
    (normalized.includes("-32602") ||
      normalized.includes("invalid params") ||
      normalized.includes("unsupported") ||
      normalized.includes("not supported") ||
      normalized.includes("not implement"))
  );
}

function isRejectedThinkingConfigOption(key: string, error: unknown): boolean {
  if (!isThinkingConfigKey(key)) {
    return false;
  }
  if (isUnsupportedControlRejection(error)) {
    return true;
  }
  const description = describeConfigOptionRejection(error);
  const describesConfigOption =
    description.includes("session/set_config_option") ||
    (description.includes("config option") &&
      description.includes(normalizeLowercaseStringOrEmpty(key)));
  return (
    describesConfigOption &&
    ACP_CONFIG_REJECTION_CODE_RE.test(description) &&
    CONFIG_OPTION_REJECTION_RE.test(description)
  );
}

/** Resolves backend-advertised controls plus locally inferred runtime control support. */
export async function resolveManagerRuntimeCapabilities(params: {
  runtime: AcpRuntime;
  handle: AcpRuntimeHandle;
  includeStatusConfigOptionKeys?: boolean;
}): Promise<AcpRuntimeCapabilities> {
  let reported: AcpRuntimeCapabilities | undefined;
  if (params.runtime.getCapabilities) {
    reported = await withAcpRuntimeErrorBoundary({
      run: async () => await params.runtime.getCapabilities!({ handle: params.handle }),
      fallbackCode: "ACP_TURN_FAILED",
      fallbackMessage: "Could not read ACP runtime capabilities.",
    });
  }
  const controls = new Set<AcpRuntimeCapabilities["controls"][number]>(reported?.controls ?? []);
  if (params.runtime.setMode) {
    controls.add("session/set_mode");
  }
  if (params.runtime.setConfigOption) {
    controls.add("session/set_config_option");
  }
  if (params.runtime.getStatus) {
    controls.add("session/status");
  }
  const normalizedKeys = new Set(
    (reported?.configOptionKeys ?? [])
      .map((entry) => normalizeText(entry))
      .filter(Boolean) as string[],
  );
  if (
    normalizedKeys.size === 0 &&
    params.includeStatusConfigOptionKeys &&
    params.runtime.getStatus
  ) {
    try {
      const status = await params.runtime.getStatus({ handle: params.handle });
      for (const key of extractRuntimeStatusConfigOptionKeys(status)) {
        normalizedKeys.add(key);
      }
    } catch (error) {
      if (isAcpOwnerRepairRequired(error)) {
        throw error;
      }
      // Status-derived option keys are an optional refinement. Keep the
      // capability result usable for runtimes that expose controls but cannot
      // answer status before a turn.
    }
  }
  return {
    controls: [...controls].toSorted(),
    ...(normalizedKeys.size > 0 ? { configOptionKeys: [...normalizedKeys] } : {}),
  };
}

/** Applies persisted runtime options to a live handle once per unique option signature. */
export async function applyManagerRuntimeControls(params: {
  sessionKey: string;
  runtime: AcpRuntime;
  handle: AcpRuntimeHandle;
  meta: SessionAcpMeta;
  getCachedRuntimeState: (sessionKey: string) => CachedRuntimeState | null;
  onOptionsChanged: (options: AcpSessionRuntimeOptions) => Promise<void>;
}): Promise<void> {
  let options = resolveRuntimeOptionsFromMeta(params.meta);
  const signature = buildRuntimeControlSignature(options);
  const cached = params.getCachedRuntimeState(params.sessionKey);
  if (cached?.appliedControlSignature === signature) {
    return;
  }

  const needsConfigOptionKeys = buildRuntimeConfigOptionPairs(options).length > 0;
  const capabilities = await resolveManagerRuntimeCapabilities({
    runtime: params.runtime,
    handle: params.handle,
    includeStatusConfigOptionKeys: needsConfigOptionKeys,
  });
  const backend = params.handle.backend || params.meta.backend;
  const runtimeMode = normalizeText(options.runtimeMode);
  const configOptions = buildRuntimeConfigOptionPairs(options, capabilities.configOptionKeys);
  const thinkingConfigKey = options.thinking
    ? resolveRuntimeConfigOptionKey("thinking", capabilities.configOptionKeys)
    : undefined;
  const advertisedKeys = new Set(
    (capabilities.configOptionKeys ?? [])
      .map((entry) => normalizeLowercaseStringOrEmpty(entry))
      .filter(Boolean),
  );

  await withAcpRuntimeErrorBoundary({
    run: async () => {
      if (runtimeMode) {
        if (!capabilities.controls.includes("session/set_mode") || !params.runtime.setMode) {
          throw createUnsupportedControlError({
            backend,
            control: "session/set_mode",
          });
        }
        await params.runtime.setMode({
          handle: params.handle,
          mode: runtimeMode,
        });
      }

      if (configOptions.length > 0) {
        if (
          !capabilities.controls.includes("session/set_config_option") ||
          !params.runtime.setConfigOption
        ) {
          throw createUnsupportedControlError({
            backend,
            control: "session/set_config_option",
          });
        }
        for (const [key, requestedValue] of configOptions) {
          // Model changes can clamp or remove unsupported thinking before its turn in the replay.
          const value = key === thinkingConfigKey ? options.thinking : requestedValue;
          if (value === undefined) {
            continue;
          }
          if (
            advertisedKeys.size > 0 &&
            !advertisedKeys.has(normalizeLowercaseStringOrEmpty(key))
          ) {
            throw new AcpRuntimeError(
              "ACP_BACKEND_UNSUPPORTED_CONTROL",
              `ACP backend "${backend}" does not accept config key "${key}".`,
            );
          }
          try {
            const result = await params.runtime.setConfigOption({
              handle: params.handle,
              key,
              value,
            });
            const accepted = reconcileAcceptedRuntimeOptions(
              options,
              result,
              normalizeLowercaseStringOrEmpty(key) === "model" ? options.thinking : undefined,
            );
            if (!runtimeOptionsEqual(options, accepted)) {
              // Persist each accepted change even if a later control fails.
              await params.onOptionsChanged(accepted);
              options = accepted;
            }
          } catch (error) {
            if (
              isUnsupportedOptionalTimeoutConfigRejection(key, error) ||
              isRejectedThinkingConfigOption(key, error)
            ) {
              continue;
            }
            throw error;
          }
        }
      }
    },
    fallbackCode: "ACP_TURN_FAILED",
    fallbackMessage: "Could not apply ACP runtime options before turn execution.",
  });

  if (cached) {
    cached.appliedControlSignature = buildRuntimeControlSignature(options);
  }
}