宜博 / lemonboss

/**
 * Graph Run Routes — P3 图执行的 REST 端点（T6）
 *
 * 挂载: /api/graph-runs（主 router）
 *      /api/company（companyRouter，只提供 GET /:companyId/graph-runs）
 *
 * 设计见 docs/todo/system-design/27-p3-p4-work-graph-console.md
 *   §四.C  五个控制操作的可用条件与语义（本文件只做「条件判断 + 转调引擎」，不含推进逻辑）
 *   §三.2  两处不能照抄 pipeline 的坑：
 *          ① 定义层与运行时层分开 → GET /:id 吐的是 **run 实例**（run + nodes + edges + 回退记录），
 *             不是图定义，前端不会再出现 CompanyDashboard.vue 那种「拿定义画运行时高亮」的空引用；
 *          ② 「重试」必须是独立、参数自解释的端点 → POST /:id/nodes/:nodeId/retry，
 *             要重试哪个节点全写在 URL 里，不需要调用方拼隐藏参数组合
 *             （pipeline 的 retryPipelineRun 调 /advance 但不传 stepIndex，点了必定 400）。
 *
 * 错误约定（与 routes/company.js 的 org-graph 端点、routes/devloop.js 保持一致）：
 *   404 { error }                                     run / 节点不存在
 *   400 { error, op, runStatus, allowed }             操作在当前 run_status 下不可用（非法状态转换，明确说清）
 *   400 { error, scope, from, to }                    引擎抛 InvalidTransitionError
 *   400 { error }                                     入参不合法
 *   500 { error }                                     其它
 */

const express = require('express');

const engine = require('../services/graph-engine');
const {
  getRun,
  getRunState,
  getRunNode,
  getRunNodeByDefId,
  listChildRuns,
  listCompanyRuns,
  listLoopRecords,
  InvalidTransitionError,
} = require('../services/work-graph-db');

const router = express.Router();
const companyRouter = express.Router();

// ────────────────────────────────────────────────────────────────
// 小工具
// ────────────────────────────────────────────────────────────────

/** async handler 包装：抛出来的异常统一落到下面的 sendError */
const h = (fn) => (req, res) => Promise.resolve(fn(req, res)).catch((err) => sendError(res, err));

function sendError(res, err) {
  if (err instanceof InvalidTransitionError || err.name === 'InvalidTransitionError') {
    return res.status(400).json({ error: err.message, scope: err.scope, from: err.from, to: err.to });
  }
  res.status(500).json({ error: err.message });
}

/** 取 run，不存在直接回 404 并返回 null（调用方 return 掉即可） */
function loadRun(req, res) {
  const run = getRun(req.params.id);
  if (!run) {
    res.status(404).json({ error: `graph run 不存在: ${req.params.id}` });
    return null;
  }
  return run;
}

/**
 * §四.C 的「可用条件」那一列。不可用时返回明确错误——说清是哪个操作、
 * 当前是什么 run_status、什么状态下才可用，不闷掉。
 *
 * @param {boolean} [alsoOk] 状态之外的额外放行条件（「启动」用，见下）；
 *                           allowed 只用于报错文案，所以额外条件也要在里面写清楚。
 */
function ensureStatus(res, run, op, allowed, alsoOk = false) {
  if (alsoOk || allowed.includes(run.run_status)) return true;
  res.status(400).json({
    error: `「${op}」在 run_status='${run.run_status}' 下不可用，仅当 run_status ∈ {${allowed.join(', ')}} 时可用`,
    op,
    runStatus: run.run_status,
    allowed,
  });
  return false;
}

/**
 * advanceOne 的 ok:false 分两类：
 *   - 「操作本身被状态挡住」→ 400，用户需要先改状态（先启动 / 先恢复）；
 *   - 「这一步没推进但不算错」（没有 ready 节点 = 可能已收敛、被别人抢走、在等子 run）→ 200 原样返回。
 */
const BLOCKED_REASONS = {
  'run-not-found': '该 run 不存在',
  'run-not-running': 'run 不在 running 状态，先调 /start',
  'pause-requested': 'run 已请求暂停，先调 /start 恢复',
  'stop-requested': 'run 已请求停止，不能再推进',
};

function sendAdvance(res, out) {
  if (!out.ok && BLOCKED_REASONS[out.reason]) {
    return res.status(400).json({
      error: BLOCKED_REASONS[out.reason],
      op: 'next',
      reason: out.reason,
      runStatus: out.runStatus,
    });
  }
  res.json(out);
}

