guanghulab/agents/zhuyuan-dev-agent/modules/executor.js

// ═══════════════════════════════════════
// 铸渊开发Agent · 执行引擎 v2.0
// 宪法约束：3次重试上限 · 统一输出接口 · 依赖链 · 紧急停止
// ═══════════════════════════════════════

const fs = require("fs");
const path = require("path");
const { execSync } = require("child_process");

const WORKSPACE = process.env.WORKSPACE_DIR || "/opt/guanghu/zhuyuan-workspace";
const CONSTITUTION = JSON.parse(fs.readFileSync(path.join(__dirname, "..", "agent-constitution.json"), "utf-8"));
const { fail_protocol, unified_interface, dependency_chain, emergency_stop } = CONSTITUTION;

if (!fs.existsSync(WORKSPACE)) fs.mkdirSync(WORKSPACE, { recursive: true });

/**
 * 统一包装输出——所有模块输出必须经过这里
 */
function uniformResult(taskId, step, ok, extra = {}) {
  return {
    ok,
    task_id: taskId,
    step,
    action: extra.action || "unknown",
    output: extra.output || null,
    error: extra.error || null,
    retries: extra.retries || 0,
    skipped: !!extra.skipped,
    blocked: !!extra.blocked
  };
}

/**
 * 带重试的执行
 */
async function executeWithRetry(substep, logFn, retriesLeft) {
  const start = Date.now();
  const attempt = fail_protocol.max_retries_per_task - retriesLeft + 1;
  logFn("exec_attempt", { step: substep.step, action: substep.action, attempt, retriesLeft });

  try {
    let result;
    switch (substep.action) {
      case "write_file":
        result = doWriteFile(substep);
        break;
      case "run_cmd":
        result = doRunCmd(substep);
        break;
      case "call_api":
        result = await doCallApi(substep);
        break;
      case "check":
        result = doCheck(substep);
        break;
      default:
        return uniformResult(substep.task_id, substep.step, false, { action: substep.action, error: "未知动作类型: " + substep.action });
    }

    if (result.ok) {
      logFn("exec_done", { step: substep.step, result: "ok", ms: Date.now() - start, attempt });
      return uniformResult(substep.task_id, substep.step, true, { action: substep.action, output: result.output || result.path, retries: attempt - 1 });
    }

    // 失败了但还有重试次数
    if (retriesLeft > 1) {
      logFn("exec_retry", { step: substep.step, error: result.error, retriesLeft: retriesLeft - 1 });
      return executeWithRetry(substep, logFn, retriesLeft - 1);
    }

    // 重试耗尽 → 跳过
    logFn("exec_skip", { step: substep.step, error: result.error, reason: "重试" + fail_protocol.max_retries_per_task + "次全部失败，跳过" });
    return uniformResult(substep.task_id, substep.step, false, { action: substep.action, error: result.error, retries: fail_protocol.max_retries_per_task, skipped: true });

  } catch (err) {
    if (retriesLeft > 1) {
      logFn("exec_retry", { step: substep.step, error: err.message, retriesLeft: retriesLeft - 1 });
      return executeWithRetry(substep, logFn, retriesLeft - 1);
    }
    logFn("exec_skip", { step: substep.step, error: err.message, reason: "重试" + fail_protocol.max_retries_per_task + "次全部失败，跳过" });
    return uniformResult(substep.task_id, substep.step, false, { action: substep.action, error: err.message, retries: fail_protocol.max_retries_per_task, skipped: true });
  }
}

function doWriteFile(s) {
  const fp = path.resolve(WORKSPACE, s.target);
  const dir = path.dirname(fp);
  if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });
  fs.writeFileSync(fp, s.content || "", "utf-8");
  return { ok: true, path: fp };
}

function doRunCmd(s) {
  const out = execSync(s.target, { cwd: s.cwd || WORKSPACE, encoding: "utf-8", timeout: 60000, maxBuffer: 10*1024*1024 });
  return { ok: true, output: out.slice(0, 5000) };
}

async function doCallApi(s) {
  const https = require("https");
  const body = JSON.stringify(s.body || {});
  const url = new URL(s.target);
  return new Promise((resolve) => {
    const req = https.request({
      hostname: url.hostname, path: url.pathname, method: s.method || "POST",
      headers: { "Content-Type": "application/json", ...(s.headers || {}) }
    }, (res) => {
      let data = "";
      res.on("data", c => data += c);
      res.on("end", () => resolve({ ok: res.statusCode < 400, status: res.statusCode, body: data.slice(0, 2000) }));
    });
    req.on("error", e => resolve({ ok: false, error: e.message }));
    if (body !== "{}") req.write(body);
    req.end();
  });
}

function doCheck(s) {
  const fp = path.resolve(WORKSPACE, s.target);
  return { ok: fs.existsSync(fp), exists: fs.existsSync(fp), path: fp };
}

/**
 * 执行完整规划——宪法约束版本
 */
async function executePlan(plan, logFn) {
  console.log("[Executor v2] 开始执行: " + plan.plan_id + " · 宪法约束模式");
  logFn("constitution_loaded", { max_retries: fail_protocol.max_retries_per_task });

  const results = [];
  let doneCount = 0, failCount = 0, skipCount = 0, consecutiveFails = 0;
  const blockedTasks = new Set();

  for (const task of plan.tasks) {
    const substeps = task.substeps || [];

    for (const ss of substeps) {
      // 检查是否被阻塞
      if (ss.depends_on && blockedTasks.has(ss.depends_on)) {
        const blocked = uniformResult(task.id, ss.step, false, { action: ss.action, blocked: true, error: "上游依赖 " + ss.depends_on + " 失败，此任务阻塞" });
        results.push(blocked);
        logFn("exec_blocked", { step: ss.step, depends_on: ss.depends_on });
        failCount++;
        continue;
      }

      // 执行（带重试）
      const r = await executeWithRetry({ ...ss, task_id: task.id }, logFn, fail_protocol.max_retries_per_task);
      results.push(r);

      if (r.ok) {
        doneCount++;
        consecutiveFails = 0;
      } else if (r.skipped) {
        skipCount++;
        consecutiveFails++;
        // 被跳过的任务 → 下游阻塞
        blockedTasks.add(task.id + ":" + ss.step);
      } else {
        failCount++;
        consecutiveFails++;
        blockedTasks.add(task.id + ":" + ss.step);
      }

      // 紧急停止检查
      if (consecutiveFails >= 5) {
        logFn("emergency_stop", { reason: "连续" + consecutiveFails + "个任务失败", summary: { done: doneCount, fail: failCount, skip: skipCount } });
        return { plan_id: plan.plan_id, emergency_stop: true, done: doneCount, failed: failCount, skipped: skipCount, results };
      }
    }
  }

  const summary = { plan_id: plan.plan_id, total: results.length, done: doneCount, failed: failCount, skipped: skipCount, results };
  console.log("[Executor v2] 完成 · 成功 " + doneCount + " / 失败 " + failCount + " / 跳过 " + skipCount);
  return summary;
}

module.exports = { executePlan };