import {
  type AgentRunHandle,
  CapabilityError,
  type CapabilityRegistry,
  type ExternalUsage,
  type HandlerContext,
  redactString,
  type SubagentEventBody,
} from "@umt/agent";
import {
  type AgentBudget,
  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:等子会话结束(轮询业务表,不调用模型),返回脱敏后的汇报摘要。
// - 事件:父会话运行进程把子会话事件按父 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;
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);

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

export interface LiveRun {
  handle: AgentRunHandle;
  /** 把子会话的新事件 / 新派生 / 结束镜像进本会话事件流(串行) */
  pump(): Promise<void>;
  /** 子孙会话已用量(同步读缓存,pump 刷新) */
  externalUsage(): ExternalUsage;
}

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,
): 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 };
  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,
  };
}

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

/** 全部子孙会话(不含自身) */
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,
    maxTokens: b.maxTokens ?? s.agent.maxTokensPerRun ?? null,
    maxCostUsd:
      b.maxCostUsd ?? (s.agent.maxCostUsdPerRun && s.agent.maxCostUsdPerRun > 0 ? s.agent.maxCostUsdPerRun : 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);
  }
}

// ── 能力处理器 ──

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", "只有 Agent 会话可以派生 / 等待子会话");
  const run = await ctx.db.selectFrom("agent_run").selectAll().where("id", "=", id).executeTakeFirst();
  if (!run) throw new CapabilityError("not_found", "Agent 会话不存在");
  return run;
}

export function registerSubagentCapabilities(registry: CapabilityRegistry, ctx: ServerContext, deps: SpawnDeps) {
  registry.register(capabilitySpec("agent.spawn"), 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", "子代理未开启(设置 → Agent → 允许子代理)");
    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();
    let child = existing;
    if (!child) {
      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,
      });
      child = r.run;
    }
    await liveRun(parent.id)?.pump();
    const result = input.await
      ? ((await awaitChildren(ctx, parent.id, [child.id], input.timeoutSec, c.signal)).results[0] ?? null)
      : null;
    const cs = (child.scope ?? {}) as Partial<RunScope>;
    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);
    return awaitChildren(ctx, parent.id, input.runIds, input.timeoutSec, c.signal);
  });
}

/**
 * 授权上限:子会话的每次授权调用都要落在全部祖先会话的能力白名单与类别上限之内(再与自身范围取交集)。
 * 返回拒绝原因或 null。
 */
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>;
    if (sc.capabilities?.length && !sc.capabilities.includes(capabilityId))
      return `超出父会话 ${a.id} 的能力白名单:${capabilityId}`;
    if (sc.allowedKinds && !sc.allowedKinds.includes(kind)) return `父会话 ${a.id} 不允许 ${kind} 类能力`;
  }
  return null;
}