// ────────────────────────────────────────────────────────────────
// 控制类端点（§四.C 五个操作）
// ────────────────────────────────────────────────────────────────

/**
 * POST /api/graph-runs/:id/start — 「启动」
 * body: { mode?: 'manual' | 'auto' }（不传沿用 run 上已存的 execution_mode）
 * 可用条件：run_status ∈ {ready, paused, blocked}，**或 running 且 pause_requested=1**
 *
 * 后半条不是可选的宽松（R6-F3 问题 1）：「暂停」是软信号，running → paused 要等节点边界
 * （landPause 手头有 running 节点就不落）。而 org 层每个节点都是 subgraph，子 run 在飞时
 * 父节点恒为 running —— root run 于是**永久停在 `running` + `pause_requested=1`**，
 * 只看 run_status 的话「启动」永远 400，用户只能一步步点「下一步」，验收 4 的后半不成立。
 * 引擎侧本来就对：startRun → startRunSync 会清全树软信号并把级联 paused 的后代拉回 running，
 * 这里放行的正是「撤回暂停意图」。已经在 stop_requested 的 run 不在此列（它正在收摊）。
 *
 * auto 模式**不等**引擎跑完就返回：一次自动跑到底是分钟~小时级的，
 * 绝不能像 DevLoop 那样把整个循环塞进一次 HTTP 请求-响应周期（§三.1）。
 */
router.post('/:id/start', h((req, res) => {
  const run = loadRun(req, res);
  if (!run) return;
  const pausing = run.run_status === 'running' && Boolean(run.pause_requested) && !run.stop_requested;
  if (!ensureStatus(res, run, '启动', ['ready', 'paused', 'blocked', 'running(pause_requested=1)'], pausing)) return;

  const { mode } = req.body || {};
  if (mode !== undefined && mode !== 'manual' && mode !== 'auto') {
    return res.status(400).json({ error: `mode 只能是 'manual' 或 'auto'，收到: ${mode}` });
  }

  // driver 是 auto 循环的 promise，不可序列化也不该 await —— 丢掉，让它自己在后台跑
  const { driver, ...out } = engine.startRun(run.id, mode ? { mode } : {});
  res.json({ ...out, executionMode: mode || run.execution_mode, auto: Boolean(driver) });
}));

/**
 * POST /api/graph-runs/:id/next — 「下一步」
 * body: { task?: string }（可选，透传给节点 executor 作任务描述）
 * 可用条件：run_status ∈ {running, paused}
 *
 * 只推进**一个** ready 节点，完成后把满足条件的下游置 ready 就返回，不继续往下跑。
 * 会真正发起一次 agent 调用，所以这个请求本身可能是秒~分钟级的（§四.C 表格最后一列）。
 */
router.post('/:id/next', h(async (req, res) => {
  const run = loadRun(req, res);
  if (!run) return;
  if (!ensureStatus(res, run, '下一步', ['running', 'paused'])) return;

  const out = await engine.nextStep(run.id, req.body && req.body.task ? { task: req.body.task } : {});
  sendAdvance(res, out);
}));

/**
 * POST /api/graph-runs/:id/pause — 「暂停」
 * 可用条件：run_status = 'running'
 * 软信号：不打断正在 running 的节点，在下一个节点边界生效。
 */
router.post('/:id/pause', h((req, res) => {
  const run = loadRun(req, res);
  if (!run) return;
  if (!ensureStatus(res, run, '暂停', ['running'])) return;
  res.json({ ...engine.pauseRun(run.id), runStatus: (getRun(run.id) || {}).run_status });
}));

/**
 * POST /api/graph-runs/:id/stop — 「停止」
 * 可用条件：run_status ∈ {running, paused, blocked}
 * 仍 pending/ready 的节点批量 skipped，run → stopped。正在 running 的节点能否硬取消见 T9。
 */
router.post('/:id/stop', h((req, res) => {
  const run = loadRun(req, res);
  if (!run) return;
  if (!ensureStatus(res, run, '停止', ['running', 'paused', 'blocked'])) return;
  res.json(engine.stopRun(run.id));
}));

/**
 * POST /api/graph-runs/:id/nodes/:nodeId/retry — 「重试」
 * 可用条件：节点 status ∈ {fail, blocked}
 *
 * §三.2 坑 2 的正解：重试的对象写在 URL 里，端点参数自解释，body 可以是空的。
 * nodeId 优先按 graph_run_nodes.id 解析；解析不到再按 def_node_id 在本 run 内找一次
 * ——画布上拿到的可能是定义层节点 id，两种都能直接重试，不用调用方自己换算。
 */
