import { basename, dirname } from "node:path";
import { fileURLToPath } from "node:url";
import type { AgentMessage } from "@earendil-works/pi-agent-core";
import {
  SessionManager,
  type ExtensionAPI,
  type ExtensionContext,
  type SessionStartEvent,
} from "@earendil-works/pi-coding-agent";
import {
  AnonNotificationClient,
  type InboundAgentCommand,
  type NotificationPayload,
  type NotificationContextUsage,
  type NotifyStatus,
} from "./client.ts";
import { CommandJournal } from "./command-journal.ts";
import { commandLeaseUsable, commandMarker, deliverCommandBatch } from "./command-polling.ts";
import { loadConfig, renderSource, type AnonNotifyConfig, type LoadedConfig } from "./config.ts";
import { classifyRunOutcome } from "./lifecycle.ts";
import {
  createPresenceRuntimeId,
  presenceEnabled,
  PresenceReporter,
  type PresenceEligibility,
  type PresenceSnapshot,
} from "./presence.ts";
import {
  ANON_NOTIFY_ACTIVATION_ENTRY_TYPE,
  ANON_NOTIFY_STATE_CHANNEL,
  buildAnonNotifyExtensionState,
  createAnonNotifyActivationRecord,
  readAnonNotifyActivation,
  sessionServicesEnabled,
  SessionActivation,
} from "./session-state.ts";
import { assistantText, latestAssistant, normalizeText, sanitizeEventPart } from "./summary.ts";
import { evaluateRecipient, type RecipientReadiness } from "./target.ts";

const EXTENSION_DIR = dirname(fileURLToPath(import.meta.url));
const STATUS_KEY = "anon-notify";

interface TurnState {
  startedToolCalls: Set<string>;
  completedToolCalls: Set<string>;
  declaredToolCalls: number;
  allTerminate: boolean;
  anyError: boolean;
}

interface RunState {
  sessionId: string;
  idBase: string;
  startedAt: number;
  task: string;
  source: string;
  project: string;
  sequence: number;
  lowLevelRuns: number;
  turns: number;
  toolsStarted: number;
  toolsCompleted: number;
  toolErrors: number;
  providerErrors: number;
  abortObserved: boolean;
  terminalEscapeObserved: boolean;
  detachAbortListener?: () => void;
  currentTurn?: TurnState;
  lastTurn?: TurnState;
  activeTools: Map<string, string>;
  recentTools: string[];
  lastAssistantText: string;
  stopReason?: string;
  errorMessage?: string;
  lastToolError?: string;
  lastProgressAt: number;
  lastProgressTools: number;
  progressInFlight: boolean;
  lastToolErrorNotifiedAt: number;
  progressTimeout?: NodeJS.Timeout;
  progressInterval?: NodeJS.Timeout;
}

interface ExternalNotification {
  status?: NotifyStatus;
  task?: string;
  summary?: string;
  source?: string;
  eventId?: string;
  url?: string;
}

function messageStopReason(message: AgentMessage | undefined): string | undefined {
  return message?.role === "assistant" ? message.stopReason : undefined;
}

function messageError(message: AgentMessage | undefined): string | undefined {
  return message?.role === "assistant" ? message.errorMessage : undefined;
}

function modelName(ctx: ExtensionContext): string {
  return ctx.model ? `${ctx.model.provider}/${ctx.model.id}` : "unknown";
}

function notificationContextUsage(ctx: ExtensionContext): NotificationContextUsage | null {
  try {
    const usage = ctx.getContextUsage?.();
    if (!usage) return null;
    const { tokens, contextWindow, percent } = usage;
    if (!Number.isSafeInteger(contextWindow) || contextWindow <= 0) return null;
    if (tokens === null && percent === null) {
      return { usedTokens: null, contextWindow, percent: null };
    }
    if (typeof tokens !== "number" || !Number.isSafeInteger(tokens) || tokens < 0
      || typeof percent !== "number" || !Number.isFinite(percent) || percent < 0) return null;
    return { usedTokens: tokens, contextWindow, percent };
  } catch {
    // Unknown usage (including stale contexts) must not suppress the notification.
    return null;
  }
}

function isSubagentSession(ctx: ExtensionContext): boolean {
  const header = ctx.sessionManager.getHeader();
  const sessionDir = ctx.sessionManager.getSessionDir().replace(/[\\/]+$/, "");
  return Boolean(header?.parentSession && basename(sessionDir) === "tasks");
}

