import { createHash, randomUUID } from "node:crypto";
import * as fs from "node:fs";
import os from "node:os";
import path from "node:path";
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { OperationJournal } from "./operation-journal.js";
import { registerPlanExecRpc } from "./plan-exec-rpc.js";
import { singleChildWorkflowScript } from "./workflow-spawn.js";

// Protocol evidence (installed sources verified against pi-subagents@0.60.0
// and @tintinweb/pi-tasks@0.9.0):
// - @tintinweb/pi-tasks src/index.ts:103-119 reply channel/envelope,
//   126-133 spawn/stop params, 137-157 strict PROTOCOL_VERSION=2,
//   207-260 completed/failed/stopped fields.
// - pi-subagents src/extension/rpc.ts exposes the v1 channel and async-only spawn;
//   public-execution.ts requires workflowScript and rejects clarify entirely.
// - pi-subagents src/runs/background/result-watcher.ts:49-57 result file fields,
//   141-164 child output/status normalization, 193-204 async-complete payload;
//   subagent-runner.ts:3066-3073 complete/failed/paused state values.
// - pi-subagents src/agents/agents.ts:31-40 builtin names and
//   src/agents/agent-selection.ts:4-19 exact-name merge; pi-tasks examples need aliases.
const PING_CHANNEL = "subagents:rpc:ping";
const SPAWN_CHANNEL = "subagents:rpc:spawn";
const STOP_CHANNEL = "subagents:rpc:stop";
const COMPLETED_EVENT = "subagents:completed";
const FAILED_EVENT = "subagents:failed";
const READY_EVENT = "subagents:ready";
const WARNING_EVENT = "subagents:warning";
const NB_REQUEST_CHANNEL = "subagents:rpc:v1:request";
const NB_COMPLETE_EVENT = "subagent:async-complete";
const NB_REPLY_PREFIX = "subagents:rpc:v1:reply:";
const DEFAULT_SPAWN_TIMEOUT_MS = 24_000;
const DEFAULT_COMPLETION_POLL_INTERVAL_MS = 2_000;
const DEFAULT_MAX_ACTIVE_RUNS = 2;
const DEFAULT_MAX_TURNS = 12;
const DEFAULT_TERMINAL_RESULT_GRACE_MS = 5_000;
const DEFAULT_SPAWN_REPLY_CACHE_TTL_MS = 60_000;
const DEFAULT_ACCEPTED_RUN_LEASE_MS = 30_000;
const MAX_FAILURE_PARTIAL_OUTPUT_CHARS = 4_000;

const AGENT_TYPE_ALIASES = new Map<string, string>([
  ["general-purpose", "delegate"],
  ["Explore", "scout"],
  ["explore", "scout"],
]);

const BRIDGE_ACCEPTANCE_CONFIG = {
  level: "none",
  reason:
    "pi-tasks bridge manages task lifecycle and result propagation; do not require pi-subagents acceptance reports.",
} as const;

const BRIDGE_CONTROL_CONFIG = {
  enabled: false,
} as const;

interface BridgeOptions {
  spawnTimeoutMs?: number;
  planExecJournalPath?: string;
  completionPollIntervalMs?: number;
  maxActiveRuns?: number;
  defaultMaxTurns?: number;
  terminalResultGraceMs?: number;
  spawnReplyCacheTtlMs?: number;
  acceptedRunLeaseMs?: number;
  acceptedRunReconcileIntervalMs?: number;
  getSessionId?: () => string | undefined;
  operationJournal?: OperationJournal;
}

type BridgeHost = Pick<ExtensionAPI, "events">;
type Unsubscribe = () => void;
type RpcReply<T> =
  { success: true; data: T } | { success: false; error: string };
type CompletionKind = "completed" | "failed" | "stopped";

interface BridgeRegistration {
  dispose(): void;
}

interface SpawnReplyCacheEntry {
  identity: string;
  reply: RpcReply<{ id: string }>;
  expiresAt: number;
}

interface InFlightSpawnReply {
  identity: string;
  reply: Promise<RpcReply<{ id: string }>>;
}

interface PendingLegacyBinding {
  requestId: string;
  requestDigest: string;
  sessionId: string;
  runId: string;
}

interface BridgeState {
  ownedRunIds: Set<string>;
  completedRunIds: Set<string>;
  stoppingRunIds: Set<string>;
  pendingSpawnCount: number;
  inFlightSpawnReplies: Map<string, InFlightSpawnReply>;
  spawnReplyCache: Map<string, SpawnReplyCacheEntry>;
  terminalResultDeadlines: Map<string, number>;
  pendingLegacyBindings: Map<string, PendingLegacyBinding>;
  journal?: OperationJournal;
  registration?: BridgeRegistration;
}

