import {
  type AgentRunHandle,
  CapabilityError,
  type CapabilityRegistry,
  type ExternalUsage,
  type HandlerContext,
  redactString,
  type SubagentEventBody,
} from "@umt/agent";
import {
  type AgentBudget,
  type AgentPlanStep,
  type AgentRunEvent,
  type AgentRunMode,
  type AgentRunStatus,
  type AgentUsageTotals,
  agentRunEventToBusEvent,
  CAPABILITY_CATALOG,
  type CapabilityKind,
  capabilitySpec,
  type GlobalSettings,
  type KnownCapabilityId,
  type SubagentBudget,
  type SubagentResult,
} from "@umt/contracts";
import { type AgentRunRow, type DB, type Kysely, sql } from "@umt/db";
import type { ServerContext } from "../types.ts";
import type { AgentEnv, RunScope } from "./runner.ts";

// 子代理(W7-B,REQ §10 "一个会话处理一个有界任务;子代理继承父任务授权上限,共享总预算与取消状态"):
// - agent.spawn:父会话派生子 agent_run(parent_run_id / depth / spawn_reason),同 runAgent 任务、独立会话文件;
//   能力白名单 ⊆ 父、模式不高于父、预算从父剩余预算划拨;settings.pi.subagents{enabled,maxConcurrent,maxDepth}。
// - 并发:子会话在 runAgent 里先占名额(父会话下同时 running 的子会话 < maxConcurrent,咨询锁内原子判定),
//   没名额就 waitForFact 挂起(不占 worker、不占模型)。
// - agent.await / agent.spawn{await:true}:等子会话结束,返回脱敏后的汇报摘要。runAgent 驱动的会话里,
//   子会话还没结束时工具调用立即返回(挂起):关闭本会话(工具调用留在 JSONL 里未完成),runAgent 在 agent_run.await_wake
//   登记唤醒信号后用 waitForFact 等子会话落定(不占 worker 槽位、不调用模型);子会话落定时 wakeParent 写 fact
//   唤醒父会话,再打开会话 resume —— 两个能力都是 replay=safe,恢复时重放一次,子会话汇报作为工具结果续跑。
//   见 AwaitSuspension。
// - 事件:父会话运行进程把子会话事件按父 seq 镜像进父事件流(subagent_spawned / subagent_event / subagent_completed);
//   父会话已不在运行时,由直接写入器在咨询锁内补记 subagent_completed。
// - 取消:父会话取消递归传播到全部子孙会话;预算:父会话运行时把子孙会话用量计入自己的预算。

export const SUBAGENT_CAPABILITIES = ["agent.spawn", "agent.await"] as const;
/** 子会话"已结束":终态;waiting(模型已结束,只在等下载/入库落定)对父会话来说也算有结果 */
export const TERMINAL: AgentRunStatus[] = ["succeeded", "failed", "cancelled", "budget_exceeded"];
const SETTLED: AgentRunStatus[] = [...TERMINAL, "waiting"];
const ALL_KINDS: CapabilityKind[] = ["read", "preview", "propose", "approve", "execute"];
const PLAN_ONLY_KINDS: CapabilityKind[] = ["read", "preview", "propose"];
/** 一个会话最多派生的子会话数(预算之外的兜底) */
export const MAX_CHILDREN_PER_RUN = 10;
/** 同目标去重窗口(AG-4):已完成的子会话在这段时间内按同目标复用 */
export const SPAWN_DEDUPE_WINDOW_MS = 10 * 60_000;

/** 目标归一化:全半角统一、忽略大小写、空白与标点("查一下 观众 站点。" ≡ "查一下观众站点") */
export function normalizeGoal(goal: string): string {
  return goal
    .normalize("NFKC")
    .toLowerCase()
    .replace(/[\s\p{P}\p{S}]+/gu, "");
}

/**
 * 同一父会话里同目标、可复用的子会话:还没落定(排队 / 运行中),或 10 分钟内成功完成 / 已在等下载入库。
 * 失败、取消、超预算的不复用(父会话可以换个预算或说法重派)。
 */
export async function findSameGoalChild(
  db: Kysely<DB>,
  parentId: string,
  goal: string,
  now = Date.now(),
): Promise<AgentRunRow | null> {
  const key = normalizeGoal(goal);
  if (!key) return null;
  const rows = await db
    .selectFrom("agent_run")
    .selectAll()
    .where("parent_run_id", "=", parentId)
    .orderBy("created_at", "desc")
    .execute();
  for (const r of rows) {
    if (normalizeGoal(r.goal) !== key) continue;
    if (!SETTLED.includes(r.status)) return r;
    if (r.status !== "succeeded" && r.status !== "waiting") continue;
    const at = (r.finished_at ?? r.updated_at)?.getTime() ?? 0;
    if (now - at <= SPAWN_DEDUPE_WINDOW_MS) return r;
  }
  return null;
}
const MIRRORED_TYPES = [
  "run_started",
  "tool_call_start",
  "tool_call_end",
  "tool_call_rejected",
  "run_completed",
  "run_failed",
] as const;

type SubagentSettings = GlobalSettings["pi"]["subagents"];

