OpenClaw源码剖析 #04 · Agent Runtime：任务编排与执行

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

// agent-command.ts
export async function agentCommand(opts: AgentCommandOpts, runtime: RuntimeEnv, deps?: CliDeps) {
  return await agentCommandInternal({
    ...opts,
    senderIsOwner: opts.senderIsOwner ?? true,
    allowModelOverride: opts.allowModelOverride ?? true,
  }, runtime, deps);
}

async function agentCommandInternal(opts, runtime = defaultRuntime, deps?) {
  // Step 1: 准备执行
  const prepared = await prepareAgentCommandExecution(opts, runtime);

  // Step 2: 判断运行模式
  // ACP 模式：直接调用 ACP Runtime
  if (acpResolution?.kind === "ready") {
    return await acpManager.runTurn({ ... });
  }

  // 标准模式：通过 Harness 执行
  // Step 3: 解析 Skills Snapshot
  const skillsSnapshot = needsSkillsSnapshot
    ? await buildWorkspaceSkillSnapshot(workspaceDir, { ... })
    : currentSkillsSnapshot;

  // Step 4: 解析模型
  const { provider, model } = resolveModelRef(...);

  // Step 5: 执行 Attempt（带 Fallback）
  const result = await runWithModelFallback({
    run: async (providerOverride, modelOverride) => {
      return attemptExecutionRuntime.runAgentAttempt({ ... });
    }
  });

  // Step 6: 投递结果
  return deliverAgentCommandResult({ result, ... });
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

export type AgentCommandOpts = {
  message: string;
  to?: string;
  sessionId?: string;
  sessionKey?: string;
  agentId?: string;
  model?: string;
  provider?: string;
  thinking?: string;
  thinkingOnce?: string;
  verbose?: string;
  timeout?: string;
  lane?: string;
  deliver?: boolean;
  threadId?: string | number;
  spawnedBy?: string;
  groupId?: string;
  groupChannel?: string;
  groupSpace?: string;
  workspaceDir?: string;
  images?: string[];
  imageOrder?: string[];
  clientTools?: Record<string, unknown>;
  extraSystemPrompt?: string;
  bootstrapContextMode?: string;
  bootstrapContextRunKind?: string;
  internalEvents?: unknown[];
  inputProvenance?: unknown;
  streamParams?: unknown;
  modelRun?: boolean;
  promptMode?: "none" | "default";
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

export type AgentHarness = {
  id: string;
  label: string;
  pluginId?: string;

  // 支持性检查
  supports(ctx: AgentHarnessSupportContext): AgentHarnessSupport;

  // 执行 Attempt
  runAttempt(params: AgentHarnessAttemptParams): Promise<AgentHarnessAttemptResult>;

  // 结果分类（可选）
  classify?(result, ctx): AgentHarnessResultClassification | undefined;

  // 压缩会话（可选）
  compact?(params): Promise<AgentHarnessCompactResult | undefined>;

  // 重置会话（可选）
  reset?(params): Promise<void> | void;

  // 释放资源（可选）
  dispose?(): Promise<void> | void;
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

export function resolveAgentHarnessPolicy(params: {
  provider: string;
  modelId: string;
  config: OpenClawConfig;
  agentId: string;
  sessionKey: string;
}): { runtime: string; fallback: string } {
  // 1. 检查会话是否固定了 Harness
  // 2. 检查 Agent 配置
  // 3. 检查 Provider 策略
  // 4. 回退到默认策略
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

export function runAgentAttempt(params: {
  providerOverride: string;
  modelOverride: string;
  originalProvider: string;
  cfg: OpenClawConfig;
  sessionEntry: SessionEntry | undefined;
  sessionId: string;
  sessionKey: string | undefined;
  sessionAgentId: string;
  sessionFile: string;
  workspaceDir: string;
  body: string;
  isFallbackRetry: boolean;
  resolvedThinkLevel: ThinkLevel;
  timeoutMs: number;
  runId: string;
  opts: AgentCommandOpts;
  runContext: ReturnType<typeof resolveAgentRunContext>;
  spawnedBy: string | undefined;
  messageChannel: ReturnType<typeof resolveMessageChannel>;
  skillsSnapshot: ReturnType<typeof buildWorkspaceSkillSnapshot> | undefined;
  resolvedVerboseLevel: VerboseLevel | undefined;
  agentDir: string;
  authProfileProvider: string;
  sessionStore?: Record<string, SessionEntry>;
  storePath?: string;
}) {
  // 1. 解析 Prompt（含重试上下文）
  const effectivePrompt = annotateInterSessionPromptText(resolvedPrompt, ...);

  // 2. 选择 Harness
  const agentHarnessPolicy = resolveAgentHarnessPolicy({ ... });
  const sessionPinnedAgentHarnessId = resolveSessionPinnedAgentHarnessId({ ... });

  // 3. 构建认证计划
  const harnessAuthSelection = resolveHarnessAuthProfileSelection({ ... });
  const runtimeAuthPlan = buildAgentRuntimeAuthPlan({ ... });

  // 4. 分发到对应执行器
  if (isCliProvider(cliExecutionProvider)) {
    return runCliAgent({ ... });
  }
  return runEmbeddedPiAgent({ ... });
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

// attempt-execution.ts 中的执行循环
for (;;) {
  try {
    const fallbackResult = await runWithModelFallback({
      provider,
      model,
      runId,
      classifyResult: ({ provider, model, result }) =>
        classifyEmbeddedPiRunResultForModelFallback({ provider, model, result }),
      run: async (providerOverride, modelOverride) => {
        return attemptExecutionRuntime.runAgentAttempt({
          providerOverride,
          modelOverride,
          ...
        });
      },
    });
    result = fallbackResult.result;
    break;
  } catch (err) {
    if (err instanceof LiveSessionModelSwitchError) {
      // Live 模型切换重试
      liveSwitchRetries++;
      if (liveSwitchRetries > MAX_LIVE_SWITCH_RETRIES) throw err;
      provider = err.provider;
      model = err.model;
      continue;
    }
    throw err;
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

export type SpawnAcpParams = {
  task: string;              // 任务描述
  label?: string;            // 会话标签
  agentId?: string;          // 目标 Agent ID
  resumeSessionId?: string;   // 恢复会话 ID
  model?: string;            // 模型覆盖
  thinking?: string;         // 思考层级
  runTimeoutSeconds?: number; // 超时（秒）
  cwd?: string;              // 工作目录
  mode?: "run" | "session"; // 运行模式
  thread?: boolean;           // 是否绑定线程
  sandbox?: "inherit" | "require"; // 沙箱模式
  streamTo?: "parent";       // 输出流目标
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

type SpawnAcpAcceptedResult = {
  status: "accepted";
  childSessionKey: string;
  runId: string;
  mode: "run" | "session";
  streamLogPath?: string;
  note?: string;
};

type SpawnAcpFailedResult = {
  status: "forbidden" | "error";
  error: string;
  errorCode: SpawnAcpErrorCode;
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

export const ACP_SPAWN_ERROR_CODES = [
  "acp_disabled",           // ACP 被策略禁用
  "requester_session_required", // 需要请求者会话
  "runtime_policy",           // 运行时策略冲突
  "resume_forbidden",        // 恢复会话被禁止
  "subagent_policy",         // 子 Agent 策略限制
  "thread_required",         // 需要线程上下文
  "target_agent_required",   // 需要目标 Agent
  "runtime_agent_mismatch",   // Agent 类型不匹配
  "agent_forbidden",          // Agent 被策略禁止
  "cwd_resolution_failed",    // 工作目录解析失败
  "thread_binding_invalid",   // 线程绑定无效
  "spawn_failed",             // 孵化失败
  "dispatch_failed",          // 分发失败
] as const;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

export function resolveSession(params: {
  cfg: OpenClawConfig;
  to?: string;
  sessionId?: string;
  sessionKey?: string;
  agentId?: string;
}) {
  // 1. 按优先级尝试解析：
  //    sessionKey > sessionId > (to + agentId)
  // 2. 查找或创建会话
  // 3. 返回会话元数据
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

export async function updateSessionStoreAfterAgentRun(params: {
  cfg: OpenClawConfig;
  sessionId, sessionKey, storePath, sessionStore,
  defaultProvider, defaultModel,
  fallbackProvider, fallbackModel,
  result,
  touchInteraction: boolean,
}) {
  // 更新会话的：
  // - 模型使用计数
  // - Token 使用量
  // - 最后交互时间
  // - 运行元数据
}

1
2
3
4
5
6

runWithModelFallback({
  provider, model,
  fallbacks: [...],
  run: (p, m) => attempt(...),
  classifyResult: (...) => { ... }
})