const bridgeStates = new WeakMap<BridgeHost["events"], BridgeState>();
const PROCESS_OWNER_KEY = Symbol.for(
  "pi-subagents-bridge.accepted-run-owner.v1",
);

interface ProcessOwnerStore {
  pid: number;
  instanceId: string;
}

function acceptedRunOwner(): ProcessOwnerStore {
  const target = globalThis as typeof globalThis & {
    [PROCESS_OWNER_KEY]?: ProcessOwnerStore;
  };
  const existing = target[PROCESS_OWNER_KEY];
  if (existing?.pid === process.pid) return existing;
  const created = { pid: process.pid, instanceId: randomUUID() };
  target[PROCESS_OWNER_KEY] = created;
  return created;
}

interface SpawnOptionsRaw {
  model?: unknown;
  maxTurns?: unknown;
}

interface AsyncCompleteRaw {
  runId?: unknown;
  mode?: unknown;
  asyncId?: unknown;
  id?: unknown;
  success?: unknown;
  state?: unknown;
  summary?: unknown;
  output?: unknown;
  error?: unknown;
  results?: unknown;
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}

function text(value: unknown): string | undefined {
  return typeof value === "string" && value.trim().length > 0
    ? value.trim()
    : undefined;
}

function replyChannel(channel: string, requestId: string): string {
  return `${channel}:reply:${requestId}`;
}

function nbReplyChannel(requestId: string): string {
  return `${NB_REPLY_PREFIX}${requestId}`;
}

function emitReply<T>(
  events: BridgeHost["events"],
  channel: string,
  requestId: string,
  reply: RpcReply<T>,
): void {
  events.emit(replyChannel(channel, requestId), reply);
}

function extractSpawnRunId(reply: unknown): string | undefined {
  if (!isRecord(reply)) return undefined;

  const details = isRecord(reply.details) ? reply.details : undefined;
  return (
    text(details?.runId) ??
    text(details?.asyncId) ??
    text(reply.runId) ??
    text(reply.asyncId)
  );
}

function extractRunId(payload: AsyncCompleteRaw): string | undefined {
  return text(payload.runId) ?? text(payload.asyncId) ?? text(payload.id);
}

function extractChildOutputs(payload: AsyncCompleteRaw): string[] {
  if (!Array.isArray(payload.results)) return [];

  const outputs: string[] = [];
  for (const result of payload.results) {
    if (!isRecord(result)) continue;
    const output = text(result.output) ?? text(result.error);
    if (output) outputs.push(output);
  }
  return outputs;
}

function extractCompletedResult(payload: AsyncCompleteRaw): string | undefined {
  const childOutput = extractChildOutputs(payload).join("\n\n") || undefined;
  if (text(payload.mode) === "workflow" && childOutput) return childOutput;
  return text(payload.summary) ?? text(payload.output) ?? childOutput;
}

function extractStoppedResult(payload: AsyncCompleteRaw): string | undefined {
  return (
    (extractChildOutputs(payload).join("\n\n") || undefined) ??
    text(payload.output) ??
    text(payload.summary)
  );
}

function extractFailureError(payload: AsyncCompleteRaw): string {
  let error = text(payload.error);
  if (!error && Array.isArray(payload.results)) {
    for (const result of payload.results) {
      if (!isRecord(result)) continue;
      error = text(result.error);
      if (error) break;
    }
  }

  const partialOutput =
    text(payload.output) ??
    (Array.isArray(payload.results)
      ? payload.results
          .filter(isRecord)
          .map((result) => text(result.output))
          .filter((output): output is string => output !== undefined)
          .join("\n\n")
      : undefined);
  if (!partialOutput) return error ?? "Agent failed";

  const truncatedOutput = partialOutput.slice(
    0,
    MAX_FAILURE_PARTIAL_OUTPUT_CHARS,
  );
  const suffix =
    partialOutput.length > truncatedOutput.length ? "\n[truncated]" : "";
  return `${error ?? "Agent failed"}\n\nPartial output:\n${truncatedOutput}${suffix}`;
}

function classifyCompleteEvent(payload: AsyncCompleteRaw): CompletionKind {
  const state = text(payload.state);
  if (state === "paused" || state === "stopped") return "stopped";
  if (state === "failed" || state === "aborted" || payload.success === false)
    return "failed";
  if (state === "complete" || payload.success === true) return "completed";
  return "failed";
}