export function subagentSettings(s: GlobalSettings): SubagentSettings {
  return s.pi.subagents ?? { enabled: s.pi.subagentsEnabled, maxConcurrent: 2, maxDepth: 1 };
}

const clip = (s: string, n: number) => (s.length > n ? `${s.slice(0, n)}…` : s);
const safe = (s: string | null | undefined, n = 1500) => clip(redactString(s ?? ""), n);

// ── 进程内活跃会话(镜像泵 + 用量缓存) ──

/**
 * 挂起中的工具调用(agent.spawn{await:true})抛出这个错误:账本记 failed、不缓存结果,
 * 会话恢复时重放同一调用(幂等键相同 → 同一个子会话)再取汇报。会话此时已关闭,模型看不到它。
 */
export class AwaitSuspendedError extends CapabilityError {
  constructor() {
    super("precondition", "会话已挂起等待子会话结束，恢复后重放本调用");
  }
}

/** runAgent 挂起期间登记在 agent_run.await_wake 的唤醒信号(子会话落定时 wakeParent 按它 signal) */
export interface AwaitWake {
  engineRunId: string;
  fact: string;
}

/**
 * 会话挂起请求:runAgent 据此 waitForFact,条件满足后以 resumedAwaits 重开会话并重放挂起的调用。
 * - subagents(缺省):agent.await / agent.spawn{await:true} 等子会话落定;
 * - options:agent.propose_options 等用户选方案(decisionId);
 * - approval:需审批的能力等管理员批准 / 拒绝(approvalId)。
 */
export interface AwaitSuspension {
  kind?: "subagents" | "options" | "approval";
  /** 挂起的工具调用(恢复时重放,同一 invocationId 不再挂起) */
  invocationId: string;
  /** 等待的子会话;null = 本会话全部子会话 */
  runIds: string[] | null;
  /** 超过这个时间(ISO)不再等,恢复后返回当前状态(未结束的标 pending / 按超时处理) */
  deadline: string;
  decisionId?: string;
  approvalId?: string;
}

export interface LiveRun {
  handle: AgentRunHandle;
  /** runAgent 的 drive 打开的会话才可挂起;已挂起过的 invocationId 恢复时不再挂起 */
  suspend?: { resumed: ReadonlySet<string>; requested: AwaitSuspension | null };
  /** 把子会话的新事件 / 新派生 / 结束镜像进本会话事件流(串行) */
  pump(): Promise<void>;
  /** 子孙会话已用量(同步读缓存,pump 刷新) */
  externalUsage(): ExternalUsage;
  /** agent.plan 的当前步骤清单(首次用到时从事件流载入) */
  plan?: { steps: AgentPlanStep[]; revision: number } | null;
}

const live = new Map<string, LiveRun>();
export const liveRun = (runId: string) => live.get(runId);

/** drive 打开会话后调用;返回的 stop() 在关闭前做最后一次镜像 */
export async function attachLiveRun(
  env: AgentEnv,
  runId: string,
  handle: AgentRunHandle,
  opts: { resumedAwaits?: readonly string[] } = {},
): Promise<{ run: LiveRun; stop(): Promise<void> }> {
  const db = env.ctx.db;
  const prior = await db
    .selectFrom("agent_run_event")
    .select(["type", "payload"])
    .where("run_id", "=", runId)
    .where("type", "in", ["subagent_spawned", "subagent_event", "subagent_completed"])
    .execute();
  const spawned = new Set<string>();
  const completed = new Set<string>();
  const mirrored = new Map<string, number>();
  for (const e of prior) {
    const p = e.payload as { childRunId: string; childSeq?: number };
    if (e.type === "subagent_spawned") spawned.add(p.childRunId);
    else if (e.type === "subagent_completed") completed.add(p.childRunId);
    else mirrored.set(p.childRunId, Math.max(mirrored.get(p.childRunId) ?? -1, p.childSeq ?? -1));
  }
  let usage: ExternalUsage = await descendantUsage(env.ctx, runId);
  let chain: Promise<void> = Promise.resolve();

  const once = async () => {
    const children = await db
      .selectFrom("agent_run")
      .selectAll()
      .where("parent_run_id", "=", runId)
      .orderBy("created_at")
      .execute();
    for (const c of children) {
      if (!spawned.has(c.id)) {
        if (!handle.appendEvent(spawnedBody(c))) return;
        spawned.add(c.id);
      }
      const events = await db
        .selectFrom("agent_run_event")
        .select(["seq", "type", "payload"])
        .where("run_id", "=", c.id)
        .where("seq", ">", mirrored.get(c.id) ?? -1)
        .where("type", "in", [...MIRRORED_TYPES])
        .orderBy("seq")
        .limit(200)
        .execute();
      for (const e of events) {
        if (!handle.appendEvent(childEventBody(c.id, Number(e.seq), e.type, e.payload as Record<string, unknown>)))
          return;
        mirrored.set(c.id, Number(e.seq));
      }
      if (TERMINAL.includes(c.status) && !completed.has(c.id)) {
        if (!handle.appendEvent(completedBody(c))) return;
        completed.add(c.id);
      }
    }
    usage = await descendantUsage(env.ctx, runId);
  };
  const pump = () => {
    const next = chain
      .then(once)
      .catch((err) => env.ctx.logger.warn({ runId, err: { message: (err as Error).message } }, "子会话事件镜像失败"));
    chain = next;
    return next;
  };
  const run: LiveRun = {
    handle,
    pump,
    externalUsage: () => usage,
    suspend: { resumed: new Set(opts.resumedAwaits ?? []), requested: null },
  };
  live.set(runId, run);
  return {
    run,
    async stop() {
      await pump();
      if (live.get(runId) === run) live.delete(runId);
    },
  };
}

