OpenClaw 如何保证长程任务的稳定性：从架构到源码实现 #

// src/tasks/task-executor.ts
function createQueuedTaskRun(params) { /* → status: "queued" */ }
function createRunningTaskRun(params) { /* → status: "running" */ }
function startTaskRunByRunId(params)  { /* markTaskRunningByRunId */ }
function recordTaskRunProgressByRunId(params) { /* 进度更新 */ }
function completeTaskRunByRunId(params) { /* → status: "succeeded" */ }
function failTaskRunByRunId(params)     { /* → status: "failed" */ }

// src/tasks/detached-task-runtime.ts
const DETACHED_TASK_RECOVERY_WARN_MS = 5_000;

const DEFAULT_DETACHED_TASK_LIFECYCLE_RUNTIME = {
  createQueuedTaskRun,
  createRunningTaskRun,
  startTaskRunByRunId,
  recordTaskRunProgressByRunId,
  completeTaskRunByRunId,
  failTaskRunByRunId,
  setDetachedTaskDeliveryStatusByRunId,
  cancelDetachedTaskRunById
};

async function tryRecoverTaskBeforeMarkLost(params) {
  // 5 秒超时保护：恢复钩子不能阻塞太久
  const startedAt = Date.now();
  const result = await hook(params);
  const elapsedMs = Date.now() - startedAt;
  if (elapsedMs >= 5_000) log.warn("Detached task recovery hook was slow");
  ...
}

// src/tasks/task-registry.paths.ts
function resolveTaskRegistrySqlitePath() {
  return path.join(resolveTaskRegistryDir(), "runs.sqlite");
}

// src/tasks/task-executor-policy.ts
function isTerminalTaskStatus(status) {
  return status === "succeeded" || status === "failed" ||
         status === "timed_out" || status === "cancelled" ||
         status === "lost";
}

// src/infra/heartbeat-wake.ts
const DEFAULT_COALESCE_MS = 250;   // 合并窗口 250ms
const DEFAULT_RETRY_MS    = 1_000;  // 失败重试 1s

// 优先级体系：retry > interval > action(wake/exec-event/hook) > default
const REASON_PRIORITY = {
  RETRY:   0,
  INTERVAL: 1,
  DEFAULT:  2,
  ACTION:   3
};

function requestHeartbeatNow(opts) {
  queuePendingWakeReason(opts);
  schedule(opts?.coalesceMs ?? 250, "normal");
}

// src/agents/pi-embedded-runner/ （bundle 中可见的关键逻辑）
// 空闲超时：模型长时间不输出 chunk
if (idleTimedOut) {
  const timeoutText = "The model did not produce a response before the LLM idle timeout.";
  // 触发 abort
}

// 总超时：整个 agent run 超时
if (timedOut && !timedOutDuringCompaction) {
  // 在 compaction 期间的超时不计入失败，支持自动重试
}

// src/process/command-queue.ts

// 全局单例状态（通过 Symbol.for 跨 chunk 共享）
const COMMAND_QUEUE_STATE_KEY = Symbol.for("openclaw.commandQueueState");

function getLaneState(lane) {
  return {
    lane,
    queue: [],             // 等待队列
    activeTaskIds: new Set(),  // 活跃任务 ID 集合
    maxConcurrent: 1,      // 默认串行
    draining: false,       // 排空中
    generation: 0          // 代际号，防旧任务干扰
  };
}

function enqueueCommandInLane(lane, task, opts) {
  if (getQueueState().gatewayDraining)
    return Promise.reject(new GatewayDrainingError());

  return new Promise((resolve, reject) => {
    state.queue.push({
      task: () => task(),
      resolve, reject,
      enqueuedAt: Date.now(),
      warnAfterMs: opts?.warnAfterMs ?? 2_000,
      onWait: opts?.onWait  // 排队过久时触发 warn 日志
    });
    drainLane(cleaned);
  });
}

function drainLane(lane) {
  const pump = () => {
    while (state.activeTaskIds.size < state.maxConcurrent && state.queue.length > 0) {
      const entry = state.queue.shift();
      const taskId = getQueueState().nextTaskId++;
      state.activeTaskIds.add(taskId);

      (async () => {
        try {
          const result = await entry.task();
          if (completeTask(state, taskId, taskGeneration)) {
            notifyActiveTaskWaiters();
            pump();  // 递归排空下一个
          }
          entry.resolve(result);
        } catch (err) {
          // 非预期的 lane 失败才会记录 error
          if (!isExpectedNonErrorLaneFailure(err))
            diagnosticLogger.error(`lane task error: lane=${lane}`);
          entry.reject(err);
        }
      })();
    }
  };
  pump();
}

// src/agents/session-write-lock.ts
const DEFAULT_STALE_MS          = 1_800_000;  // 30 分钟
const DEFAULT_MAX_HOLD_MS       =   300_000;  // 5 分钟
const DEFAULT_WATCHDOG_INTERVAL =    60_000;  // 1 分钟
const DEFAULT_TIMEOUT_GRACE_MS  =   120_000;  // 2 分钟

