Arcrun/cypher-executor/src/lib/paused-runs.ts

/**
 * Paused workflow runs：節點回 pending 時把 run state 持久化進 KV，
 * webhook callback 進來時撿回繼續執行
 *
 * SDD: matrix/arcrun/.agents/specs/resumable-workflow/design.md §2.1
 *
 * KV key: paused_run:{task_id}
 * TTL: 24h（避免 KV 累積，超過就 GC）
 *
 * 設計筆記：
 * - 用 task_id 當 key（daemon 派的 unique id），不用 run_id（同 run 可能多 paused 節點 v2）
 * - consume = load + delete 原子操作（避免重複 callback 重複執行）
 */

import type { ExecutionGraph, TraceStep } from '../types';

export interface PausedRunState {
  run_id: string;
  graph: ExecutionGraph;
  paused_node_id: string;
  paused_context: Record<string, unknown>;
  paused_pending_result: Record<string, unknown>;  // 節點回的 {pending, task_id, ...}
  trace_so_far: TraceStep[];
  api_key?: string;
  expires_at: number;  // unix ms
  // resume 時用來 parse callback result 的 recipe output 規格（resumable + recipe 整合）
  recipe_output_format?: 'text' | 'json';
  recipe_output_required_fields?: string[];
}

const KEY_PREFIX = 'paused_run:';
/**
 * Per-user paused index：列出某 api_key 當前 paused tasks 不依賴 CF KV list（強 eventual
 * consistent，30-60s 延遲）。改維護一個 user-keyed JSON list，list 操作改 single KV.get。
 *
 * Key: `paused_idx:{api_key}`
 * Value: JSON Array<{task_id, paused_node_id, run_id, workflow_name?, expires_at, persisted_at}>
 *
 * 對應 LI SDD M2.1 — /executions/paused endpoint 即時性。
 */
const IDX_PREFIX = 'paused_idx:';
const TTL_SECONDS = 24 * 60 * 60;

export type PausedIndexEntry = {
  task_id: string;
  run_id: string;
  paused_node_id: string;
  workflow_name?: string;
  expires_at: number;
  persisted_at: number;
};

type KvBinding = {
  get: (key: string) => Promise<string | null>;
  put: (key: string, value: string, options?: { expirationTtl?: number }) => Promise<void>;
  delete: (key: string) => Promise<void>;
};

async function readIndex(kv: KvBinding, apiKey: string): Promise<PausedIndexEntry[]> {
  const raw = await kv.get(`${IDX_PREFIX}${apiKey}`);
  if (!raw) return [];
  try {
    const arr = JSON.parse(raw);
    return Array.isArray(arr) ? arr : [];
  } catch {
    return [];
  }
}

async function writeIndex(kv: KvBinding, apiKey: string, entries: PausedIndexEntry[]): Promise<void> {
  // 過濾過期項目（避免 index 爆量）
  const now = Date.now();
  const fresh = entries.filter((e) => e.expires_at > now);
  await kv.put(`${IDX_PREFIX}${apiKey}`, JSON.stringify(fresh), { expirationTtl: TTL_SECONDS });
}

export async function persistPausedRun(
  kv: KvBinding,
  taskId: string,
  state: PausedRunState,
): Promise<void> {
  // 1) 寫單一 task state
  await kv.put(`${KEY_PREFIX}${taskId}`, JSON.stringify(state), { expirationTtl: TTL_SECONDS });

  // 2) 維護 per-user index（讓 /executions/paused list 不靠 KV list 即時拿到）
  if (state.api_key) {
    const idx = await readIndex(kv, state.api_key);
    // 去重（重複 paused 同 task_id 時覆蓋）
    const filtered = idx.filter((e) => e.task_id !== taskId);
    filtered.unshift({
      task_id: taskId,
      run_id: state.run_id,
      paused_node_id: state.paused_node_id,
      workflow_name: state.graph.name,
      expires_at: state.expires_at,
      persisted_at: Date.now(),
    });
    // 限 100 筆避免 index 無限長（超過捨棄最舊）
    await writeIndex(kv, state.api_key, filtered.slice(0, 100));
  }
}