function spawnedBody(c: AgentRunRow): SubagentEventBody {
  const scope = (c.scope ?? {}) as Partial<RunScope>;
  const budget = Object.fromEntries(
    Object.entries((c.budget ?? {}) as Record<string, unknown>).filter(([, v]) => typeof v === "number"),
  ) as Record<string, number>;
  return {
    type: "subagent_spawned",
    childRunId: c.id,
    goal: safe(c.goal, 500),
    mode: c.mode,
    depth: c.depth,
    capabilities: scope.capabilities ?? [],
    budget,
  };
}

function childEventBody(childId: string, seq: number, type: string, p: Record<string, unknown>): SubagentEventBody {
  const str = (k: string) => (typeof p[k] === "string" ? (p[k] as string) : "");
  const summary =
    type === "tool_call_start"
      ? str("inputSummary")
      : type === "tool_call_end"
        ? `${p.isError ? "失败 " : ""}${str("outputSummary")}`
        : type === "tool_call_rejected"
          ? str("reason")
          : type === "run_completed"
            ? str("finalText")
            : type === "run_failed"
              ? `${str("reason")} ${str("message")}`
              : type === "run_started"
                ? p.resumed
                  ? "续跑"
                  : "开始"
                : "";
  return {
    type: "subagent_event",
    childRunId: childId,
    childSeq: seq,
    childType: type,
    ...(typeof p.capabilityId === "string" ? { capabilityId: p.capabilityId } : {}),
    summary: safe(summary, 300),
  };
}

function usageOf(r: { usage: unknown }): AgentUsageTotals {
  const u = (r.usage ?? {}) as Partial<AgentUsageTotals>;
  return {
    inputTokens: u.inputTokens ?? 0,
    outputTokens: u.outputTokens ?? 0,
    cacheReadTokens: u.cacheReadTokens ?? 0,
    cacheWriteTokens: u.cacheWriteTokens ?? 0,
    totalTokens: u.totalTokens ?? 0,
    costUsd: u.costUsd ?? 0,
  };
}

function completedBody(c: AgentRunRow): SubagentEventBody {
  return {
    type: "subagent_completed",
    childRunId: c.id,
    status: c.status,
    summary: safe(c.summary, 1500),
    usage: usageOf(c),
    toolCalls: c.tool_call_count,
    ...(c.started_at && c.finished_at
      ? { durationMs: Math.max(0, new Date(c.finished_at).getTime() - new Date(c.started_at).getTime()) }
      : {}),
  };
}

// ── 用量与会话树 ──

/** 全部子孙会话(不含自身) */
export async function descendants(ctx: ServerContext, runId: string): Promise<AgentRunRow[]> {
  const r = await sql<AgentRunRow>`
    with recursive tree as (
      select * from agent_run where parent_run_id = ${runId}
      union all
      select a.* from agent_run a join tree t on a.parent_run_id = t.id where t.depth < 16
    )
    select * from tree order by depth, created_at`.execute(ctx.db);
  return r.rows;
}

/** 子孙会话已用量(工具调用实时;token/费用在各自本轮结束时写入) */
export async function descendantUsage(ctx: ServerContext, runId: string): Promise<ExternalUsage> {
  const rows = await descendants(ctx, runId);
  return rows.reduce(
    (a, r) => {
      const u = usageOf(r);
      return {
        toolCalls: a.toolCalls + r.tool_call_count,
        totalTokens: a.totalTokens + u.totalTokens,
        costUsd: a.costUsd + u.costUsd,
      };
    },
    { toolCalls: 0, totalTokens: 0, costUsd: 0 },
  );
}

/** 祖先链(不含自身,近到远) */
export async function ancestors(db: Kysely<DB>, runId: string): Promise<AgentRunRow[]> {
  const r = await sql<AgentRunRow & { lvl: number }>`
    with recursive up as (
      select p.*, 1 as lvl from agent_run c join agent_run p on p.id = c.parent_run_id where c.id = ${runId}
      union all
      select p.*, u.lvl + 1 from agent_run p join up u on p.id = u.parent_run_id where u.lvl < 16
    )
    select * from up order by lvl`.execute(db);
  return r.rows;
}

// ── 取消传播 ──