function resolveAgentType(type: string): string {
  return AGENT_TYPE_ALIASES.get(type) ?? type;
}

function requestNicobailonRpc<T>(
  events: BridgeHost["events"],
  method: "spawn" | "status",
  params: Record<string, unknown>,
  timeoutMs: number,
  signal?: AbortSignal,
): Promise<T> {
  const requestId = randomUUID();

  return new Promise<T>((resolve, reject) => {
    let settled = false;

    const cleanup = (): void => {
      if (settled) return;
      settled = true;
      unsubscribe();
      clearTimeout(timeout);
      signal?.removeEventListener("abort", onAbort);
    };

    const fail = (error: Error): void => {
      if (settled) return;
      cleanup();
      reject(error);
    };

    const succeed = (value: T): void => {
      if (settled) return;
      cleanup();
      resolve(value);
    };

    const onAbort = (): void => {
      fail(new Error("Bridge disposed"));
    };

    const timeout = setTimeout(() => {
      fail(
        new Error(`nicobailon "${method}" RPC timed out after ${timeoutMs}ms`),
      );
    }, timeoutMs);

    const unsubscribe = events.on(nbReplyChannel(requestId), (raw: unknown) => {
      if (!isRecord(raw) || typeof raw.success !== "boolean") {
        fail(new Error("Malformed nicobailon RPC reply."));
        return;
      }

      if (raw.success) {
        succeed(raw.data as T);
        return;
      }

      const message = isRecord(raw.error)
        ? text(raw.error.message)
        : text(raw.error);
      fail(new Error(message ?? "nicobailon RPC error"));
    });

    signal?.addEventListener("abort", onAbort, { once: true });
    if (signal?.aborted) {
      onAbort();
      return;
    }

    events.emit(NB_REQUEST_CHANNEL, {
      version: 1,
      requestId,
      method,
      params,
    });
  });
}

function normalizeSpawnOptions(raw: unknown): SpawnOptionsRaw | undefined {
  return isRecord(raw) ? raw : undefined;
}

function extractRpcText(reply: unknown): string | undefined {
  return isRecord(reply) ? text(reply.text) : undefined;
}

function parseStatusState(statusText: string): string | undefined {
  const match = /^State:\s+(.+)$/im.exec(statusText);
  return match?.[1]?.trim().toLowerCase();
}

function parseResultPath(statusText: string): string | undefined {
  const match = /^Result:\s+(.+)$/im.exec(statusText);
  return match?.[1]?.trim();
}

function classifyStatusText(statusText: string): CompletionKind | undefined {
  const state = parseStatusState(statusText);
  if (!state) return undefined;
  if (state === "paused" || state === "stopped") return "stopped";
  if (state === "failed" || state === "aborted") return "failed";
  if (state === "complete") return "completed";
  return undefined;
}

function readResultPayload(resultPath: string | undefined): {
  payload?: AsyncCompleteRaw;
  error?: string;
} {
  if (!resultPath) return {};
  try {
    const parsed: unknown = JSON.parse(fs.readFileSync(resultPath, "utf8"));
    if (!isRecord(parsed)) {
      return { error: "result payload is not a JSON object" };
    }
    return { payload: parsed };
  } catch {
    return { error: "result payload is not readable yet" };
  }
}

function positiveIntegerOrDefault(
  value: number | undefined,
  fallback: number,
): number {
  return value !== undefined && Number.isInteger(value) && value > 0
    ? value
    : fallback;
}

function getBridgeState(events: BridgeHost["events"]): BridgeState {
  const existing = bridgeStates.get(events);
  if (existing) return existing;

  const created: BridgeState = {
    ownedRunIds: new Set<string>(),
    completedRunIds: new Set<string>(),
    stoppingRunIds: new Set<string>(),
    pendingSpawnCount: 0,
    inFlightSpawnReplies: new Map(),
    spawnReplyCache: new Map(),
    terminalResultDeadlines: new Map(),
    pendingLegacyBindings: new Map(),
  };
  bridgeStates.set(events, created);
  return created;
}