export async function loadPausedRun(
  kv: KvBinding,
  taskId: string,
): Promise<PausedRunState | null> {
  const raw = await kv.get(`${KEY_PREFIX}${taskId}`);
  if (!raw) return null;
  try {
    return JSON.parse(raw) as PausedRunState;
  } catch {
    return null;
  }
}

/**
 * 列某 api_key 當前 paused tasks。走 per-user index（強 consistent，無 KV list 延遲）
 */
export async function listPausedRunsByApiKey(
  kv: KvBinding,
  apiKey: string,
  limit = 20,
): Promise<PausedIndexEntry[]> {
  const idx = await readIndex(kv, apiKey);
  const now = Date.now();
  return idx.filter((e) => e.expires_at > now).slice(0, limit);
}

/**
 * 原子讀+刪：避免同 task_id 重複 callback 重複執行下游
 * （CF KV 沒真原子操作，但 delete 失敗不影響 load 已成功）
 */
export async function consumePausedRun(
  kv: KvBinding,
  taskId: string,
): Promise<PausedRunState | null> {
  const state = await loadPausedRun(kv, taskId);
  if (!state) return null;
  await kv.delete(`${KEY_PREFIX}${taskId}`).catch(() => {
    // delete 失敗不擋，最多就重複執行一次（接受）
  });
  // 同步從 per-user index 移除
  if (state.api_key) {
    const idx = await readIndex(kv, state.api_key);
    const filtered = idx.filter((e) => e.task_id !== taskId);
    await writeIndex(kv, state.api_key, filtered).catch(() => {});
  }
  return state;
}

/** 偵測 component result 是否為「需要 resume」的 pending pattern */
export function isResumablePending(result: unknown): { task_id: string } | null {
  if (!result || typeof result !== 'object') return null;
  const r = result as Record<string, unknown>;
  if (r.pending !== true) return null;
  if (typeof r.task_id !== 'string' || !r.task_id) return null;
  return { task_id: r.task_id };
}

/**
 * Parse claude_api result with recipe output format.
 * 同步路徑跟 resume 路徑都用同一個解析器，避免邏輯歪掉。
 *
 * 輸入：result（可能是 {data:{text:"..."}} 或 {text:"..."}）
 * 輸出：parsed object 或 fallback 結構
 */
export function parseRecipeOutput(
  result: unknown,
  format: 'text' | 'json' | undefined,
  requiredFields: string[] | undefined,
): unknown {
  if (format !== 'json' || !result || typeof result !== 'object') return result;
  const r = result as Record<string, unknown>;
  const text = (r.data as Record<string, unknown> | undefined)?.text ?? r.text;
  if (typeof text !== 'string') return result;

  // 剝除 ```json ... ``` markdown fence（Claude 常這樣包）
  let jsonText = String(text).trim();
  const fenceMatch = jsonText.match(/^```(?:json)?\s*\n([\s\S]*?)\n```$/);
  if (fenceMatch) jsonText = fenceMatch[1].trim();

  try {
    const parsed = JSON.parse(jsonText);
    if (requiredFields && parsed && typeof parsed === 'object') {
      const missing = requiredFields.filter((f) => !(f in (parsed as Record<string, unknown>)));
      if (missing.length > 0) {
        return { success: false, error: `recipe output 缺欄位: ${missing.join(', ')}`, raw: parsed };
      }
    }
    // 把 parsed 的欄位 spread 到 top-level，FOREACH / 下游 {{var}} 都好取
    return { success: true, data: parsed, ...(parsed && typeof parsed === 'object' ? parsed as Record<string, unknown> : {}) };
  } catch (e) {
    return { success: false, error: `recipe output JSON parse 失敗: ${e instanceof Error ? e.message : String(e)}`, raw_text: text };
  }
}