/** 父会话取消 → 全部未结束的子孙会话写取消标志(各自的会话监视器中止模型、finalize 按取消收尾) */
export async function cascadeCancel(ctx: ServerContext, runId: string, reason: string): Promise<number> {
  const rows = (await descendants(ctx, runId)).filter((r) => !TERMINAL.includes(r.status));
  if (!rows.length) return 0;
  const now = new Date();
  const ids = rows.map((r) => r.id);
  await ctx.db
    .updateTable("agent_run")
    .set({
      cancel_requested_at: now,
      error: JSON.stringify({ abortReason: `父会话已取消：${reason}`, by: `agent:${runId}` }),
    })
    .where("id", "in", ids)
    .where("cancel_requested_at", "is", null)
    .execute();
  const taskIds = rows.map((r) => r.task_id).filter((x): x is string => !!x);
  if (taskIds.length)
    await ctx.db
      .updateTable("task")
      .set({ cancel_requested_at: now, cancel_reason: `父会话已取消：${reason}` })
      .where("id", "in", taskIds)
      .where("cancel_requested_at", "is", null)
      .execute();
  return rows.length;
}

// ── 父会话不在运行时:直接补记 subagent_completed(咨询锁内去重) ──

export async function recordChildCompletion(ctx: ServerContext, child: AgentRunRow): Promise<boolean> {
  const parentId = child.parent_run_id;
  if (!parentId || !TERMINAL.includes(child.status)) return false;
  const event = await ctx.db.transaction().execute(async (trx) => {
    await sql`select pg_advisory_xact_lock(hashtext(${`agent_run_event:${parentId}`}))`.execute(trx);
    const parent = await trx.selectFrom("agent_run").select(["status"]).where("id", "=", parentId).executeTakeFirst();
    // 父会话正在运行:由它的镜像泵按自身 seq 写入
    if (!parent || parent.status === "running" || parent.status === "queued") return null;
    const dup = await sql<{ n: number }>`
      select count(*)::int as n from agent_run_event
       where run_id = ${parentId} and type = 'subagent_completed' and payload->>'childRunId' = ${child.id}`.execute(
      trx,
    );
    if ((dup.rows[0]?.n ?? 0) > 0) return null;
    const max = await sql<{ m: string | null }>`
      select max(seq)::text as m from agent_run_event where run_id = ${parentId}`.execute(trx);
    const seq = max.rows[0]?.m === null || max.rows[0]?.m === undefined ? 0 : Number(max.rows[0].m) + 1;
    const body = completedBody(child);
    const { type, ...payload } = body;
    const at = new Date();
    await trx
      .insertInto("agent_run_event")
      .values({ run_id: parentId, seq, type, at, payload: JSON.stringify(payload) })
      .execute();
    await trx
      .updateTable("agent_run")
      .set({ last_event_seq: sql`greatest(last_event_seq, ${seq})` })
      .where("id", "=", parentId)
      .execute();
    return { ...body, runId: parentId, seq, at: at.toISOString() } as AgentRunEvent;
  });
  if (!event) return false;
  const bus = agentRunEventToBusEvent(event);
  if (bus) void ctx.events.publish(bus.type, bus.payload);
  return true;
}

/** 父会话收尾后:把已结束但还没记进父事件流的子会话补记上 */
export async function catchUpChildren(ctx: ServerContext, runId: string): Promise<number> {
  const rows = await ctx.db
    .selectFrom("agent_run")
    .selectAll()
    .where("parent_run_id", "=", runId)
    .where("status", "in", TERMINAL)
    .execute();
  let n = 0;
  for (const r of rows) if (await recordChildCompletion(ctx, r)) n++;
  return n;
}

// ── 并发名额 ──

/**
 * 子会话占名额:父会话下同时 running 的子会话 < maxConcurrent 时把自己置为 running(咨询锁内原子判定)。
 * 返回 "claimed" | "cancelled" | null(没名额,继续等)。非子会话或已占过名额直接 "claimed"。
 */
export async function claimSubagentSlot(ctx: ServerContext, runId: string): Promise<"claimed" | "cancelled" | null> {
  const s = subagentSettings(await ctx.settings.all());
  return ctx.db.transaction().execute(async (trx) => {
    const run = await trx
      .selectFrom("agent_run")
      .select(["parent_run_id", "status", "cancel_requested_at", "started_at"])
      .where("id", "=", runId)
      .executeTakeFirst();
    if (!run?.parent_run_id) return "claimed";
    if (run.cancel_requested_at) return "cancelled";
    if (run.started_at) return "claimed";
    await sql`select pg_advisory_xact_lock(hashtext(${`subagent_slot:${run.parent_run_id}`}))`.execute(trx);
    const busy = await sql<{ n: number }>`
      select count(*)::int as n from agent_run
       where parent_run_id = ${run.parent_run_id} and id <> ${runId} and status = 'running'`.execute(trx);
    if ((busy.rows[0]?.n ?? 0) >= s.maxConcurrent) return null;
    await trx
      .updateTable("agent_run")
      .set({ status: "running", started_at: new Date() })
      .where("id", "=", runId)
      .execute();
    return "claimed";
  });
}

// ── 预算划拨 ──

interface Limits {
  maxToolCalls: number;
  maxTokens: number | null;
  maxCostUsd: number | null;
  maxDurationSec: number;
}

function limitsOf(run: AgentRunRow, s: GlobalSettings): Limits {
  const b = (run.budget ?? {}) as AgentBudget;
  return {
    maxToolCalls: b.maxToolCalls ?? s.agent.maxToolCalls,
    // token / 费用只认本会话显式划拨的额度;设置里不再有全局上限(2026-10-06)
    maxTokens: b.maxTokens ?? null,
    maxCostUsd: b.maxCostUsd && b.maxCostUsd > 0 ? b.maxCostUsd : null,
    maxDurationSec: b.maxDurationSec ?? s.agent.maxDurationSec,
  };
}