async function acquireSessionWriteLock(params) {
  const lockPath = `${normalizedSessionFile}.lock`;

  // 1. 可重入检查（默认允许）
  const held = HELD_LOCKS.get(normalizedSessionFile);
  if (allowReentrant && held) {
    held.count += 1;
    return { release: () => releaseHeldLock(...) };
  }

  // 2. 重试循环（直到 timeout）
  const startedAt = Date.now();
  while (Date.now() - startedAt < timeoutMs) {
    try {
      // wx 标志：文件存在则失败，实现互斥
      const handle = await fs.open(lockPath, "wx");
      await handle.writeFile(JSON.stringify({
        pid: process.pid,
        createdAt: new Date().toISOString(),
        starttime: getProcessStartTime(process.pid)
      }), "utf8");
      HELD_LOCKS.set(normalizedSessionFile, { count: 1, handle, ... });
      return { release: () => releaseHeldLock(...) };
    } catch (err) {
      if (err.code !== "EEXIST") throw err;
      // 3. 检查是否为 stale lock（进程已死 or 超时）
      if (await shouldReclaimContendedLockFile(lockPath, inspected)) {
        await fs.rm(lockPath, { force: true });
        continue;
      }
      // 4. 指数退避
      const delay = Math.min(1_000, 50 * attempt, remainingMs);
      await new Promise(r => setTimeout(r, delay));
    }
  }
  throw new Error(`session file locked (timeout ${timeoutMs}ms)`);
}

function ensureWatchdogStarted(intervalMs) {
  watchdogState.timer = setInterval(() => {
    runLockWatchdogCheck();  // 释放超时锁
  }, intervalMs);  // 默认 60s
}

async function runLockWatchdogCheck() {
  for (const [sessionFile, held] of HELD_LOCKS) {
    const heldForMs = Date.now() - held.acquiredAt;
    if (heldForMs > held.maxHoldMs) {
      // 强制释放超时锁
      await releaseHeldLock(sessionFile, held, { force: true });
    }
  }
}

function registerCleanupHandlers() {
  process.on("exit", releaseAllLocksSync);
  for (const signal of ["SIGINT", "SIGTERM", "SIGQUIT", "SIGABRT"])
    process.on(signal, () => handleTerminationSignal(signal));
}

function inspectLockPayload(payload, staleMs, nowMs) {
  const pidAlive = pid !== null && isPidAlive(pid);
  const pidRecycled = pidAlive && storedStarttime !== getProcessStartTime(pid);

  const staleReasons = [];
  if (!pidAlive) staleReasons.push("dead-pid");
  if (pidRecycled) staleReasons.push("recycled-pid");  // 关键！
  if (ageMs > staleMs) staleReasons.push("too-old");

  return { stale: staleReasons.length > 0, staleReasons };
}

// src/plugin-sdk/keyed-async-queue.ts
function enqueueKeyedTask(params) {
  const current = (params.tails.get(params.key) ?? Promise.resolve())
    .catch(() => void 0)
    .then(params.task)
    .finally(() => params.hooks?.onSettle?.());

  const tail = current.then(() => void 0, () => void 0);
  params.tails.set(params.key, tail);

  // 自动清理
  const cleanup = () => {
    if (params.tails.get(params.key) === tail) params.tails.delete(params.key);
  };
  tail.then(cleanup, cleanup);

  return current;
}

// src/agents/pi-embedded-runner/run/preemptive-compaction.ts
// 当 active JSONL 文件达到 maxActiveTranscriptBytes 阈值时，
// 在实际 LLM 调用前先触发 compaction

// src/agents/pi-embedded-runner/run/compaction-retry-aggregate-timeout.ts
// compaction 失败时自动 retry，并聚合超时计算
// 防止多次 retry 导致总耗时超过 agent timeout

// src/agents/pi-embedded-runner/session-truncation.ts
// 截断规则：
// 1. 保留 session header
// 2. 保留非 message 条目（custom, model_change, thinking_level_change, compaction 等）
// 3. 保留未压缩分支的条目
// 4. 保留从 firstKeptEntryId 开始的"未压缩尾部"
// 5. 被移除条目的子条目重新绑定到最近的保留祖先

// src/agents/pi-embedded-runner/compact-reasons.ts
function classifyCompactionReason(reason) {
  if (text.includes("nothing to compact")) return "no_compactable_entries";
  if (text.includes("below threshold"))     return "below_threshold";
  if (text.includes("timed out"))           return "timeout";
  if (text.includes("400") || text.includes("401")) return "provider_error_4xx";
  if (text.includes("500") || text.includes("502")) return "provider_error_5xx";
  ...
}

// before_compaction / after_compaction 钩子
runBeforeCompactionHooks(...);   // 压缩前
runAfterCompactionHooks(...);    // 压缩后
runPostCompactionSideEffects(...); // 后置副作用

// 1. 获取 session write lock
const lock = await acquireSessionWriteLock({
  sessionFile,
  timeoutMs,
  maxHoldMs: resolveSessionLockMaxHoldFromTimeout({ timeoutMs })
});

// 2. 打开 SessionManager
const sessionManager = SessionManager.open(sessionFile);

// 3. run 结束后释放锁
await lock.release();

// 使用文件锁防止多个 Gateway 同时运行

// SIGUSR1 热重启时的关键步骤：
// 1. markGatewayDraining() → 新任务直接拒绝
// 2. waitForActiveTasks(timeout) → 等待活跃任务完成
// 3. resetAllLanes() → bump generation，清 stale task IDs
// 4. 新进程接管

// 清理路径：
// 1. 关闭 tracked browser tabs
// 2. 处置 MCP stdio 子进程 (disposeSessionMcpRuntime)
// 3. 清理 cron session 所有权