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

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

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;
  assistant_msg_id?: number;
};

export async function createTaskRun(body: CreateTaskRunBody): Promise<CreateTaskRunResponse> {
  const res = await apiFetch('/api/v1/tasks/runs', {
    method: 'POST',
    headers: getBearerAuthHeaders('application/json'),
    body: JSON.stringify(body),
  });
  if (!res.ok) {
    const text = await res.text();
    throw projectSettingsErrorFromResponse(res.status, text, `create task run failed: HTTP ${res.status}`);
  }
  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));
}

export class TaskRunHttpError extends Error {
  status: number;
  constructor(status: number) {
    super(`subscribe task run failed: HTTP ${status}`);
    this.name = 'TaskRunHttpError';
    this.status = status;
  }
}

function isNonRetryableStatus(status: number): boolean {
  return status === 401 || status === 404;
}

/** 单次 SSE 读取；意外 EOF 时返回 { terminal:false, lastSeq } 供上层重连。 */
export async function readTaskRunSSEOnce(
  runId: string,
  afterSeq: number,
  onEvent: (ev: StreamEvent) => void,
  signal?: AbortSignal,
  opts?: { replayOnly?: boolean },
): Promise<{ terminal: boolean; lastSeq: number }> {
  const replay = opts?.replayOnly ? '&replay_only=1' : '';
  const url = `/api/v1/tasks/runs/${encodeURIComponent(runId)}/events?after_seq=${afterSeq}${replay}`;
  const resp = await apiFetch(
    url,
    {
      method: 'GET',
      headers: mergeAuthHeaders({ Accept: 'text/event-stream' }),
      signal,
    },
    { retry: false },
  );
  if (!resp.ok || !resp.body) {
    throw new TaskRunHttpError(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。
 * 重连前可经 onGap 轮询 run 终态：若已 completed 等则停止订阅（避免 done 丢失后永久 pending）。
 */
export async function subscribeTaskRunEvents(
  runId: string,
  handlers: {
    onEvent: (ev: StreamEvent) => void;
    afterSeq?: number;
    maxReconnects?: number;
    reconnectDelayMs?: number;
    sleep?: (ms: number) => Promise<void>;
    /** Return true to stop reconnecting (caller already settled from GET /runs). */
    onGap?: (ctx: { afterSeq: number; attempt: number }) => Promise<boolean>;
  },
  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 — 先问 onGap，再按 after_seq 重连
      if (handlers.onGap) {
        const stop = await handlers.onGap({ afterSeq, attempt });
        if (stop) return;
      }
    } catch (err) {
      if (signal?.aborted) return;
      if (err instanceof TaskRunHttpError && isNonRetryableStatus(err.status)) {
        throw err;
      }
      if (handlers.onGap) {
        const stop = await handlers.onGap({ afterSeq, attempt }).catch(() => false);
        if (stop) 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 resumeTaskRun(runId: string): Promise<CreateTaskRunResponse> {
  const res = await apiFetch(`/api/v1/tasks/runs/${encodeURIComponent(runId)}/resume`, {
    method: 'POST',
    headers: getBearerAuthHeaders('application/json'),
    body: JSON.stringify({ confirm: true }),
  });
  if (!res.ok) {
    const text = await res.text();
    throw projectSettingsErrorFromResponse(res.status, text, `resume task run failed: HTTP ${res.status}`);
  }
  return res.json();
}

export async function cancelTaskRun(runId: string): Promise<void> {
  const res = await apiFetch(`/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 type ActiveTaskRun = {
  run_id: string;
  status: string;
  task_key?: string;
  workspace_id?: string;
  assistant_message_id?: number;
  error_message?: string;
  session_id?: string;
};

export async function listActiveTaskRuns(sessionId: string): Promise<ActiveTaskRun[]> {
  const url = `/api/v1/tasks/runs?session_id=${encodeURIComponent(sessionId)}&active=1`;
  const res = await apiFetch(url, { headers: getBearerAuthHeaders() });
  if (!res.ok) {
    throw new Error(`list active task runs failed: HTTP ${res.status}`);
  }
  const body = (await res.json()) as { runs?: ActiveTaskRun[] };
  return Array.isArray(body.runs) ? body.runs : [];
}

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