/**
 * 父会话剩余预算 = 上限 − 自身已用 − Σ 子会话占用(已结束按实际用量,未结束按划拨额与实际用量取大)。
 * 子会话预算:请求值(超过剩余直接拒绝)或剩余的一半。
 */
export async function allocateBudget(
  ctx: ServerContext,
  parent: AgentRunRow,
  s: GlobalSettings,
  requested: SubagentBudget | undefined,
): Promise<SubagentBudget> {
  const L = limitsOf(parent, s);
  const own = liveRun(parent.id)?.handle.stats();
  const ownCalls = own?.toolCalls ?? parent.tool_call_count;
  const ownUsage = own?.usage ?? usageOf(parent);
  const kids = await descendants(ctx, parent.id);
  const direct = kids.filter((k) => k.parent_run_id === parent.id);
  let calls = ownCalls;
  let tokens = ownUsage.totalTokens;
  let cost = ownUsage.costUsd;
  for (const k of direct) {
    const sub = [k, ...kids.filter((x) => x.depth > k.depth && isUnder(kids, x, k.id))];
    const actual = sub.reduce(
      (a, r) => ({
        calls: a.calls + r.tool_call_count,
        tokens: a.tokens + usageOf(r).totalTokens,
        cost: a.cost + usageOf(r).costUsd,
      }),
      { calls: 0, tokens: 0, cost: 0 },
    );
    const b = (k.budget ?? {}) as SubagentBudget;
    const done = TERMINAL.includes(k.status);
    calls += done ? actual.calls : Math.max(actual.calls, b.maxToolCalls ?? 0);
    tokens += done ? actual.tokens : Math.max(actual.tokens, b.maxTokens ?? 0);
    cost += done ? actual.cost : Math.max(actual.cost, b.maxCostUsd ?? 0);
  }
  const elapsed = parent.started_at ? (Date.now() - parent.started_at.getTime()) / 1000 : 0;
  const remain = {
    maxToolCalls: Math.floor(L.maxToolCalls - calls),
    maxTokens: L.maxTokens === null ? null : Math.floor(L.maxTokens - tokens),
    maxCostUsd: L.maxCostUsd === null ? null : L.maxCostUsd - cost,
    maxDurationSec: Math.floor(L.maxDurationSec - elapsed),
  };
  const out: SubagentBudget = {};
  const pick = <K extends keyof SubagentBudget>(k: K, left: number | null, label: string, min: number) => {
    const want = requested?.[k];
    if (left === null) {
      if (want !== undefined) out[k] = want;
      return;
    }
    if (left < min) throw new CapabilityError("budget", `本会话剩余${label}不足，不能再派生子会话（剩余 ${left}）`);
    if (want !== undefined && want > left)
      throw new CapabilityError("budget", `子会话${label} ${want} 超过本会话剩余预算 ${left}`);
    const v = want ?? (k === "maxCostUsd" ? left / 2 : Math.max(min, Math.floor(left / 2)));
    out[k] = v as SubagentBudget[K];
  };
  pick("maxToolCalls", remain.maxToolCalls, "工具调用次数", 1);
  pick("maxTokens", remain.maxTokens, " token", 1);
  pick("maxCostUsd", remain.maxCostUsd, "费用", 0.0001);
  pick("maxDurationSec", remain.maxDurationSec, "时长（秒）", 5);
  return out;
}

function isUnder(all: AgentRunRow[], x: AgentRunRow, rootId: string): boolean {
  let cur: AgentRunRow | undefined = x;
  for (let i = 0; cur && i < 16; i++) {
    if (cur.parent_run_id === rootId) return true;
    const pid: string | null = cur.parent_run_id;
    cur = all.find((r) => r.id === pid);
  }
  return false;
}

// ── 能力白名单与模式 ──

export function spawnScope(
  parent: AgentRunRow,
  mode: AgentRunMode,
  requested: readonly string[] | undefined,
): { capabilities: string[]; allowedKinds: CapabilityKind[] | null } {
  const ps = (parent.scope ?? {}) as Partial<RunScope>;
  const parentCaps = ps.capabilities?.length ? ps.capabilities : Object.keys(CAPABILITY_CATALOG);
  const parentKinds = ps.allowedKinds ?? null;
  const kinds = (mode === "plan_only" ? PLAN_ONLY_KINDS : ALL_KINDS).filter(
    (k) => !parentKinds || parentKinds.includes(k),
  );
  const kindOf = (id: string) => CAPABILITY_CATALOG[id as KnownCapabilityId]?.kind;
  if (requested?.length) {
    const outside = requested.filter((id) => !parentCaps.includes(id));
    if (outside.length) throw new CapabilityError("unauthorized", `子会话能力超出本会话白名单：${outside.join(", ")}`);
    const tooStrong = requested.filter((id) => !kinds.includes(kindOf(id) as CapabilityKind));
    if (tooStrong.length)
      throw new CapabilityError("unauthorized", `子会话模式/授权上限不允许这些能力：${tooStrong.join(", ")}`);
  }
  const caps = (requested?.length ? [...new Set(requested)] : parentCaps).filter((id) =>
    kinds.includes(kindOf(id) as CapabilityKind),
  );
  return {
    capabilities: caps,
    allowedKinds: kinds.length === ALL_KINDS.length ? null : kinds,
  };
}

