/**
 * Workflow runner.
 *
 * Walks a validated Workflow (schema.ts) as a state machine: starting at `entry_point`,
 * it executes the current node, then follows the first `route` whose `when` condition
 * holds, until a route reaches `$end`/`$fail` or a `terminate` step.
 *
 * DECOUPLED FROM ANY HOST BY DESIGN. The runner imports neither a UI, a relay, nor Pi —
 * every effectful capability is INJECTED (RunnerDeps). That keeps the security-critical
 * control flow unit-testable with fakes, and lets a host (a Pi extension, a daemon, a CLI)
 * wire the real seams:
 *   - `runAgent`        → drive one headless LLM turn under an auto-approve gate
 *   - `askGate`         → surface a human_gate as an approval in the host UI
 *   - `runScript`       → a gated shell exec (see the fail-closed rule below)
 *   - `deferForApproval`→ record a "needs approval" notice when unattended
 *   - `loadSubWorkflow` → resolve + load a local sub-workflow ref (optional)
 *
 * SECURITY — the fail-closed posture lives HERE, not in the schema:
 *   A `script` step (and, later, a dangerous write-tool) BYPASSES the agent permission
 *   gate. When the run is UNATTENDED (no controller driving), the runner MUST NOT execute
 *   it. It pauses, records a "needs approval" notice, and stops with status `deferred`.
 *   Only when a controller is attached does a `script` step first pass through `askGate`
 *   (authenticated approval) before running.
 *
 * Template/condition evaluation is delegated to `./expr.ts` — a confined, hand-written
 * recursive-descent interpreter (no eval/Function, prototype-safe path lookups, DoS
 * bounds). That is the ONLY place the runner evaluates attacker-influenceable `{{ ... }}`
 * text (imported workflows carry it), so the eval surface is quarantined there on purpose.
 */
import { validateWorkflow, type Step, type Workflow } from "./schema.ts";
import { evalCondition, evaluate, renderTemplate } from "./expr.ts";

// ── Injected capabilities ─────────────────────────────────────────────────────

export interface AgentRunSpec {
  prompt: string;
  model?: string;
  tools?: string[];
  cwd: string;
}
export interface AgentRunResult {
  // Raw text the agent produced (for output_mode:raw / display).
  text: string;
  // Parsed structured output when the step declared an `output:` schema, else {}.
  output: Record<string, unknown>;
  status: "ok" | "error";
  error?: string;
}

export interface ScriptRunResult {
  output: Record<string, unknown>;
  status: "ok" | "error";
  error?: string;
  exitCode: number;
}

export interface RunnerDeps {
  // Drive one headless agent turn (under a safe-tools gate).
  runAgent(spec: AgentRunSpec): Promise<AgentRunResult>;
  // Execute a gated shell step. The runner ONLY calls this after the fail-closed
  // posture check (attended + approved). Never call it directly for an unattended run.
  runScript(step: Extract<Step, { type: "script" }>, cwd: string): Promise<ScriptRunResult>;
  // Surface a human_gate to the user as an approval. Resolves to the chosen option name,
  // or null when there's no controller / the user dismissed it.
  askGate(step: Extract<Step, { type: "human_gate" }>, promptText: string): Promise<string | null>;
  // Is a controller (a live UI) currently attached and driving this run?
  attended(): boolean;
  // Record a "needs approval / deferred" notice. Used when an effectful step is reached
  // unattended — the run stops, the user catches up later.
  deferForApproval(reason: string): Promise<void>;
  // Sleep for a `wait` step. Injected so tests don't actually block.
  sleep(ms: number): Promise<void>;
  log(msg: string): void;
  // Optional live event sink (stream step transitions/text to a UI).
  onEvent?(ev: RunnerEvent): void;
  // OPTIONAL: resolve + load a local sub-workflow by its (schema-validated) relative ref.
  // The host confines the path (e.g. to its workflows dir) and returns a validated
  // Workflow, or null if it can't be found/loaded. Absent → a `workflow` step fails
  // closed ("sub-workflows not supported by this host") rather than silently skipping.
  loadSubWorkflow?(ref: string): Promise<Workflow | null> | Workflow | null;
  // OPTIONAL: default fan-out concurrency cap for `parallel` groups and the ceiling for
  // `for_each` (min'd with the group's own max_concurrent). Default 4.
  concurrency?: number;
}

