Agent-System/web/src/api/taskRuns.ts

import { parseSSEChunk, type StreamEvent } from './chatStream';
import { mergeAuthHeaders, getBearerAuthHeaders } from './authHeaders';

export type CreateTaskRunBody = Record<string, unknown> & {
  session_id: string;
  project_id: number;
  user_id?: number;
  client_turn_id?: string;
  user_message?: string;
};

export type CreateTaskRunResponse = {
  run_id: string;
  status: string;
  workspace_id?: string;
};

export async function createTaskRun(body: CreateTaskRunBody): Promise<CreateTaskRunResponse> {
  const res = await fetch('/api/v1/tasks/runs', {
    method: 'POST',
    headers: getBearerAuthHeaders('application/json'),
    body: JSON.stringify(body),
  });
  if (!res.ok) {
    let detail = '';
    let errorCode = '';
    try {
      const payload = (await res.json()) as {
        message?: string;
        error?: string;
        error_code?: string;
        detail?: string | { reason?: string };
      };
      const rawCode = payload.error_code || payload.error || '';
      errorCode = rawCode === 'document_busy' ? 'DOC_BUSY' : rawCode;
      const structuredDetail = typeof payload.detail === 'object' ? payload.detail?.reason : payload.detail;
      detail = payload.message || structuredDetail || errorCode || '';
    } catch {
      // ignore non-json body
    }
    throw Object.assign(
      new Error(`create task run failed: HTTP ${res.status}${detail ? `: ${detail}` : ''}`),
      { httpStatus: res.status, errorCode, detail },
    );
  }
  return res.json();
}

function isTerminalTaskEvent(ev: StreamEvent): boolean {
  return ev.type === 'done' || ev.type === 'error';
}

export function sleepMs(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

/** 单次 SSE 读取；意外 EOF 时返回 { terminal:false, lastSeq } 供上层重连。 */
export async function readTaskRunSSEOnce(
  runId: string,
  afterSeq: number,
  onEvent: (ev: StreamEvent) => void,
  signal?: AbortSignal,
): Promise<{ terminal: boolean; lastSeq: number }> {
  const url = `/api/v1/tasks/runs/${encodeURIComponent(runId)}/events?after_seq=${afterSeq}`;
  const resp = await fetch(url, {
    method: 'GET',
    headers: mergeAuthHeaders({ Accept: 'text/event-stream' }),
    signal,
  });
  if (!resp.ok || !resp.body) {
    throw new Error(`subscribe task run failed: HTTP ${resp.status}`);
  }
  const reader = resp.body.getReader();
  const decoder = new TextDecoder();
  let buf = '';
  let lastSeq = afterSeq;
  let terminal = false;
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buf += decoder.decode(value, { stream: true });
    const parsed = parseSSEChunk(buf);
    buf = parsed.rest;
    if (parsed.lastSeq !== undefined) {
      lastSeq = Math.max(lastSeq, parsed.lastSeq);
    }
    for (const ev of parsed.events) {
      onEvent(ev);
      if (isTerminalTaskEvent(ev)) terminal = true;
    }
  }
  if (buf) {
    const flushed = parseSSEChunk(buf, { flush: true });
    if (flushed.lastSeq !== undefined) {
      lastSeq = Math.max(lastSeq, flushed.lastSeq);
    }
    for (const ev of flushed.events) {
      onEvent(ev);
      if (isTerminalTaskEvent(ev)) terminal = true;
    }
  }
  return { terminal, lastSeq };
}

/**
 * 订阅任务 SSE：意外 EOF / 不完整 chunked 时按最后 seq 重订阅同一 run，禁止调用方据此 POST 新 run。
 */
export async function subscribeTaskRunEvents(
  runId: string,
  handlers: {
    onEvent: (ev: StreamEvent) => void;
    afterSeq?: number;
    maxReconnects?: number;
    reconnectDelayMs?: number;
    sleep?: (ms: number) => Promise<void>;
  },
  signal?: AbortSignal,
): Promise<void> {
  let afterSeq = handlers.afterSeq ?? 0;
  const maxReconnects = handlers.maxReconnects ?? 40;
  const baseDelay = handlers.reconnectDelayMs ?? 500;
  const sleep = handlers.sleep ?? sleepMs;

  for (let attempt = 0; attempt <= maxReconnects; attempt++) {
    if (signal?.aborted) return;
    try {
      const { terminal, lastSeq } = await readTaskRunSSEOnce(runId, afterSeq, handlers.onEvent, signal);
      afterSeq = Math.max(afterSeq, lastSeq);
      if (terminal) return;
      // 意外 EOF：未见到 done/error，按 after_seq 重连
    } catch (err) {
      if (signal?.aborted) return;
      if (attempt >= maxReconnects) throw err;
    }
    if (attempt >= maxReconnects) {
      throw new Error(`task run SSE disconnected after ${maxReconnects} reconnects (after_seq=${afterSeq})`);
    }
    await sleep(Math.min(baseDelay * (attempt + 1), 5000));
  }
}

export async function cancelTaskRun(runId: string): Promise<void> {
  const res = await fetch(`/api/v1/tasks/runs/${encodeURIComponent(runId)}/cancel`, {
    method: 'POST',
    headers: getBearerAuthHeaders(),
  });
  if (!res.ok) {
    throw new Error(`cancel task run failed: HTTP ${res.status}`);
  }
}

export async function getTaskRun(runId: string) {
  const res = await fetch(`/api/v1/tasks/runs/${encodeURIComponent(runId)}`, {
    headers: getBearerAuthHeaders(),
  });
  if (!res.ok) {
    throw new Error(`get task run failed: HTTP ${res.status}`);
  }
  return res.json();
}