maestro/src/bridge/delegate-runs-api.ts

import { Router, type Request, type Response } from 'express';
import { existsSync, readFileSync } from 'fs';
import { join } from 'path';
import { type Repository, localTaskRepoName } from '../db/repository.js';
import { logger } from '../logger.js';
import { logRoot } from '../spaces/runtime-paths.js';
import { parseEventLine, type EventBase } from '../progress/event-log.js';
import { reconstructDelegateRuns } from '../progress/delegate-runs.js';
import { canViewTask, resolveSpaceAccess, collectAllSubJobs, readJobEvents, isJobWithinWorkspace } from './local-api-helpers.js';

const MAX_SUBJOBS = 200;

/** 巨大 events.jsonl の安全弁: 末尾 N バイトだけ読む（先頭行が途中で切れたら捨てる）。 */
const MAX_EVENTS_BYTES = 32 * 1024 * 1024;

function readEvents(task: { workspacePath: string | null; runtimeDir?: string | null }): EventBase[] {
  if (!task.workspacePath) return [];
  const path = join(logRoot({ workspacePath: task.workspacePath, runtimeDir: task.runtimeDir }), 'events.jsonl');
  if (!existsSync(path)) return [];
  let raw: string;
  try {
    raw = readFileSync(path, 'utf-8');
  } catch {
    return [];
  }
  if (raw.length > MAX_EVENTS_BYTES) raw = raw.slice(raw.length - MAX_EVENTS_BYTES);
  const out: EventBase[] = [];
  for (const line of raw.split('\n')) {
    if (!line.trim()) continue;
    const r = parseEventLine(line);
    if (r.kind === 'ok') out.push(r.event);
  }
  return out;
}

export function createDelegateRunsRouter(repo: Repository): Router {
  const router = Router();

  router.get('/:id/delegate-runs', async (req: Request, res: Response) => {
    try {
      const taskId = Number(req.params.id);
      const viewer = req.user as Express.User | undefined;
      const task = await repo.getLocalTask(taskId, viewer ? { viewer } : undefined);
      if (!canViewTask(req, res, task, resolveSpaceAccess(repo, task, viewer))) return;

      const runs = reconstructDelegateRuns(readEvents(task!));

      const subtasks: Array<{
        jobId: string;
        issueNumber: number;
        depth: number;
        status: string;
        runs: ReturnType<typeof reconstructDelegateRuns>;
      }> = [];

      // ローカルタスクは issueNumber === taskId の擬似リポジトリパターンで格納される
      const latestJob = await repo.getLatestJobForIssue(localTaskRepoName(taskId), taskId);
      if (latestJob) {
        const nodes = await collectAllSubJobs(repo, latestJob.id);
        for (const { job, depth } of nodes.slice(0, MAX_SUBJOBS)) {
          if (!job.worktreePath) continue;
          // fail-closed: workspacePath 未設定なら containment を評価できないので読まない。
          // NOTE(将来): runtime_dir が有効化され task.workspacePath が null になるスペースモデルでは、
          // ここで全サブタスクグループが隠れる。その際は latestJob.worktreePath をアンカーにする。
          if (!isJobWithinWorkspace(task!.workspacePath, job.worktreePath)) continue;
          const subRuns = reconstructDelegateRuns(readJobEvents(job.worktreePath));
          if (subRuns.length === 0) continue;
          subtasks.push({ jobId: job.id, issueNumber: job.issueNumber, depth, status: job.status, runs: subRuns });
        }
      }

      res.json({ runs, subtasks });
    } catch (err) {
      logger.error(`[delegate-runs] list error: ${err}`);
      res.status(500).json({ error: 'Failed to fetch delegate runs' });
    }
  });

  router.get('/:id/delegate-runs/:delegateRunId/timeline', async (req: Request, res: Response) => {
    try {
      const taskId = Number(req.params.id);
      const runId = req.params.delegateRunId;
      const ownerJobId = typeof req.query.jobId === 'string' ? req.query.jobId : null;
      // reconstructDelegateRuns と同じ帰属判定。tool_call/tool_result は
      // ペアリング用に correlationId を per-tool UUID で上書きするため、実行への
      // 帰属は delegateRunId タグで見る（旧ログは correlationId フォールバック）。
      const belongsToRun = (e: EventBase): boolean => (e.delegateRunId ?? e.correlationId) === runId;
      const viewer = req.user as Express.User | undefined;
      const task = await repo.getLocalTask(taskId, viewer ? { viewer } : undefined);
      if (!canViewTask(req, res, task, resolveSpaceAccess(repo, task, viewer))) return;

      let events: EventBase[];
      if (ownerJobId) {
        // jobId 指定: そのジョブの events.jsonl を読む（fail-closed containment チェック付き）
        const job = await repo.getJob(ownerJobId, viewer ? { viewer } : undefined);
        if (!job || !job.worktreePath ||
            !isJobWithinWorkspace(task!.workspacePath, job.worktreePath)) {
          res.status(404).json({ error: 'job not found' });
          return;
        }
        events = readJobEvents(job.worktreePath);
      } else {
        // jobId 未指定: 親 events を読み、run 帰属イベントが無ければサブジョブを線形探索
        events = readEvents(task!);
        if (!events.some(belongsToRun)) {
          const latestJob = await repo.getLatestJobForIssue(localTaskRepoName(taskId), taskId);
          if (latestJob) {
            for (const { job } of (await collectAllSubJobs(repo, latestJob.id)).slice(0, MAX_SUBJOBS)) {
              if (!job.worktreePath) continue;
              if (!isJobWithinWorkspace(task!.workspacePath, job.worktreePath)) continue;
              const ev = readJobEvents(job.worktreePath);
              if (ev.some(belongsToRun)) {
                events = ev;
                break;
              }
            }
          }
        }
      }

      res.json({ events: events.filter(belongsToRun) });
    } catch (err) {
      logger.error(`[delegate-runs] timeline error: ${err}`);
      res.status(500).json({ error: 'Failed to fetch delegate run timeline' });
    }
  });

  return router;
}