export type RunnerEvent =
  | { type: "step_start"; name: string; kind: string }
  | { type: "step_output"; name: string; output: Record<string, unknown> }
  | { type: "route"; from: string; to: string }
  | { type: "text"; name: string; text: string };

export type RunStatus = "success" | "failed" | "deferred";
export interface RunResult {
  status: RunStatus;
  // Workflow-level `output:` rendered against the final context (best-effort).
  output: Record<string, unknown>;
  // Why a run ended failed/deferred.
  reason?: string;
  // Every step result keyed by name — the accumulated context, for inspection/tests.
  context: RunContext;
}

// The accumulated run context. `workflow.input` holds the run inputs; every executed
// node stores its result under its own name (`{ output }`, plus `{ choice }` for gates).
export interface RunContext {
  workflow: { input: Record<string, unknown> };
  [stepName: string]: unknown;
}

// Recursion backstop for `workflow` steps — a deep or cyclic composition chain can't
// pin the runner (each level also spends the child's own max_iterations budget).
const MAX_SUBWORKFLOW_DEPTH = 8;
const DEFAULT_CONCURRENCY = 4;

// ── The node index (steps + fan-out groups share one name space) ────────────────

type Node =
  | { kind: "step"; step: Step }
  | { kind: "parallel"; group: Workflow["parallel"][number] }
  | { kind: "for_each"; group: Workflow["for_each"][number] };

function indexNodes(wf: Workflow): Map<string, Node> {
  const idx = new Map<string, Node>();
  for (const step of wf.steps) idx.set(step.name, { kind: "step", step });
  for (const group of wf.parallel) idx.set(group.name, { kind: "parallel", group });
  for (const group of wf.for_each) idx.set(group.name, { kind: "for_each", group });
  return idx;
}

// ── The runner ──────────────────────────────────────────────────────────────────

export async function runWorkflow(
  wf: Workflow,
  input: Record<string, unknown>,
  deps: RunnerDeps,
): Promise<RunResult> {
  return runWorkflowAt(wf, input, deps, 0);
}

// Internal entry that carries the sub-workflow recursion depth.
async function runWorkflowAt(
  wf: Workflow,
  input: Record<string, unknown>,
  deps: RunnerDeps,
  depth: number,
): Promise<RunResult> {
  // Refuse a graph with a hole in it — a dangling route is a fail-closed stop.
  const graphErrors = validateWorkflow(wf);
  if (graphErrors.length > 0) {
    return { status: "failed", output: {}, reason: `invalid graph: ${graphErrors.join("; ")}`, context: { workflow: { input } } };
  }

  const nodes = indexNodes(wf);
  const ctx: RunContext = { workflow: { input } };
  const defaultCwd = process.cwd();
  const maxIter = wf.workflow.limits.max_iterations;

  let cursor: string = wf.workflow.entry_point;
  let iterations = 0;

  while (cursor !== "$end" && cursor !== "$fail") {
    if (++iterations > maxIter) {
      return finish("failed", ctx, wf, `exceeded max_iterations (${maxIter}) — cyclic or runaway graph`);
    }
    const node = nodes.get(cursor);
    if (!node) return finish("failed", ctx, wf, `route reached unknown node "${cursor}"`);

    // A malformed `{{ }}` in a prompt/value fails LOUD (expr.ts throws TemplateError) — but
    // as a clean `failed` run, not an unhandled rejection out of the host.
    let outcome: { halt?: RunStatus; reason?: string };
    try {
      outcome = await execNode(node, ctx, wf, nodes, deps, defaultCwd, depth);
    } catch (err) {
      return finish("failed", ctx, wf, `step "${cursor}" errored: ${err instanceof Error ? err.message : String(err)}`);
    }
    if (outcome.halt) return finish(outcome.halt, ctx, wf, outcome.reason);

    // Follow the route graph. `null` next means the node had no matching route → $end.
    const next = pickRoute(node, ctx);
    if (cursor !== next) deps.onEvent?.({ type: "route", from: cursor, to: next ?? "$end" });
    cursor = next ?? "$end";
  }

  return finish(cursor === "$fail" ? "failed" : "success", ctx, wf);
}