// ── 等待子会话 ──

const sleep = (ms: number, signal?: AbortSignal) =>
  new Promise<void>((resolve) => {
    const t = setTimeout(resolve, ms);
    signal?.addEventListener(
      "abort",
      () => {
        clearTimeout(t);
        resolve();
      },
      { once: true },
    );
  });

function resultOf(r: AgentRunRow): SubagentResult {
  const u = usageOf(r);
  const settled = SETTLED.includes(r.status);
  return {
    runId: r.id,
    goal: safe(r.goal, 300),
    status: r.status,
    summary: settled ? safe(r.summary, 1500) : null,
    toolCalls: r.tool_call_count,
    usage: { totalTokens: u.totalTokens, costUsd: u.costUsd },
    pending: !settled,
  };
}

/** 轮询业务表等子会话结束(不调用模型);父会话取消 / 调用被中止 / 超时即返回当前状态 */
export async function awaitChildren(
  ctx: ServerContext,
  parentId: string,
  ids: readonly string[] | undefined,
  timeoutSec: number,
  signal: AbortSignal,
  pollMs = 500,
): Promise<{ results: SubagentResult[]; allDone: boolean }> {
  const deadline = Date.now() + timeoutSec * 1000;
  for (;;) {
    let q = ctx.db.selectFrom("agent_run").selectAll().where("parent_run_id", "=", parentId);
    if (ids?.length) q = q.where("id", "in", [...ids]);
    const rows = await q.orderBy("created_at").execute();
    if (ids?.length && rows.length !== new Set(ids).size) {
      const missing = ids.filter((id) => !rows.some((r) => r.id === id));
      throw new CapabilityError("not_found", `不是本会话派生的子会话：${missing.join(", ")}`);
    }
    await liveRun(parentId)?.pump();
    const allDone = rows.every((r) => SETTLED.includes(r.status));
    const parent = await ctx.db
      .selectFrom("agent_run")
      .select("cancel_requested_at")
      .where("id", "=", parentId)
      .executeTakeFirst();
    if (allDone || signal.aborted || parent?.cancel_requested_at || Date.now() >= deadline)
      return { results: rows.map(resultOf), allDone };
    await sleep(pollMs, signal);
  }
}

/** runAgent 挂起期间的唤醒条件:等待的子会话全部落定 / 超时 / 本会话被取消 */
export async function awaitReady(ctx: ServerContext, parentId: string, sus: AwaitSuspension): Promise<true | null> {
  if (Date.now() >= Date.parse(sus.deadline)) return true;
  const parent = await ctx.db
    .selectFrom("agent_run")
    .select("cancel_requested_at")
    .where("id", "=", parentId)
    .executeTakeFirst();
  if (!parent || parent.cancel_requested_at) return true;
  let q = ctx.db.selectFrom("agent_run").select("status").where("parent_run_id", "=", parentId);
  if (sus.runIds?.length) q = q.where("id", "in", sus.runIds);
  const rows = await q.execute();
  return rows.every((r) => SETTLED.includes(r.status)) ? true : null;
}

/** 挂起前登记 / 恢复后清除唤醒信号(幂等) */
export async function setAwaitWake(ctx: ServerContext, runId: string, wake: AwaitWake | null): Promise<null> {
  await ctx.db
    .updateTable("agent_run")
    .set({ await_wake: wake ? JSON.stringify(wake) : null })
    .where("id", "=", runId)
    .execute();
  return null;
}

/**
 * 子会话落定(终态或 waiting)后唤醒挂起等待的父会话:按父会话登记的 await_wake 写 fact + 发引擎信号。
 * 父会话没挂起时什么也不做;失败只记日志(父会话还有兜底轮询)。
 */
export async function wakeParent(ctx: ServerContext, childRunId: string): Promise<boolean> {
  try {
    const r = await ctx.db
      .selectFrom("agent_run as c")
      .innerJoin("agent_run as p", "p.id", "c.parent_run_id")
      .select(["p.id as parentId", "p.await_wake as wake"])
      .where("c.id", "=", childRunId)
      .executeTakeFirst();
    const w = r?.wake as AwaitWake | null | undefined;
    if (!w?.engineRunId || !w.fact) return false;
    await ctx.engine.signal(w.engineRunId, w.fact, { childRunId });
    return true;
  } catch (err) {
    ctx.logger.warn({ childRunId, err: { message: (err as Error).message } }, "唤醒父会话失败（父会话会兜底轮询）");
    return false;
  }
}