export function registerBridge(
  pi: BridgeHost,
  options: BridgeOptions = {},
): { dispose: () => void } {
  const state = getBridgeState(pi.events);
  if (state.registration) return state.registration;
  const processOwner = acceptedRunOwner();

  const spawnTimeoutMs = positiveIntegerOrDefault(
    options.spawnTimeoutMs,
    DEFAULT_SPAWN_TIMEOUT_MS,
  );
  const completionPollIntervalMs = positiveIntegerOrDefault(
    options.completionPollIntervalMs,
    DEFAULT_COMPLETION_POLL_INTERVAL_MS,
  );
  const maxActiveRuns = positiveIntegerOrDefault(
    options.maxActiveRuns,
    DEFAULT_MAX_ACTIVE_RUNS,
  );
  const defaultMaxTurns = positiveIntegerOrDefault(
    options.defaultMaxTurns,
    DEFAULT_MAX_TURNS,
  );
  const terminalResultGraceMs = positiveIntegerOrDefault(
    options.terminalResultGraceMs,
    DEFAULT_TERMINAL_RESULT_GRACE_MS,
  );
  const spawnReplyCacheTtlMs = positiveIntegerOrDefault(
    options.spawnReplyCacheTtlMs,
    DEFAULT_SPAWN_REPLY_CACHE_TTL_MS,
  );
  const acceptedRunLeaseMs = positiveIntegerOrDefault(
    options.acceptedRunLeaseMs,
    DEFAULT_ACCEPTED_RUN_LEASE_MS,
  );
  const fallbackSessionId = `process:${processOwner.instanceId}`;
  const currentSessionId = (): string | undefined => {
    const sessionId = options.getSessionId?.()?.trim();
    if (sessionId) return sessionId;
    return options.getSessionId ? undefined : fallbackSessionId;
  };

  const {
    ownedRunIds,
    completedRunIds,
    stoppingRunIds,
    inFlightSpawnReplies,
    spawnReplyCache,
    terminalResultDeadlines,
    pendingLegacyBindings,
  } = state;
  const pendingRpcControllers = new Set<AbortController>();
  const completionPollTimers = new Map<string, ReturnType<typeof setTimeout>>();
  const volatileOwnedRuns = new Map<string, string>();
  const unsubscribes: Unsubscribe[] = [];
  let disposed = false;
  if (options.operationJournal) {
    if (state.journal && state.journal !== options.operationJournal) {
      throw new Error(
        "bridge was already registered with a different operation journal",
      );
    }
    state.journal = options.operationJournal;
  } else if (options.planExecJournalPath) {
    const journalPath = path.resolve(options.planExecJournalPath);
    if (state.journal && state.journal.filePath !== journalPath) {
      throw new Error(
        "bridge was already registered with a different operation journal",
      );
    }
    state.journal ??= new OperationJournal(journalPath);
  }
  const bridgeJournal = state.journal;
  const planExecRpc = registerPlanExecRpc(pi.events, {
    timeoutMs: spawnTimeoutMs,
    ...(options.planExecJournalPath
      ? { journalPath: options.planExecJournalPath }
      : {}),
  });

  const track = (unsubscribe: Unsubscribe | void): void => {
    if (typeof unsubscribe === "function") unsubscribes.push(unsubscribe);
  };

  const clearCompletionPoll = (runId: string): void => {
    const timer = completionPollTimers.get(runId);
    if (timer) clearTimeout(timer);
    completionPollTimers.delete(runId);
  };

  const shouldStopPolling = (runId: string): boolean =>
    disposed || completedRunIds.has(runId) || !ownedRunIds.has(runId);

  const emitCompletion = (
    runId: string,
    kind: CompletionKind,
    payload?: AsyncCompleteRaw,
  ): void => {
    if (!ownedRunIds.has(runId) || completedRunIds.has(runId)) return;
    const sessionId = currentSessionId();
    if (!sessionId) return;
    if (bridgeJournal && volatileOwnedRuns.get(runId) !== sessionId) {
      let renewed: boolean;
      try {
        renewed = bridgeJournal.renewAcceptedRun(
          runId,
          processOwner.instanceId,
          sessionId,
        );
      } catch (error: unknown) {
        console.error(`Failed to renew accepted bridge run '${runId}':`, error);
        return;
      }
      if (!renewed) {
        ownedRunIds.delete(runId);
        clearCompletionPoll(runId);
        return;
      }
    }

    completedRunIds.add(runId);
    ownedRunIds.delete(runId);
    stoppingRunIds.delete(runId);
    terminalResultDeadlines.delete(runId);
    clearCompletionPoll(runId);

    if (kind === "stopped") {
      const result = payload ? extractStoppedResult(payload) : undefined;
      pi.events.emit(FAILED_EVENT, {
        id: runId,
        ...(result ? { result } : {}),
        status: "stopped",
      });
    } else if (kind === "failed") {
      pi.events.emit(FAILED_EVENT, {
        id: runId,
        error: payload ? extractFailureError(payload) : "Agent failed",
        status: "failed",
      });
    } else {
      const result = payload ? extractCompletedResult(payload) : undefined;
      pi.events.emit(COMPLETED_EVENT, {
        id: runId,
        ...(result ? { result } : {}),
      });
    }

    try {
      bridgeJournal?.completeRun(
        runId,
        processOwner.instanceId,
        sessionId,
      );
      volatileOwnedRuns.delete(runId);
    } catch (error: unknown) {
      console.error(
        `Failed to record delivered bridge completion for '${runId}':`,
        error,
      );
    }
  };

  const pollRunCompletion = async (runId: string): Promise<void> => {
    completionPollTimers.delete(runId);
    if (shouldStopPolling(runId)) {
      clearCompletionPoll(runId);
      return;
    }

    const controller = new AbortController();
    pendingRpcControllers.add(controller);

    try {
      const reply = await requestNicobailonRpc<unknown>(
        pi.events,
        "status",
        { id: runId },
        spawnTimeoutMs,
        controller.signal,
      );
      if (shouldStopPolling(runId)) {
        clearCompletionPoll(runId);
        return;
      }

      const statusText = extractRpcText(reply);
      const kind = statusText ? classifyStatusText(statusText) : undefined;
      if (kind && statusText) {
        const resultPath = parseResultPath(statusText);
        const result = readResultPayload(resultPath);
        if (result.payload) {
          terminalResultDeadlines.delete(runId);
          emitCompletion(runId, kind, result.payload);
          return;
        }

        if (resultPath) {
          const now = Date.now();
          const deadline =
            terminalResultDeadlines.get(runId) ?? now + terminalResultGraceMs;
          terminalResultDeadlines.set(runId, deadline);
          if (now < deadline) {
            const timer = setTimeout(
              () => {
                void pollRunCompletion(runId);
              },
              Math.min(completionPollIntervalMs, deadline - now),
            );
            timer.unref();
            completionPollTimers.set(runId, timer);
            return;
          }

          emitCompletion(runId, kind, {
            [kind === "failed" ? "error" : "summary"]:
              `Bridge warning: ${result.error ?? "result payload was unavailable"} after ${terminalResultGraceMs}ms.`,
          });
          return;
        }

        emitCompletion(runId, kind);
        return;
      }
    } catch {
      if (disposed) return;
    } finally {
      pendingRpcControllers.delete(controller);
    }

    if (shouldStopPolling(runId)) {
      clearCompletionPoll(runId);
      return;
    }

    const timer = setTimeout(() => {
      void pollRunCompletion(runId);
    }, completionPollIntervalMs);
    timer.unref();
    completionPollTimers.set(runId, timer);
  };

  const ensureCompletionPoll = (runId: string): void => {
    if (completionPollTimers.has(runId)) return;
    const timer = setTimeout(() => {
      void pollRunCompletion(runId);
    }, completionPollIntervalMs);
    timer.unref();
    completionPollTimers.set(runId, timer);
  };

  const rearmAcceptedRun = (runId: string): string | undefined => {
    const sessionId = currentSessionId();
    if (!sessionId) return "Pi session identity is not initialized";
    if (bridgeJournal && volatileOwnedRuns.get(runId) !== sessionId) {
      try {
        if (!bridgeJournal.acceptRun(runId, processOwner, sessionId)) {
          const claimed = bridgeJournal.claimAcceptedRuns(
            processOwner,
            sessionId,
            acceptedRunLeaseMs,
          );
          if (!claimed.some((run) => run.runId === runId)) {
            return `native run '${runId}' is already owned by another bridge session`;
          }
        }
      } catch (error: unknown) {
        return error instanceof Error ? error.message : String(error);
      }
    }
    completedRunIds.delete(runId);
    ownedRunIds.add(runId);
    ensureCompletionPoll(runId);
    return undefined;
  };

  const reconcileAcceptedRuns = (): void => {
    if (!bridgeJournal || disposed) return;
    const sessionId = currentSessionId();
    if (!sessionId) return;
    for (const [runId, pending] of pendingLegacyBindings) {
      if (pending.sessionId !== sessionId) continue;
      try {
        bridgeJournal.bindLegacySpawn(
          pending.requestId,
          pending.requestDigest,
          pending.sessionId,
          pending.runId,
        );
        pendingLegacyBindings.delete(runId);
      } catch (error: unknown) {
        console.error(
          `Failed to retry legacy bridge run '${runId}' binding:`,
          error,
        );
      }
    }
    try {
      const claimed = bridgeJournal.claimAcceptedRuns(
        processOwner,
        sessionId,
        acceptedRunLeaseMs,
      );
      const claimedIds = new Set(claimed.map((run) => run.runId));
      for (const runId of [...ownedRunIds]) {
        const volatileSessionId = volatileOwnedRuns.get(runId);
        if (volatileSessionId === sessionId) {
          try {
            if (bridgeJournal.acceptRun(runId, processOwner, sessionId)) {
              volatileOwnedRuns.delete(runId);
              claimedIds.add(runId);
              continue;
            }
            volatileOwnedRuns.delete(runId);
          } catch (error: unknown) {
            console.error(
              `Failed to persist volatile bridge run '${runId}':`,
              error,
            );
            continue;
          }
        } else if (volatileSessionId) {
          volatileOwnedRuns.delete(runId);
        } else if (claimedIds.has(runId)) {
          continue;
        }
        ownedRunIds.delete(runId);
        clearCompletionPoll(runId);
        pi.events.emit(WARNING_EVENT, {
          code: "accepted_run_ownership_lost",
          id: runId,
          message:
            "Bridge stopped delivering this run because another session owns its completion record.",
        });
      }
      for (const accepted of claimed) {
        ownedRunIds.add(accepted.runId);
        ensureCompletionPoll(accepted.runId);
      }
    } catch (error: unknown) {
      console.error("Failed to reconcile accepted bridge runs:", error);
    }
  };

  reconcileAcceptedRuns();
  const acceptedRunReconcileInterval = bridgeJournal
    ? setInterval(
        reconcileAcceptedRuns,
        positiveIntegerOrDefault(
          options.acceptedRunReconcileIntervalMs,
          Math.max(
            1_000,
            Math.min(
              completionPollIntervalMs,
              Math.floor(acceptedRunLeaseMs / 3),
            ),
          ),
        ),
      )
    : undefined;
  acceptedRunReconcileInterval?.unref();

  // pi-subagents/src/agents/agents.ts exposes exact runtime names and has no
  // general-purpose/Explore builtins. Keep only these pi-tasks compatibility aliases.
  track(
    pi.events.on(PING_CHANNEL, (raw: unknown) => {
      if (!isRecord(raw)) return;
      const requestId = text(raw.requestId);
      if (!requestId) return;

      emitReply(pi.events, PING_CHANNEL, requestId, {
        success: true,
        data: { version: 2 },
      } satisfies RpcReply<{ version: number }>);
    }),
  );

  const pruneSpawnReplyCache = (): void => {
    const now = Date.now();
    for (const [requestId, entry] of spawnReplyCache) {
      if (entry.expiresAt <= now) spawnReplyCache.delete(requestId);
    }
  };

  const handleSpawn = (raw: unknown): void => {
    if (!isRecord(raw)) return;
    const requestId = text(raw.requestId);
    if (!requestId) return;

    const agentType = text(raw.type);
    const prompt = text(raw.prompt);
    if (!agentType || !prompt) {
      emitReply(pi.events, SPAWN_CHANNEL, requestId, {
        success: false,
        error: "spawn requires string type and prompt",
      });
      return;
    }
    const sessionId = currentSessionId();
    if (!sessionId) {
      emitReply(pi.events, SPAWN_CHANNEL, requestId, {
        success: false,
        error: "Pi session identity is not initialized",
      });
      return;
    }
    const optionsRaw = normalizeSpawnOptions(raw.options);
    const model = text(optionsRaw?.model);
    const maxTurns =
      typeof optionsRaw?.maxTurns === "number" &&
      Number.isInteger(optionsRaw.maxTurns) &&
      optionsRaw.maxTurns > 0
        ? optionsRaw.maxTurns
        : defaultMaxTurns;
    const spawnParams: Record<string, unknown> = {
      workflowScript: singleChildWorkflowScript(
        resolveAgentType(agentType),
        prompt,
        { control: BRIDGE_CONTROL_CONFIG },
      ),
      async: true,
      context: "fresh",
      acceptance: BRIDGE_ACCEPTANCE_CONFIG,
      control: BRIDGE_CONTROL_CONFIG,
      ...(model ? { model } : {}),
      turnBudget: { maxTurns },
    };
    const legacyRequestDigest = `sha256:${createHash("sha256")
      .update(JSON.stringify({ agentType, prompt, spawnParams }))
      .digest("hex")}`;
    const requestIdentity = `${sessionId}:${legacyRequestDigest}`;

    pruneSpawnReplyCache();
    const cached = spawnReplyCache.get(requestId);
    if (cached) {
      if (cached.identity !== requestIdentity) {
        emitReply(pi.events, SPAWN_CHANNEL, requestId, {
          success: false,
          error: "spawn requestId was already used by another request",
        });
        return;
      }
      if (cached.reply.success) {
        const error = rearmAcceptedRun(cached.reply.data.id);
        if (error) {
          emitReply(pi.events, SPAWN_CHANNEL, requestId, {
            success: false,
            error,
          });
          return;
        }
      }
      emitReply(pi.events, SPAWN_CHANNEL, requestId, cached.reply);
      return;
    }

    const inFlight = inFlightSpawnReplies.get(requestId);
    if (inFlight) {
      if (inFlight.identity !== requestIdentity) {
        emitReply(pi.events, SPAWN_CHANNEL, requestId, {
          success: false,
          error: "spawn requestId was already used by another request",
        });
        return;
      }
      void inFlight.reply.then((reply) => {
        if (!disposed) emitReply(pi.events, SPAWN_CHANNEL, requestId, reply);
      });
      return;
    }

    const spawn = async (): Promise<RpcReply<{ id: string }>> => {
      if (bridgeJournal) {
        try {
          const durable = bridgeJournal.getLegacySpawn(requestId);
          if (durable) {
            if (
              durable.requestDigest !== legacyRequestDigest ||
              durable.sessionId !== sessionId
            ) {
              return {
                success: false,
                error: "spawn requestId was already used by another request",
              };
            }
            if (durable.binding === "bound" && durable.runId) {
              const error = rearmAcceptedRun(durable.runId);
              return error
                ? { success: false, error }
                : { success: true, data: { id: durable.runId } };
            }
            return {
              success: false,
              error:
                durable.error ??
                "legacy spawn outcome is unknown; manual recovery is required",
            };
          }
        } catch (error: unknown) {
          return {
            success: false,
            error: error instanceof Error ? error.message : String(error),
          };
        }
      }

      if (ownedRunIds.size + state.pendingSpawnCount >= maxActiveRuns) {
        return {
          success: false,
          error: `bridge capacity reached: at most ${maxActiveRuns} active runs`,
        };
      }

      let durableOperationStarted = false;
      if (bridgeJournal) {
        try {
          const begun = bridgeJournal.beginLegacySpawn(
            requestId,
            legacyRequestDigest,
            sessionId,
          );
          if (!begun.created) {
            return {
              success: false,
              error: "spawn requestId was concurrently claimed",
            };
          }
          durableOperationStarted = true;
        } catch (error: unknown) {
          return {
            success: false,
            error: error instanceof Error ? error.message : String(error),
          };
        }
      }

      const controller = new AbortController();
      state.pendingSpawnCount += 1;
      pendingRpcControllers.add(controller);

      try {
        const reply = await requestNicobailonRpc<unknown>(
          pi.events,
          "spawn",
          spawnParams,
          spawnTimeoutMs,
          controller.signal,
        );
        if (disposed || controller.signal.aborted) {
          throw new Error("Bridge disposed after native dispatch");
        }

        const runId = extractSpawnRunId(reply);
        if (!runId) {
          throw new Error("nicobailon spawn reply did not include a run id");
        }

        try {
          bridgeJournal?.bindLegacySpawn(
            requestId,
            legacyRequestDigest,
            sessionId,
            runId,
          );
          pendingLegacyBindings.delete(runId);
          durableOperationStarted = false;
        } catch (error: unknown) {
          pendingLegacyBindings.set(runId, {
            requestId,
            requestDigest: legacyRequestDigest,
            sessionId,
            runId,
          });
          console.error(
            `Failed to persist legacy bridge run '${runId}' binding:`,
            error,
          );
        }

        try {
          const accepted = bridgeJournal?.acceptRun(
            runId,
            processOwner,
            sessionId,
          );
          if (accepted === false) {
            return {
              success: false,
              error: `native run '${runId}' is already owned by another bridge session`,
            };
          }
        } catch (error: unknown) {
          // The native run already exists. Returning a failed spawn would invite
          // TaskExecute to launch a duplicate, so keep session-scoped ownership
          // in memory and report the durability loss explicitly.
          volatileOwnedRuns.set(runId, sessionId);
          console.error(`Failed to persist accepted bridge run '${runId}':`, error);
        }
        ownedRunIds.add(runId);
        ensureCompletionPoll(runId);
        return { success: true, data: { id: runId } };
      } catch (error: unknown) {
        if (durableOperationStarted) {
          try {
            bridgeJournal?.markLegacySpawnUnknown(
              requestId,
              legacyRequestDigest,
              sessionId,
              error instanceof Error ? error.message : String(error),
            );
          } catch (journalError: unknown) {
            console.error(
              `Failed to persist unknown legacy spawn '${requestId}':`,
              journalError,
            );
          }
        }
        return {
          success: false,
          error: error instanceof Error ? error.message : String(error),
        };
      } finally {
        state.pendingSpawnCount -= 1;
        pendingRpcControllers.delete(controller);
      }
    };

    const spawnReply = spawn();
    const inFlightEntry: InFlightSpawnReply = {
      identity: requestIdentity,
      reply: spawnReply,
    };
    inFlightSpawnReplies.set(requestId, inFlightEntry);
    void spawnReply.then((reply) => {
      if (inFlightSpawnReplies.get(requestId) === inFlightEntry) {
        inFlightSpawnReplies.delete(requestId);
      }
      if (
        reply.success ||
        !reply.error.startsWith("bridge capacity reached:")
      ) {
        spawnReplyCache.set(requestId, {
          identity: requestIdentity,
          reply,
          expiresAt: Date.now() + spawnReplyCacheTtlMs,
        });
      }
      if (!disposed) emitReply(pi.events, SPAWN_CHANNEL, requestId, reply);
    });
  };

  track(pi.events.on(SPAWN_CHANNEL, handleSpawn));

  track(
    pi.events.on(STOP_CHANNEL, (raw: unknown) => {
      if (!isRecord(raw)) return;
      const requestId = text(raw.requestId);
      if (!requestId) return;

      const agentId = text(raw.agentId) ?? text(raw.id) ?? text(raw.runId);
      if (agentId && ownedRunIds.has(agentId) && !stoppingRunIds.has(agentId)) {
        stoppingRunIds.add(agentId);
        try {
          // pi-subagents rpc.ts normalizes stop targets from id/runId; use canonical id here.
          pi.events.emit(NB_REQUEST_CHANNEL, {
            version: 1,
            requestId: randomUUID(),
            method: "stop",
            params: { id: agentId },
          });
        } catch {
          // pi-tasks ignores stop failures; still acknowledge success locally.
        }
      }

      emitReply(pi.events, STOP_CHANNEL, requestId, {
        success: true,
        data: undefined,
      } satisfies RpcReply<void>);
    }),
  );

  track(
    pi.events.on(NB_COMPLETE_EVENT, (raw: unknown) => {
      if (!isRecord(raw)) return;
      const payload = raw as AsyncCompleteRaw;
      const runId = extractRunId(payload);
      if (!runId || !ownedRunIds.has(runId) || completedRunIds.has(runId))
        return;

      emitCompletion(runId, classifyCompleteEvent(payload), payload);
    }),
  );

  for (const runId of ownedRunIds) ensureCompletionPoll(runId);

  pi.events.emit(READY_EVENT, {});

  const registration: BridgeRegistration = {
    dispose() {
      if (disposed) return;
      disposed = true;
      planExecRpc.dispose();
      if (state.registration === registration) {
        delete state.registration;
      }

      for (const controller of pendingRpcControllers) {
        controller.abort();
      }
      pendingRpcControllers.clear();

      for (const timer of completionPollTimers.values()) {
        clearTimeout(timer);
      }
      completionPollTimers.clear();
      if (acceptedRunReconcileInterval) {
        clearInterval(acceptedRunReconcileInterval);
      }

      while (unsubscribes.length > 0) {
        const unsubscribe = unsubscribes.pop();
        try {
          unsubscribe?.();
        } catch {
          // Best effort cleanup.
        }
      }
    },
  };
  state.registration = registration;
  return registration;
}

export default function bridgeExtension(pi: ExtensionAPI): void {
  let sessionId: string | undefined;
  let registration: BridgeRegistration | undefined;
  const register = (): void => {
    registration ??= registerBridge(pi, {
      planExecJournalPath: path.join(
        os.homedir(),
        ".pi",
        "pi-subagents-bridge",
        "plan-exec-operations.sqlite",
      ),
      getSessionId: () => sessionId,
    });
  };

  pi.on("session_start", (_event, ctx) => {
    sessionId = ctx.sessionManager.getSessionId();
    register();
  });
  pi.on("session_shutdown", () => {
    registration?.dispose();
    registration = undefined;
    sessionId = undefined;
  });
  register();
}