// Execute one node, mutating `ctx` with its result. Returns a halt directive when the
// node ends the whole run (terminate step, the fail-closed defer of a script step, or a
// fan-out group whose failure_mode says a failed member sinks the run).
async function execNode(
  node: Node,
  ctx: RunContext,
  wf: Workflow,
  nodes: Map<string, Node>,
  deps: RunnerDeps,
  defaultCwd: string,
  depth: number,
): Promise<{ halt?: RunStatus; reason?: string }> {
  const name = nodeName(node);
  deps.onEvent?.({ type: "step_start", name, kind: nodeKind(node) });

  if (node.kind === "parallel") return execParallel(node.group, ctx, nodes, deps, defaultCwd);
  if (node.kind === "for_each") return execForEach(node.group, ctx, wf, deps, defaultCwd);

  // node.kind === "step" — dispatch on the discriminated `type`.
  const step = node.step;
  switch (step.type) {
    case "agent": {
      const r = await runAgentStep(step, ctx, deps, defaultCwd, wf);
      if (r.status === "error") return { halt: "failed", reason: `agent "${name}" failed: ${r.error ?? "unknown"}` };
      return {};
    }

    case "set": {
      const values: Record<string, unknown> = {};
      if (step.value !== undefined) {
        ctx[name] = { output: { result: renderTemplate(step.value, ctx) } };
      } else {
        for (const [k, expr] of Object.entries(step.values ?? {})) values[k] = renderTemplate(expr, ctx);
        ctx[name] = { output: values };
      }
      deps.onEvent?.({ type: "step_output", name, output: (ctx[name] as any).output });
      return {};
    }

    case "wait": {
      const ms = parseDuration(step.duration, ctx);
      await deps.sleep(ms);
      return {};
    }

    case "human_gate": {
      const choice = await deps.askGate(step, renderTemplate(step.prompt, ctx));
      // No controller / dismissed → fail-closed: don't guess a choice, stop the run.
      if (choice === null) return { halt: "deferred", reason: `human_gate "${name}" had no response` };
      ctx[name] = { choice };
      return {};
    }

    case "script": {
      // THE fail-closed seam. A script is direct code execution.
      if (!deps.attended()) {
        await deps.deferForApproval(`Workflow "${wf.workflow.name}" paused at script step "${name}" — approval required to run "${step.command}".`);
        return { halt: "deferred", reason: `script "${name}" requires approval (unattended)` };
      }
      // Attended: require an explicit authenticated approval before running.
      const ok = await deps.askGate(
        { type: "human_gate", name, prompt: `Run \`${step.command} ${step.args.join(" ")}\`?`, options: [{ name: "approve" }, { name: "deny" }], routes: [] },
        `Run \`${step.command} ${step.args.join(" ")}\`?`,
      );
      if (ok !== "approve") return { halt: "deferred", reason: `script "${name}" not approved` };
      const r = await deps.runScript(step, step.working_dir ?? defaultCwd);
      ctx[name] = { output: r.output };
      if (r.status === "error") return { halt: "failed", reason: `script "${name}" exited ${r.exitCode}: ${r.error ?? ""}` };
      return {};
    }

    case "workflow":
      return execSubWorkflow(step, ctx, deps, depth);

    case "terminate": {
      return { halt: step.status === "success" ? "success" : "failed", reason: step.reason };
    }
  }
}

// ── Fan-out: parallel ────────────────────────────────────────────────────────────

// Run a static group of already-defined agent steps concurrently, bounded by the cap,
// honoring failure_mode. Each member writes its own ctx[member] entry (distinct keys, so
// concurrent writes don't race); the group also aggregates them under ctx[group] so a
// route can read `{{ group.outputs.member.field }}`.
async function execParallel(
  group: Workflow["parallel"][number],
  ctx: RunContext,
  nodes: Map<string, Node>,
  deps: RunnerDeps,
  cwd: string,
): Promise<{ halt?: RunStatus; reason?: string }> {
  const members = group.agents
    .map((n) => nodes.get(n))
    .filter((m): m is Extract<Node, { kind: "step" }> => m?.kind === "step" && m.step.type === "agent")
    .map((m) => m.step as Extract<Step, { type: "agent" }>);

  const cap = poolSize(deps);
  const stopOnError = group.failure_mode === "fail_fast";
  const results = await runPool(members, cap, stopOnError, (step) => runAgentStep(step, ctx, deps, cwd));

  const outputs: Record<string, unknown> = {};
  let firstError: string | undefined;
  for (let i = 0; i < members.length; i++) {
    const r = results[i];
    if (r) {
      outputs[members[i].name] = r.output;
      if (r.status === "error" && firstError === undefined) firstError = `${members[i].name}: ${r.error ?? "unknown"}`;
    }
  }
  ctx[group.name] = { outputs };
  deps.onEvent?.({ type: "step_output", name: group.name, output: { outputs } });

  // continue_on_error tolerates member failures; fail_fast/all_or_nothing sink the run.
  if (firstError !== undefined && group.failure_mode !== "continue_on_error") {
    return { halt: "failed", reason: `parallel group "${group.name}" failed (${firstError})` };
  }
  return {};
}