/** 唤醒挂起中的会话自身(用户选了方案 / 审批已决定):按 await_wake 发引擎信号;没挂起时什么也不做 */
export async function wakeRun(
  ctx: ServerContext,
  runId: string,
  payload: Record<string, string | boolean>,
): Promise<boolean> {
  try {
    const r = await ctx.db.selectFrom("agent_run").select("await_wake").where("id", "=", runId).executeTakeFirst();
    const w = r?.await_wake as AwaitWake | null | undefined;
    if (!w?.engineRunId || !w.fact) return false;
    await ctx.engine.signal(w.engineRunId, w.fact, payload);
    return true;
  } catch (err) {
    ctx.logger.warn({ runId, err: { message: (err as Error).message } }, "唤醒会话失败（会话会兜底轮询）");
    return false;
  }
}

/**
 * 子会话还没落定时请求挂起:记下等待条件并关闭会话(本调用在 JSONL 里保持未完成),drive 看到 requested 后返回
 * suspended。同一批里已有别的调用在挂起时不再重复登记。
 */
async function requestSuspension(live: LiveRun, invocationId: string, runIds: string[] | null, timeoutSec: number) {
  await suspendRun(live, {
    kind: "subagents",
    invocationId,
    runIds,
    deadline: new Date(Date.now() + timeoutSec * 1000).toISOString(),
  });
}

/** 登记挂起条件并关闭会话(挂起的调用在 JSONL 里保持未完成);同一批里已有挂起时不再重复登记 */
export async function suspendRun(live: LiveRun, sus: AwaitSuspension): Promise<void> {
  if (!live.suspend || live.suspend.requested) return;
  live.suspend.requested = sus;
  await new Promise((r) => setTimeout(r, 0));
  await live.handle.close();
}

// ── 能力处理器 ──

const spawnLocks = new Map<string, Promise<void>>();
/** 同一父会话的派生判定串行(进程内);返回释放函数 */
async function spawnLock(parentId: string): Promise<() => void> {
  const prev = spawnLocks.get(parentId) ?? Promise.resolve();
  let release!: () => void;
  const mine = new Promise<void>((r) => {
    release = r;
  });
  const tail = prev.then(() => mine);
  spawnLocks.set(parentId, tail);
  await prev;
  return () => {
    release();
    if (spawnLocks.get(parentId) === tail) spawnLocks.delete(parentId);
  };
}

export interface SpawnDeps {
  /** 建子 agent_run + 任务行并入队 runAgent(service.startRun) */
  start(p: {
    parent: AgentRunRow;
    goal: string;
    mode: AgentRunMode;
    capabilities: string[];
    allowedKinds: CapabilityKind[] | null;
    budget: SubagentBudget;
    upstreamId: string | null;
    reason: string | null;
    dedupeKey: string;
  }): Promise<{ run: AgentRunRow; created: boolean }>;
}

async function parentOf(ctx: ServerContext, c: HandlerContext): Promise<AgentRunRow> {
  const id = c.agentRunId;
  if (!id) throw new CapabilityError("precondition", "只有智能体会话可以派生 / 等待子会话");
  const run = await ctx.db.selectFrom("agent_run").selectAll().where("id", "=", id).executeTakeFirst();
  if (!run) throw new CapabilityError("not_found", "智能体会话不存在");
  return run;
}