function taskFromPrompt(pi: ExtensionAPI, prompt: string, maxChars: number): string {
  const sessionName = pi.getSessionName()?.trim();
  if (sessionName) return normalizeText(sessionName, maxChars);
  const firstLine = prompt.split(/\r?\n/, 1)[0] || prompt;
  return normalizeText(firstLine || "Pi Agent 工作", maxChars);
}

function errorText(value: unknown): string {
  if (value instanceof Error) return value.message;
  if (typeof value === "string") return value;
  try {
    return JSON.stringify(value) ?? String(value);
  } catch {
    return String(value);
  }
}

export default function anonNotifyExtension(pi: ExtensionAPI): void {
  const presenceRuntimeId = createPresenceRuntimeId();
  let loaded: LoadedConfig | undefined;
  let client: AnonNotificationClient | undefined;
  let currentCtx: ExtensionContext | undefined;
  let terminalInputUnsubscribe: (() => void) | undefined;
  let commandJournal: CommandJournal | undefined;
  let commandPollTimer: NodeJS.Timeout | undefined;
  let commandPollInFlight: Promise<void> | undefined;
  let deferredCommand: InboundAgentCommand | undefined;
  let lifecycleGeneration = 0;
  let activationGeneration = 0;
  let nextCommandTask: string | undefined;
  let commandPollGeneration = 0;
  let presenceReporter: PresenceReporter | undefined;
  let lastPresenceActivityAt = Date.now();
  let pendingPrompt = "";
  let run: RunState | undefined;
  let sessionSuppressed = false;
  const sessionActivation = new SessionActivation();

  function config(): AnonNotifyConfig | undefined {
    return loaded?.config;
  }

  function sessionId(ctx: ExtensionContext): string {
    return ctx.sessionManager.getSessionId();
  }

  function sessionStatus(ctx: ExtensionContext): "active" | "inactive" | "subagent-suppressed" {
    return sessionActivation.status(sessionId(ctx), sessionSuppressed);
  }

  function sessionIsActive(ctx: ExtensionContext): boolean {
    return sessionServicesEnabled(sessionStatus(ctx), config()?.enabled === true);
  }

  function warnActivation(message: string, error?: unknown): void {
    const detail = error === undefined ? "" : `：${errorText(error).slice(0, 300)}`;
    console.warn(`[anon-notify] ${message}${detail}`);
  }

  function restoreSessionActivation(event: SessionStartEvent, ctx: ExtensionContext): void {
    const id = sessionId(ctx);
    if (event.reason === "new") {
      sessionActivation.restore(id, false);
      return;
    }

    if (event.reason === "fork") {
      let active = false;
      if (!event.previousSessionFile) {
        warnActivation("fork/clone 缺少源 session 路径，子 session 按 inactive 初始化");
      } else {
        try {
          active = readAnonNotifyActivation(SessionManager.open(event.previousSessionFile).getEntries());
        } catch (error) {
          warnActivation("读取 fork/clone 源 session 激活状态失败，子 session 按 inactive 初始化", error);
        }
      }

      try {
        pi.appendEntry(ANON_NOTIFY_ACTIVATION_ENTRY_TYPE, createAnonNotifyActivationRecord(active));
        sessionActivation.restore(id, active);
      } catch (error) {
        sessionActivation.restore(id, false);
        warnActivation("写入 fork/clone 子 session 激活快照失败，当前运行时按 inactive 初始化", error);
      }
      return;
    }

    sessionActivation.restore(id, readAnonNotifyActivation(ctx.sessionManager.getEntries()));
  }

  function persistSessionActivation(ctx: ExtensionContext, active: boolean): void {
    const id = sessionId(ctx);
    pi.appendEntry(ANON_NOTIFY_ACTIVATION_ENTRY_TYPE, createAnonNotifyActivationRecord(active));
    sessionActivation.restore(id, active);
  }

  function publishStructuredState(ctx: ExtensionContext): void {
    const id = sessionId(ctx);
    pi.events.emit(ANON_NOTIFY_STATE_CHANNEL, buildAnonNotifyExtensionState({
      sessionId: id,
      configLoaded: loaded !== undefined,
      globallyEnabled: loaded?.config.enabled === true,
      sessionSuppressed,
      sessionActive: sessionActivation.isActive(id),
      tokenConfigured: Boolean(loaded?.token),
    }));
  }

  async function recipientReadiness(ctx: ExtensionContext): Promise<RecipientReadiness> {
    const cfg = config();
    if (!cfg) {
      return evaluateRecipient({ qq: "", anonQq: "" });
    }
    if (!client || !loaded?.token) {
      return evaluateRecipient(cfg.recipient);
    }
    try {
      return evaluateRecipient(cfg.recipient, await client.getHealth(ctx.signal));
    } catch (error) {
      return evaluateRecipient(cfg.recipient, undefined, errorText(error));
    }
  }

  function updateUiStatus(ctx: ExtensionContext): void {
    if (!ctx.hasUI) return;
    if (!loaded) {
      ctx.ui.setStatus(STATUS_KEY, ctx.ui.theme.fg("dim", "anon unavailable"));
      return;
    }
    const status = sessionStatus(ctx);
    if (status === "subagent-suppressed") {
      ctx.ui.setStatus(STATUS_KEY, ctx.ui.theme.fg("dim", "anon main-only"));
      return;
    }
    if (!loaded?.config.enabled) {
      ctx.ui.setStatus(STATUS_KEY, ctx.ui.theme.fg("dim", "anon off"));
      return;
    }
    if (status === "inactive") {
      ctx.ui.setStatus(STATUS_KEY, ctx.ui.theme.fg("dim", "anon inactive"));
      return;
    }
    const color = loaded.token ? "success" : "warning";
    const label = loaded.token ? "anon on" : "anon no-token";
    ctx.ui.setStatus(STATUS_KEY, ctx.ui.theme.fg(color, label));
  }

  function userInterruptObserved(state: RunState | undefined): boolean {
    return Boolean(state?.abortObserved || state?.terminalEscapeObserved);
  }

  function stopCommandPolling(): void {
    commandPollGeneration += 1;
    if (commandPollTimer) clearInterval(commandPollTimer);
    commandPollTimer = undefined;
    deferredCommand = undefined;
    // Keep an old request as a guard until it settles. Its generation can no
    // longer inject, and its finally must not clear newer request ownership.
  }

  async function pollCommands(ctx: ExtensionContext, generation: number, boundary = false): Promise<void> {
    const cfg = config();
    const activeClient = client;
    const journal = commandJournal;
    const id = sessionId(ctx);
    const boundaryRun = run;
    const eligible = () => generation === commandPollGeneration
      && currentCtx !== undefined && sessionId(currentCtx) === id
      && cfg === config() && cfg?.enabled && cfg.command.enabled
      && activeClient === client && loaded?.token && sessionIsActive(ctx);
    const safeBoundary = () => !boundary || (run === boundaryRun
      && !userInterruptObserved(boundaryRun) && !ctx.signal?.aborted
      && boundaryRun?.stopReason !== "error" && boundaryRun?.stopReason !== "aborted");
    if (!eligible() || !journal || !activeClient || (!boundary && !ctx.isIdle())) return;
    if (commandPollInFlight) {
      if (!boundary) return;
      await commandPollInFlight;
      if (!eligible() || !safeBoundary()) return;
    }
    const request = (async () => {
      try {
        const command = deferredCommand || (await activeClient.listCommands(id, ctx.signal))[0];
        if (!eligible() || !safeBoundary()) return;
        if (!command) return;
        if (!commandLeaseUsable(command)) {
          deferredCommand = undefined;
          return;
        }
        if (!boundary && !ctx.isIdle()) {
          // The lease started (too late to cancel), but never enqueue native
          // steer mid-turn just because an idle poll returned late.
          deferredCommand = command;
          return;
        }
        deferredCommand = undefined;
        await deliverCommandBatch([command], {
          sessionId: id,
          journal,
          canDeliver: () => Boolean(eligible() && safeBoundary() && (boundary || ctx.isIdle())),
          appendMarker: (value) => {
            const marker = commandMarker(value);
            pi.appendEntry(marker.customType, marker.data);
          },
          sendMessage: (message, options) => {
            const previousTask = nextCommandTask;
            const previousPrompt = pendingPrompt;
            if (ctx.isIdle()) {
              nextCommandTask = command.instruction;
              pendingPrompt = command.instruction;
            }
            try { pi.sendMessage(message, options); }
            catch (error) {
              nextCommandTask = previousTask;
              pendingPrompt = previousPrompt;
              throw error;
            }
          },
          ackCommand: (value, status, error) => activeClient.ackCommand(value, status, error, ctx.signal),
          onLog: (message) => console.log(`[anon-notify] ${message}`),
        });
      } catch {
        // Network/journal failure leaves the lease for recovery. No new
        // error-level log and no failed background hook can crash the Agent.
      }
    })();
    commandPollInFlight = request;
    try { await request; }
    finally { if (commandPollInFlight === request) commandPollInFlight = undefined; }
  }

  function startCommandPolling(ctx: ExtensionContext): void {
    stopCommandPolling();
    const cfg = config();
    if (!cfg?.enabled || !cfg.command.enabled || !client || !loaded?.token || !sessionIsActive(ctx)) return;
    const generation = commandPollGeneration;
    void pollCommands(ctx, generation);
    commandPollTimer = setInterval(() => {
      void pollCommands(ctx, generation);
    }, cfg.command.pollIntervalMs);
    commandPollTimer.unref?.();
  }

  function presenceEligibility(ctx: ExtensionContext): PresenceEligibility {
    const cfg = config();
    return {
      configLoaded: loaded !== undefined,
      globallyEnabled: cfg?.enabled === true,
      tokenConfigured: Boolean(loaded?.token),
      commandEnabled: cfg?.command.enabled === true,
      sessionSuppressed,
      sessionActive: sessionActivation.isActive(sessionId(ctx)),
    };
  }

  function presenceSnapshot(): PresenceSnapshot | undefined {
    const ctx = currentCtx;
    if (!ctx || !presenceEnabled(presenceEligibility(ctx))) return undefined;
    return {
      sessionId: sessionId(ctx),
      runtimeId: presenceRuntimeId,
      // Only the live runtime knows whether the Agent is idle or running; the
      // server registry stores whatever the latest heartbeat reported.
      state: ctx.isIdle() ? "idle" : "running",
      project: basename(ctx.cwd) || ctx.cwd,
      model: modelName(ctx),
      lastActivityAt: new Date(lastPresenceActivityAt).toISOString(),
    };
  }

  function ensurePresenceReporter(): PresenceReporter {
    presenceReporter ??= new PresenceReporter({
      buildSnapshot: presenceSnapshot,
      sendHeartbeat: async (body) => { await client?.heartbeatPresence(body); },
      sendOffline: async (body) => { await client?.markPresenceOffline(body); },
    });
    return presenceReporter;
  }

  function startPresence(ctx: ExtensionContext): void {
    const cfg = config();
    if (!cfg || !presenceEnabled(presenceEligibility(ctx))) {
      presenceReporter?.stop();
      return;
    }
    ensurePresenceReporter().start(cfg.presence.heartbeatIntervalMs);
  }

  async function stopPresence(offline: boolean): Promise<void> {
    if (!presenceReporter) return;
    if (offline) await presenceReporter.goOffline();
    else presenceReporter.stop();
  }

  async function reload(ctx: ExtensionContext, generation: number): Promise<void> {
    if (run) throw new Error("Agent 运行中不能重载 Anon 通知配置");
    stopCommandPolling();
    let next: ReturnType<typeof loadConfig>;
    try {
      next = loadConfig(ctx, EXTENSION_DIR);
    } catch (error) {
      loaded = undefined;
      client = undefined;
      commandJournal = undefined;
      currentCtx = ctx;
      sessionSuppressed = false;
      updateUiStatus(ctx);
      publishStructuredState(ctx);
      throw error;
    }

    await client?.flush();
    if (generation !== lifecycleGeneration) return;
    const nextDeliveryAt = client?.getNextDeliveryAt() || 0;
    loaded = next;
    currentCtx = ctx;
    client = new AnonNotificationClient(next.config, next.token, nextDeliveryAt);
    commandJournal = new CommandJournal(next.commandJournalPath);
    commandJournal.reconcileSessionEntries(ctx.sessionManager.getEntries());
    sessionSuppressed = next.config.mainAgentOnly && isSubagentSession(ctx);
    updateUiStatus(ctx);
    publishStructuredState(ctx);
    startCommandPolling(ctx);
    startPresence(ctx);
  }

  function clearProgressTimers(state: RunState): void {
    if (state.progressTimeout) clearTimeout(state.progressTimeout);
    if (state.progressInterval) clearInterval(state.progressInterval);
    state.progressTimeout = undefined;
    state.progressInterval = undefined;
  }

  function bindAbortSignal(state: RunState, ctx: ExtensionContext): void {
    state.detachAbortListener?.();
    state.detachAbortListener = undefined;
    const signal = ctx.signal;
    if (!signal) return;
    const onAbort = () => {
      if (run === state) state.abortObserved = true;
    };
    if (signal.aborted) {
      onAbort();
      return;
    }
    signal.addEventListener("abort", onAbort, { once: true });
    state.detachAbortListener = () => signal.removeEventListener("abort", onAbort);
  }

  function eventId(state: RunState, kind: string): string {
    state.sequence += 1;
    return [state.idBase, sanitizeEventPart(kind), String(state.sequence)].join(":");
  }

  function decorateSummary(state: RunState, summary: string, cfg: AnonNotifyConfig): string {
    const details: string[] = [];
    if (cfg.content.includeProject) details.push(`项目：${state.project}`);
    const body = [summary.trim(), details.join("；")].filter(Boolean).join("\n");
    return normalizeText(body, cfg.content.maxSummaryChars);
  }

  async function deliver(
    state: RunState,
    kind: string,
    status: NotifyStatus,
    summary: string,
    ctx: ExtensionContext,
  ): Promise<void> {
    const cfg = config();
    if (!cfg?.enabled || !client || !sessionIsActive(ctx)
      || !currentCtx || sessionId(currentCtx) !== state.sessionId) return;
    const payload: NotificationPayload = {
      eventId: eventId(state, kind),
      sessionId: ctx.sessionManager.getSessionId(),
      source: state.source,
      task: state.task,
      status,
      summary: decorateSummary(state, summary, cfg),
      contextUsage: notificationContextUsage(ctx),
      reportedAt: new Date().toISOString(),
    };
    try {
      const isTerminal = kind === "completed" || kind === "failed";
      await client.enqueue(payload, isTerminal);
    } catch (error) {
      console.error(`[anon-notify] ${kind} 通知失败：${errorText(error)}`);
    }
  }

  function progressSummary(state: RunState): string {
    const active = [...state.activeTools.values()];
    if (active.length) return `工作还在继续，当前正在处理：${active.slice(0, 3).join("、")}。`;
    if (state.recentTools.length) return `工作有新进展，最近完成了：${state.recentTools.slice(-3).join("、")}。`;
    if (state.turns > 0) return `已经完成 ${state.turns} 轮处理，工作还在继续。`;
    return "任务仍在处理中，目前还没有新的可见工具结果。";
  }

  async function sendProgress(state: RunState, ctx: ExtensionContext): Promise<void> {
    const cfg = config();
    if (!cfg?.notify.progress || !sessionIsActive(ctx) || run !== state || state.progressInFlight) return;
    const now = Date.now();
    if (now - state.startedAt < cfg.progress.minRunDurationMs) return;
    if (state.lastProgressAt && now - state.lastProgressAt < cfg.progress.intervalMs) return;

    state.progressInFlight = true;
    state.lastProgressAt = now;
    state.lastProgressTools = state.toolsCompleted;
    try {
      await deliver(state, "progress", "info", progressSummary(state), ctx);
    } finally {
      state.progressInFlight = false;
    }
  }

  function startProgressTimers(state: RunState, ctx: ExtensionContext): void {
    const cfg = config();
    if (!cfg?.notify.progress || !sessionIsActive(ctx) || state.progressTimeout || state.progressInterval) return;
    const firstDelay = Math.max(1_000, cfg.progress.minRunDurationMs);
    state.progressTimeout = setTimeout(() => {
      void sendProgress(state, ctx).finally(() => {
        if (run !== state || !config()?.notify.progress || !sessionIsActive(ctx)) return;
        state.progressInterval = setInterval(() => {
          void sendProgress(state, ctx);
        }, config()!.progress.intervalMs);
      });
    }, firstDelay);
  }

  async function sendExternal(data: ExternalNotification): Promise<boolean> {
    const ctx = currentCtx;
    const cfg = config();
    if (!ctx || !cfg?.enabled || !client || !sessionIsActive(ctx)) return false;
    const now = Date.now();
    const synthetic: RunState = {
      sessionId: sessionId(ctx),
      idBase: `pi:${sanitizeEventPart(ctx.sessionManager.getSessionId())}:${now}`,
      startedAt: now,
      task: normalizeText(data.task || "Pi 扩展通知", cfg.content.maxTaskChars),
      source: normalizeText(data.source || renderSource(cfg.source, ctx), 120),
      project: basename(ctx.cwd) || ctx.cwd,
      sequence: 0,
      lowLevelRuns: 0,
      turns: 0,
      toolsStarted: 0,
      toolsCompleted: 0,
      toolErrors: 0,
      providerErrors: 0,
      abortObserved: false,
      terminalEscapeObserved: false,
      activeTools: new Map(),
      recentTools: [],
      lastAssistantText: "",
      lastProgressAt: 0,
      lastProgressTools: 0,
      progressInFlight: false,
      lastToolErrorNotifiedAt: 0,
    };
    const payload: NotificationPayload = {
      eventId: data.eventId || eventId(synthetic, "external"),
      sessionId: ctx.sessionManager.getSessionId(),
      source: synthetic.source,
      task: synthetic.task,
      status: data.status || "info",
      summary: normalizeText(data.summary || "Pi 扩展发来一条通知。", cfg.content.maxSummaryChars),
      url: data.url,
      contextUsage: notificationContextUsage(ctx),
      reportedAt: new Date().toISOString(),
    };
    await client.enqueue(payload);
    return true;
  }

  pi.on("session_start", async (event, ctx) => {
    const generation = ++lifecycleGeneration;
    activationGeneration += 1;
    stopCommandPolling();
    if (run) {
      clearProgressTimers(run);
      run.detachAbortListener?.();
    }
    run = undefined;
    // A session switch may not emit a separate shutdown before this hook. Mark
    // the previous live session offline before rebinding currentCtx, otherwise
    // its registry record could remain routable until TTL expiry.
    await stopPresence(true);
    if (generation !== lifecycleGeneration) return;
    restoreSessionActivation(event, ctx);
    currentCtx = ctx;
    pendingPrompt = "";
    nextCommandTask = undefined;
    run = undefined;
    sessionSuppressed = false;
    lastPresenceActivityAt = Date.now();
    terminalInputUnsubscribe?.();
    terminalInputUnsubscribe = ctx.mode === "tui"
      ? ctx.ui.onTerminalInput((data) => {
          // 精确匹配独立 Esc，避免把方向键等以 ESC 开头的控制序列误判成用户中断。
          if (data === "\u001b" && run) run.terminalEscapeObserved = true;
          return undefined;
        })
      : undefined;
    try {
      await reload(ctx, generation);
    } catch (error) {
      console.error(`[anon-notify] 加载配置失败：${errorText(error)}`);
    }
  });

  pi.on("before_agent_start", (event) => {
    pendingPrompt = event.prompt;
  });

  async function observeRun(ctx: ExtensionContext, notifyStarted = true): Promise<void> {
    const cfg = config();
    if (!cfg || sessionSuppressed || !currentCtx || sessionId(currentCtx) !== sessionId(ctx)) return;
    currentCtx = ctx;
    lastPresenceActivityAt = Date.now();
    presenceReporter?.refresh();
    if (run) {
      run.lowLevelRuns += 1;
      // 新的低层 run 代表 Pi 正在重试或继续处理；清除上一轮的临时终止状态，但保留已观察到的用户中断。
      run.stopReason = undefined;
      run.errorMessage = undefined;
      run.currentTurn = undefined;
      run.lastTurn = undefined;
      // 若 Esc 后仍启动了新的低层 run，说明它只取消了局部 UI 或没有真正终止高层任务。
      run.terminalEscapeObserved = false;
      bindAbortSignal(run, ctx);
      return;
    }
    const now = Date.now();
    const id = sessionId(ctx);
    const task = nextCommandTask === undefined
      ? taskFromPrompt(pi, pendingPrompt, cfg.content.maxTaskChars)
      : normalizeText(nextCommandTask, cfg.content.maxTaskChars);
    nextCommandTask = undefined;
    run = {
      sessionId: id,
      idBase: `pi:${sanitizeEventPart(id)}:${now}`,
      startedAt: now,
      task,
      source: normalizeText(renderSource(cfg.source, ctx), 120),
      project: basename(ctx.cwd) || ctx.cwd,
      sequence: 0,
      lowLevelRuns: 1,
      turns: 0,
      toolsStarted: 0,
      toolsCompleted: 0,
      toolErrors: 0,
      providerErrors: 0,
      abortObserved: false,
      terminalEscapeObserved: false,
      activeTools: new Map(),
      recentTools: [],
      lastAssistantText: "",
      lastProgressAt: 0,
      lastProgressTools: 0,
      progressInFlight: false,
      lastToolErrorNotifiedAt: 0,
    };
    bindAbortSignal(run, ctx);
    startProgressTimers(run, ctx);
    if (notifyStarted && cfg.notify.started) await deliver(run, "started", "info", "任务已经开始处理。", ctx);
  }

  pi.on("agent_start", async (_event, ctx) => { await observeRun(ctx); });

  pi.on("turn_start", () => {
    if (!run) return;
    run.currentTurn = {
      startedToolCalls: new Set(),
      completedToolCalls: new Set(),
      declaredToolCalls: 0,
      allTerminate: true,
      anyError: false,
    };
  });

  pi.on("tool_execution_start", (event) => {
    if (!run) return;
    run.toolsStarted += 1;
    run.activeTools.set(event.toolCallId, event.toolName);
    run.currentTurn?.startedToolCalls.add(event.toolCallId);
  });

  pi.on("tool_execution_end", async (event, ctx) => {
    const state = run;
    const cfg = config();
    if (!state || !cfg) return;
    state.toolsCompleted += 1;
    lastPresenceActivityAt = Date.now();
    state.activeTools.delete(event.toolCallId);
    state.recentTools.push(event.toolName);
    if (state.currentTurn) {
      state.currentTurn.completedToolCalls.add(event.toolCallId);
      state.currentTurn.allTerminate &&= (event.result as { terminate?: boolean }).terminate === true;
      state.currentTurn.anyError ||= event.isError;
    }
    if (!event.isError) state.terminalEscapeObserved = false;
    state.recentTools = state.recentTools.slice(-12);

    if (event.isError) {
      state.toolErrors += 1;
      state.lastToolError = `${event.toolName}: ${errorText(event.result).slice(0, 300)}`;
      const now = Date.now();
      if (
        cfg.notify.toolErrors &&
        now - state.lastToolErrorNotifiedAt >= cfg.progress.toolErrorCooldownMs
      ) {
        state.lastToolErrorNotifiedAt = now;
        await deliver(state, "tool-error", "warning", `工具 ${event.toolName} 执行失败，但 Agent 可能会继续尝试恢复。`, ctx);
      }
    }

    const enoughNewTools = state.toolsCompleted - state.lastProgressTools >= cfg.progress.minCompletedTools;
    const longEnough = Date.now() - state.startedAt >= cfg.progress.minRunDurationMs;
    const cooldown = Date.now() - state.lastProgressAt >= cfg.progress.intervalMs;
    if (cfg.notify.progress && enoughNewTools && longEnough && cooldown) {
      await sendProgress(state, ctx);
    }
  });

  pi.on("turn_end", async (event, ctx) => {
    if (run) {
      run.turns += 1;
      lastPresenceActivityAt = Date.now();
      if (event.message.role === "assistant") {
        const text = assistantText(event.message);
        if (text) run.lastAssistantText = text;
        // 只保留最新 assistant 的最终状态，避免低层 run 自动重试成功后仍残留旧错误。
        run.stopReason = messageStopReason(event.message);
        run.errorMessage = messageError(event.message);
        if (run.stopReason === "stop") run.terminalEscapeObserved = false;
        if (run.currentTurn) {
          run.currentTurn.declaredToolCalls = event.message.content.filter((part) => part.type === "toolCall").length;
          run.lastTurn = run.currentTurn;
          run.currentTurn = undefined;
        }
      }
    }
    if (run && !userInterruptObserved(run) && !ctx.signal?.aborted
      && run.stopReason !== "error" && run.stopReason !== "aborted") {
      await pollCommands(ctx, commandPollGeneration, true);
    }
  });

  pi.on("after_provider_response", (event) => {
    if (run && event.status >= 400) run.providerErrors += 1;
  });

  pi.on("agent_end", (event) => {
    if (!run) return;
    const assistant = latestAssistant(event.messages);
    const text = assistantText(assistant);
    if (text) run.lastAssistantText = text;
    if (assistant) {
      run.stopReason = messageStopReason(assistant);
      run.errorMessage = messageError(assistant);
    }
  });

  pi.on("agent_settled", async (_event, ctx) => {
    const state = run;
    const cfg = config();
    run = undefined;
    lastPresenceActivityAt = Date.now();
    presenceReporter?.refresh();
    if (!state || !cfg) return;
    clearProgressTimers(state);
    state.detachAbortListener?.();
    state.detachAbortListener = undefined;

    const outcome = classifyRunOutcome({
      stopReason: state.stopReason,
      errorMessage: state.errorMessage,
      providerErrors: state.providerErrors,
      lastToolError: state.lastToolError,
      abortObserved: userInterruptObserved(state),
      terminalToolBatch: state.lastTurn
        ? {
            declaredToolCalls: state.lastTurn.declaredToolCalls,
            startedToolCalls: state.lastTurn.startedToolCalls.size,
            completedToolCalls: state.lastTurn.completedToolCalls.size,
            allTerminate: state.lastTurn.allTerminate,
            anyError: state.lastTurn.anyError,
          }
        : undefined,
    });


    if (outcome.kind === "interrupted") {
      // TUI Esc、低层 AbortSignal 或显式 abort 造成的用户中断始终静默。
      return;
    }

    if (outcome.kind === "failed") {
      if (cfg.notify.failed) {
        await deliver(state, "failed", "failed", `任务没有正常完成：${outcome.detail}`, ctx);
      }
      return;
    }

    if (cfg.notify.completed) {
      const finalText = cfg.content.includeLastAssistantMessage ? state.lastAssistantText : "";
      const summary = finalText
        ? `任务已经完成。结果摘要：${normalizeText(finalText, cfg.content.maxSummaryChars)}`
        : "任务已经完成。";
      await deliver(state, "completed", "completed", summary, ctx);
    }
  });

  pi.on("session_shutdown", async () => {
    lifecycleGeneration += 1;
    activationGeneration += 1;
    stopCommandPolling();
    // Best-effort deregistration before the runtime identity is cleared; a
    // missed offline request is covered by the server heartbeat TTL.
    const offline = stopPresence(true);
    terminalInputUnsubscribe?.();
    terminalInputUnsubscribe = undefined;
    commandJournal = undefined;
    sessionSuppressed = false;
    currentCtx = undefined;
    const state = run;
    run = undefined;
    if (state) {
      clearProgressTimers(state);
      state.detachAbortListener?.();
      state.detachAbortListener = undefined;
    }
    // quit/reload/new/resume/fork 都是用户或会话生命周期操作，不作为 Agent 自行结束发送通知。
    await Promise.all([offline, client?.flush()]);
  });

  pi.events.on("anon-notify:send", (data) => {
    void sendExternal((data || {}) as ExternalNotification).catch((error) => {
      console.error(`[anon-notify] 外部扩展通知失败：${errorText(error)}`);
    });
  });

  pi.registerCommand("anon-notify", {
    description: "切换当前 Pi 会话的 Anon 工作通知（裸命令；再次执行关闭）",
    handler: async (args, ctx) => {
      if (args.trim()) {
        // Subcommands were removed on purpose. Treat any argument as a mistake
        // instead of interpreting it as enable/disable/status/reload/test.
        if (ctx.hasUI) {
          ctx.ui.notify("请直接使用裸命令 /anon-notify 切换当前会话；旧子命令已移除", "warning");
        }
        return;
      }

      const cfg = config();
      if (!cfg || !loaded) {
        if (ctx.hasUI) ctx.ui.notify("Anon 通知配置尚未加载", "warning");
        return;
      }

      const generation = lifecycleGeneration;
      const operation = ++activationGeneration;
      const id = sessionId(ctx);
      if (!currentCtx || sessionId(currentCtx) !== id) return;
      if (sessionActivation.isActive(id)) {
        try {
          persistSessionActivation(ctx, false);
        } catch (error) {
          console.error(`[anon-notify] Anon 通知停用状态保存失败：${errorText(error)}`);
          throw error;
        }
        stopCommandPolling();
        if (run) clearProgressTimers(run);
        await stopPresence(true);
        if (generation !== lifecycleGeneration || operation !== activationGeneration) return;
        updateUiStatus(ctx);
        publishStructuredState(ctx);
        if (ctx.hasUI) ctx.ui.notify("Anon 通知已在当前 Pi 会话停用", "info");
        return;
      }

      if (!cfg.enabled) {
        if (ctx.hasUI) ctx.ui.notify("Anon 通知被全局配置关闭，无法在当前会话启用", "warning");
        return;
      }
      if (sessionSuppressed) {
        if (ctx.hasUI) ctx.ui.notify("当前是 subagent 会话，不能启用主 Agent 通知", "warning");
        return;
      }
      const target = await recipientReadiness(ctx);
      if (generation !== lifecycleGeneration || operation !== activationGeneration
        || cfg !== config() || !currentCtx || sessionId(currentCtx) !== id || sessionId(ctx) !== id) return;
      if (!target.ok) {
        if (ctx.hasUI) ctx.ui.notify(target.message, "warning");
        else console.warn(`[anon-notify] ${target.message}`);
        return;
      }
      try {
        persistSessionActivation(ctx, true);
      } catch (error) {
        console.error(`[anon-notify] Anon 通知启用状态保存失败：${errorText(error)}`);
        throw error;
      }
      currentCtx = ctx;
      lastPresenceActivityAt = Date.now();
      updateUiStatus(ctx);
      publishStructuredState(ctx);
      if (!ctx.isIdle() && !run) await observeRun(ctx, false);
      if (run) startProgressTimers(run, ctx);
      startCommandPolling(ctx);
      startPresence(ctx);
      if (ctx.hasUI) {
        ctx.ui.notify(
          `Anon 通知已在当前 Pi 会话启用。${target.message}`,
          target.code === "health_unavailable" || target.healthStatus === "unknown" ? "warning" : "info",
        );
      }
    },
  });
}