// ── Fan-out: for_each ────────────────────────────────────────────────────────────

// Dynamic fan-out: resolve `source` to an array, then run the inline agent once per item
// with `{{ <as> }}` (and `{{ _index }}`) bound in a scoped context, bounded by
// min(max_concurrent, cap). Outputs collect into an array, or a keyed object when
// `key_by` is set (evaluated per item against the scoped context).
async function execForEach(
  group: Workflow["for_each"][number],
  ctx: RunContext,
  wf: Workflow,
  deps: RunnerDeps,
  cwd: string,
): Promise<{ halt?: RunStatus; reason?: string }> {
  const items = asArray(safeEval(group.source, ctx));
  const cap = Math.min(group.max_concurrent, poolSize(deps));
  const stopOnError = group.failure_mode === "fail_fast";

  const scopes = items.map((item, i) => ({ item, i, scoped: { ...ctx, [group.as]: item, _index: i } as RunContext }));
  const results = await runPool(scopes, cap, stopOnError, async ({ scoped }) => {
    const r = await deps.runAgent({
      prompt: renderTemplate(group.agent.prompt, scoped),
      model: group.agent.model ?? wf.workflow.default_model,
      tools: group.agent.tools,
      cwd,
    });
    return r;
  });

  const keyed: Record<string, unknown> = {};
  const list: unknown[] = [];
  let firstError: string | undefined;
  for (let i = 0; i < scopes.length; i++) {
    const r = results[i];
    if (!r) continue;
    if (r.status === "error" && firstError === undefined) firstError = `item ${i}: ${r.error ?? "unknown"}`;
    if (group.key_by) {
      const k = String(safeEval(group.key_by, scopes[i].scoped) ?? i);
      keyed[k] = r.output;
    }
    list.push(r.output);
  }

  const outputs: unknown = group.key_by ? keyed : list;
  ctx[group.name] = { outputs };
  deps.onEvent?.({ type: "step_output", name: group.name, output: { outputs } as Record<string, unknown> });

  if (firstError !== undefined && group.failure_mode !== "continue_on_error") {
    return { halt: "failed", reason: `for_each group "${group.name}" failed (${firstError})` };
  }
  return {};
}

// ── Composition: sub-workflow ────────────────────────────────────────────────────

async function execSubWorkflow(
  step: Extract<Step, { type: "workflow" }>,
  ctx: RunContext,
  deps: RunnerDeps,
  depth: number,
): Promise<{ halt?: RunStatus; reason?: string }> {
  if (!deps.loadSubWorkflow) {
    return { halt: "failed", reason: `sub-workflow step "${step.name}" — this host does not support sub-workflows` };
  }
  if (depth + 1 > MAX_SUBWORKFLOW_DEPTH) {
    return { halt: "failed", reason: `sub-workflow step "${step.name}" exceeded max nesting depth (${MAX_SUBWORKFLOW_DEPTH})` };
  }
  const child = await deps.loadSubWorkflow(step.workflow);
  if (!child) {
    return { halt: "failed", reason: `sub-workflow step "${step.name}" — could not load "${step.workflow}"` };
  }

  // Map the parent context into the child's input (each an opaque expr over the parent).
  const childInput: Record<string, unknown> = {};
  for (const [k, expr] of Object.entries(step.input_mapping ?? {})) childInput[k] = evaluate(expr, ctx);

  const childResult = await runWorkflowAt(child, childInput, deps, depth + 1);
  ctx[step.name] = { output: childResult.output, status: childResult.status };
  deps.onEvent?.({ type: "step_output", name: step.name, output: childResult.output });

  // A failed/deferred child sinks the parent — a black-box step that failed is a failure.
  if (childResult.status !== "success") {
    return { halt: childResult.status, reason: `sub-workflow "${step.name}" ${childResult.status}${childResult.reason ? `: ${childResult.reason}` : ""}` };
  }
  return {};
}

// ── Agent step ────────────────────────────────────────────────────────────────────