export function registerSubagentCapabilities(registry: CapabilityRegistry, ctx: ServerContext, deps: SpawnDeps) {
  // replay=safe:await=true 挂起后会话恢复时要重放本调用取子会话汇报;幂等键相同 → dedupe_key 命中同一个子会话
  registry.register({ ...capabilitySpec("agent.spawn"), replay: "safe" }, async (input, c) => {
    const parent = await parentOf(ctx, c);
    const s = await ctx.settings.all();
    const cfg = subagentSettings(s);
    if (!cfg.enabled) throw new CapabilityError("precondition", "子会话未开启。在设置 → 智能体里打开「允许子会话」");
    if (parent.cancel_requested_at || TERMINAL.includes(parent.status))
      throw new CapabilityError("precondition", "本会话已取消或已结束，不能派生子会话");
    const dedupeKey = `subagent:${c.idempotencyKey ?? `${parent.id}:${Date.now()}`}`;
    const existing = await ctx.db
      .selectFrom("agent_run")
      .selectAll()
      .where("dedupe_key", "=", dedupeKey)
      .executeTakeFirst();
    // AG-4 去重:不是重放(幂等键没命中)时,同目标的子会话还在跑或刚完成就直接复用。
    // 同一父会话的派生串行(同一批并行的两次同目标调用也只建一个);父会话只在一个进程里运行
    const release = await spawnLock(parent.id);
    let child = existing;
    let reused: AgentRunRow | null = null;
    try {
      reused = !child ? await findSameGoalChild(ctx.db, parent.id, input.goal) : null;
      if (reused) child = reused;
      if (!child) child = await create();
    } finally {
      release();
    }
    async function create(): Promise<AgentRunRow> {
      if (parent.depth + 1 > cfg.maxDepth)
        throw new CapabilityError("precondition", `派生层数已达上限 ${cfg.maxDepth}，子会话不能再派生`);
      const mode: AgentRunMode = input.mode ?? parent.mode;
      if (parent.mode === "plan_only" && mode === "execute")
        throw new CapabilityError("unauthorized", "只给方案的会话只能派生只给方案的子会话");
      const scope = spawnScope(parent, mode, input.capabilityAllowlist);
      if (!scope.capabilities.length) throw new CapabilityError("precondition", "子会话没有可用能力");
      const count = await sql<{ n: number }>`
        select count(*)::int as n from agent_run where parent_run_id = ${parent.id}`.execute(ctx.db);
      if ((count.rows[0]?.n ?? 0) >= MAX_CHILDREN_PER_RUN)
        throw new CapabilityError("budget", `一个会话最多派生 ${MAX_CHILDREN_PER_RUN} 个子会话`);
      const budget = await allocateBudget(ctx, parent, s, input.budget);
      const r = await deps.start({
        parent,
        goal: input.goal,
        mode,
        capabilities: scope.capabilities,
        allowedKinds: scope.allowedKinds,
        budget,
        upstreamId: input.upstreamId ?? parent.upstream_id ?? null,
        reason: input.reason ?? null,
        dedupeKey,
      });
      return r.run;
    }
    await liveRun(parent.id)?.pump();
    let result: SubagentResult | null = null;
    if (input.await) {
      const live = liveRun(parent.id);
      const inv = c.invocationId;
      if (live?.suspend && inv) {
        // runAgent 驱动:不在工具调用里等,挂起会话(不占 worker),子会话落定后重放本调用
        const snap = await awaitChildren(ctx, parent.id, [child.id], 0, c.signal);
        if (!snap.allDone && !live.suspend.resumed.has(inv)) {
          // 崩溃对账(reconcile:*)只确保子会话已建,不挂起;真正的等待交给恢复后的重放
          if (!inv.startsWith("reconcile:")) await requestSuspension(live, inv, [child.id], input.timeoutSec);
          throw new AwaitSuspendedError();
        }
        result = snap.results[0] ?? null;
      } else {
        result = (await awaitChildren(ctx, parent.id, [child.id], input.timeoutSec, c.signal)).results[0] ?? null;
      }
    }
    const cs = (child.scope ?? {}) as Partial<RunScope>;
    if (reused) {
      // 没要求等待时也带上已有子会话的当前结果(还在跑的 summary 为 null、pending = true)
      if (!result) {
        const row = await ctx.db.selectFrom("agent_run").selectAll().where("id", "=", child.id).executeTakeFirst();
        result = resultOf(row ?? child);
      }
      return {
        reused: true,
        message: result.pending
          ? "已有同目标子会话，仍在运行，未新建；需要结果时用 agent.await 等它"
          : "已有同目标子会话（10 分钟内已完成），未新建；结果见 result",
        runId: child.id,
        depth: child.depth,
        mode: child.mode,
        capabilities: cs.capabilities ?? [],
        budget: (child.budget ?? {}) as SubagentBudget,
        result,
      };
    }
    return {
      runId: child.id,
      depth: child.depth,
      mode: child.mode,
      capabilities: cs.capabilities ?? [],
      budget: (child.budget ?? {}) as SubagentBudget,
      result,
    };
  });

  registry.register(capabilitySpec("agent.await"), async (input, c) => {
    const parent = await parentOf(ctx, c);
    const live = liveRun(parent.id);
    const inv = c.invocationId;
    if (!live?.suspend || !inv) return awaitChildren(ctx, parent.id, input.runIds, input.timeoutSec, c.signal);
    // 恢复时重放:已挂起等过,直接返回当前状态
    const snap = await awaitChildren(ctx, parent.id, input.runIds, 0, c.signal);
    if (snap.allDone || live.suspend.resumed.has(inv) || live.suspend.requested) return snap;
    // 挂起:立即返回,会话关闭(本调用在 JSONL 里保持未完成,返回值不会进模型上下文);恢复后重放拿真实结果
    await requestSuspension(live, inv, input.runIds?.length ? [...input.runIds] : null, input.timeoutSec);
    return snap;
  });
}

/**
 * 授权上限:子会话的每次授权调用都要落在全部祖先会话的能力白名单与类别上限之内(再与自身范围取交集)。
 * 返回拒绝原因或 null。
 */
/** browser.solve_challenge 子流程用到的能力(与 challenge.ts 的 VISION_CAPABILITIES 一致) */
const VISION_STEP_CAPS = [
  "browser.screenshot",
  "browser.click_at",
  "browser.type_at",
  "browser.scroll",
  "browser.wait",
  "browser.snapshot",
];

export async function exceedsAncestors(
  db: Kysely<DB>,
  runId: string,
  capabilityId: string,
  kind: CapabilityKind,
): Promise<string | null> {
  const chain = await ancestors(db, runId);
  for (const a of chain) {
    const sc = (a.scope ?? {}) as Partial<RunScope>;
    // 视觉子流程(browser.solve_challenge)的截图 / 坐标动作由白名单里的 solve_challenge 覆盖
    const viaChallenge =
      VISION_STEP_CAPS.includes(capabilityId) && !!sc.capabilities?.includes("browser.solve_challenge");
    if (sc.capabilities?.length && !sc.capabilities.includes(capabilityId) && !viaChallenge)
      return `超出父会话 ${a.id} 的能力白名单：${capabilityId}`;
    if (sc.allowedKinds && !sc.allowedKinds.includes(kind)) return `父会话 ${a.id} 不允许 ${kind} 类能力`;
  }
  return null;
}