router.post('/:id/nodes/:nodeId/retry', h((req, res) => {
  const run = loadRun(req, res);
  if (!run) return;

  const { nodeId } = req.params;
  let nodeRun = getRunNode(nodeId);
  if (!nodeRun || nodeRun.run_id !== run.id) nodeRun = getRunNodeByDefId(run.id, nodeId);
  if (!nodeRun) {
    return res.status(404).json({ error: `节点 ${nodeId} 不属于 run ${run.id}` });
  }

  if (!['fail', 'blocked'].includes(nodeRun.status)) {
    return res.status(400).json({
      error: `「重试」在节点 status='${nodeRun.status}' 下不可用，仅当节点 status ∈ {fail, blocked} 时可用`,
      op: 'retry',
      nodeStatus: nodeRun.status,
      allowed: ['fail', 'blocked'],
    });
  }

  // ok:false / taken-by-other 是并发下的正常结果（§四.A.6 幂等：别人已经拿走就返回当前状态，不报错）
  res.json(engine.retryNode(run.id, nodeRun.id));
}));

// ────────────────────────────────────────────────────────────────
// 只读查询类端点
// ────────────────────────────────────────────────────────────────

/**
 * GET /api/graph-runs/:id — 一次 run 的全量状态
 * 返回 { run, nodes, edges, loopRecords, childRuns, budget }
 *   nodes  = 定义层（name / systemAgentId / nodeKind）与运行时（status / attempt / childRunId …）合并后的节点
 *   edges  = 定义层原样（含 condition / failReason / edgeKind），画布靠 edgeKind='backward' 画虚线
 *   loopRecords = 回退账本，画布上 `N/3` 徽标的数据来源（§四.B.2）
 *   budget = root run 的 budget_json 解析结果，页头预算条用（§四.B.4）；非 root run 也回这个值
 */
router.get('/:id', h((req, res) => {
  const state = getRunState(req.params.id);
  if (!state) return res.status(404).json({ error: `graph run 不存在: ${req.params.id}` });

  const rootRun = state.run.root_run_id === state.run.id ? state.run : getRun(state.run.root_run_id);
  let budget = null;
  try {
    budget = rootRun && rootRun.budget_json ? JSON.parse(rootRun.budget_json) : null;
  } catch {
    budget = null; // 预算条坏掉不该让整个 run 查不出来
  }

  res.json({
    ...state,
    loopRecords: listLoopRecords(state.run.id),
    childRuns: listChildRuns(state.run.id),
    budget,
  });
}));

/**
 * GET /api/graph-runs/:id/stream — SSE，实时推该 run 的状态转换事件
 * 客户端可带 Last-Event-ID header（或 ?lastEventId=）断线重放，语义与 devloop 的 /stream 一致。
 *
 * graph-run-sse.js 由 T5 并行落地；还没到位时回 503 而不是让整个 server 起不来。
 */
router.get('/:id/stream', (req, res) => {
  const run = getRun(req.params.id);
  if (!run) return res.status(404).json({ error: `graph run 不存在: ${req.params.id}` });

  let sse;
  try {
    sse = require('../services/graph-run-sse');
  } catch (err) {
    return res.status(503).json({ error: `SSE 通道未就绪（graph-run-sse 未加载）: ${err.message}` });
  }

  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.flushHeaders?.();

  const lastEventId = req.headers['last-event-id'] || req.query.lastEventId || null;
  sse.subscribe(run.id, res, { replayFromOffset: lastEventId != null ? lastEventId : undefined });
});

/**
 * GET /api/company/:companyId/graph-runs — 某公司的 run 列表
 * query: rootOnly=1 只列 root run（v1 一家公司同时只有一个 active root run，见 §一非目标）
 * 公司无 run 时返回空数组，不报错（与 GET /api/company/:cid/org-graph 的处理一致）。
 */
companyRouter.get('/:companyId/graph-runs', h((req, res) => {
  const rootOnly = req.query.rootOnly === '1' || req.query.rootOnly === 'true';
  res.json({
    companyId: req.params.companyId,
    runs: listCompanyRuns(req.params.companyId, { rootOnly }),
  });
}));

module.exports = router;
module.exports.companyRouter = companyRouter;