async function runAgentStep(
  step: Extract<Step, { type: "agent" }>,
  ctx: RunContext,
  deps: RunnerDeps,
  defaultCwd: string,
  wf?: Workflow,
): Promise<AgentRunResult> {
  const r = await deps.runAgent({
    prompt: renderTemplate(step.prompt, ctx),
    model: step.model ?? wf?.workflow.default_model,
    tools: step.tools,
    cwd: defaultCwd,
  });
  ctx[step.name] = { output: r.output };
  if (r.text) deps.onEvent?.({ type: "text", name: step.name, text: r.text });
  deps.onEvent?.({ type: "step_output", name: step.name, output: r.output });
  return r;
}

// ── Concurrency pool ──────────────────────────────────────────────────────────────

function poolSize(deps: RunnerDeps): number {
  const c = deps.concurrency ?? DEFAULT_CONCURRENCY;
  return c > 0 ? Math.floor(c) : DEFAULT_CONCURRENCY;
}

// Run `worker` over `items` with at most `limit` in flight. Results land at each item's
// ORIGINAL index (holes stay undefined). When `stopOnError` is set (fail_fast), a worker
// whose result has status "error" — or that throws — stops NEW launches; already-running
// tasks are awaited. A worker that throws is captured as an "error" AgentRunResult so one
// rejection can never take down the whole run.
async function runPool<T>(
  items: T[],
  limit: number,
  stopOnError: boolean,
  worker: (item: T) => Promise<AgentRunResult>,
): Promise<Array<AgentRunResult | undefined>> {
  const results: Array<AgentRunResult | undefined> = new Array(items.length);
  let next = 0;
  let stop = false;

  async function pump(): Promise<void> {
    while (true) {
      if (stop) return;
      const i = next++;
      if (i >= items.length) return;
      let r: AgentRunResult;
      try {
        r = await worker(items[i]);
      } catch (err) {
        r = { text: "", output: {}, status: "error", error: err instanceof Error ? err.message : String(err) };
      }
      results[i] = r;
      if (stopOnError && r.status === "error") stop = true;
    }
  }

  const workers = Array.from({ length: Math.max(1, Math.min(limit, items.length)) }, () => pump());
  await Promise.all(workers);
  return results;
}

// ── Routing ────────────────────────────────────────────────────────────────────────

// Pick the first route whose `when` holds; a `when`-less route is the fallthrough.
// Returns the target name, `$end`/`$fail`, or null (no route matched → treat as $end).
function pickRoute(node: Node, ctx: RunContext): string | null {
  const routes = nodeRoutes(node);
  if (!routes) return null; // terminate has no routes; it already halted.
  for (const r of routes) {
    if (r.when === undefined || evalCondition(r.when, ctx)) return r.to;
  }
  return null;
}

function finish(status: RunStatus, ctx: RunContext, wf: Workflow, reason?: string): RunResult {
  const output: Record<string, unknown> = {};
  for (const [k, expr] of Object.entries(wf.output ?? {})) {
    try { output[k] = renderTemplate(expr, ctx); } catch { output[k] = null; }
  }
  return { status, output, reason, context: ctx };
}

// ── Node helpers ────────────────────────────────────────────────────────────────

function nodeName(node: Node): string {
  return node.kind === "step" ? node.step.name : node.group.name;
}
function nodeKind(node: Node): string {
  return node.kind === "step" ? node.step.type : node.kind;
}
function nodeRoutes(node: Node): { to: string; when?: string }[] | null {
  if (node.kind === "step") return "routes" in node.step ? node.step.routes : null;
  return node.group.routes;
}

// ── Template/expression helpers (thin wrappers over the confined ./expr.ts engine) ──

// Evaluate a bare expression for a `for_each` source. Fail-closed: a malformed source
// (syntax error) yields undefined → an empty item list, never a thrown run.
function safeEval(expr: string, ctx: unknown): unknown {
  try { return evaluate(expr, ctx); } catch { return undefined; }
}

function asArray(v: unknown): unknown[] {
  return Array.isArray(v) ? v : [];
}

// Parse a `wait` duration (number = seconds, or "500ms"/"30s"/"5m"/"1h") to millis.
function parseDuration(d: number | string, ctx: RunContext): number {
  if (typeof d === "number") return d * 1000;
  const rendered = renderTemplate(d, ctx);
  const m = rendered.match(/^(\d+(?:\.\d+)?)(ms|s|m|h)$/);
  if (!m) return 0;
  const n = Number(m[1]);
  return n * ({ ms: 1, s: 1000, m: 60_000, h: 3_600_000 }[m[2]] ?? 1000);
}