/**
 * Due-job runner with reliability controls.
 *
 * Fires on:
 *  1. session_start (startup/new/resume) when jobs are due AND no CLI initial prompt
 *  2. an in-session ticker, only while idle
 *
 * Action kinds (see action.ts):
 *  - prompt: inject isolated agent task (original behavior)
 *  - shell: pi.exec command; optional agent wake via wakeOn
 *  - notify: UI/console reminder only
 *  - message: session custom message (display, no agent turn)
 *
 * Mitigations (see docs/RELIABILITY.md).
 */

import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";
import { existsSync } from "node:fs";
import {
  DEFAULT_JOB_ACTION,
  DEFAULT_SHELL_TIMEOUT_MS,
  selectShellFollowUp,
  shouldWakeForShell,
  terminalReason,
  truncateOutput,
} from "./action.js";
import { shouldSkipDueOnSessionStart } from "./cli-prompt.js";
import { RunLedger, buildRun, newRunId } from "./ledger.js";
import { JobLockManager } from "./lock.js";
import {
  DEFAULT_MISSED_WINDOW,
  DEFAULT_TIER,
  LIMITS,
  decideDue,
  idempotencyKeyFor,
} from "./policy.js";
import { PrivilegeGuard } from "./privilege.js";
import {
  buildFirePrompt,
  buildShellFollowUpPrompt,
  notifyLabel,
} from "./prompt.js";
import { StoreError, type ScheduleStore } from "./store.js";
import type {
  FireSource,
  JobAction,
  ScheduledJob,
  ShellRunResult,
} from "./types.js";

/** How often the in-session ticker checks for due jobs. */
export const TICK_MS = 30_000;

/** Poll interval while waiting for context compaction to finish. */
export const COMPACTION_POLL_MS = 500;
/** While parked on the compaction flag, probe with a real send this often. */
export const COMPACTION_PROBE_MS = 5_000;
/** Max time a delivery waits for compaction before falling back to the error path. */
export const COMPACTION_WAIT_MAX_MS = 120_000;

/**
 * Recognize pi's prompt-submission rejection raised while context
 * compaction is in flight:
 * "Cannot submit a prompt while compaction is in progress. Wait for
 * compaction to finish and retry."
 */
export function isCompactionBusyError(err: unknown): boolean {
  return err instanceof Error && err.message.includes("compaction is in progress");
}

/**
 * Candidate Git Bash (MSYS) binaries on Windows, in preference order. The bare
 * `"bash"` is ambiguous on Windows: PATH resolution can pick
 * `C:\Windows\System32\bash.exe` (the WSL launcher), which exits non-zero
 * with NO execution when no WSL distro is installed (`execvpe(/bin/bash)
 * failed`). Git Bash matches pi's own bash tool + the `/c/` path conventions
 * in `shellCommandPrefix`, so prefer it.
 */
const WIN_GIT_BASH_PATHS = [
  "C:\\Program Files\\Git\\bin\\bash.exe",
  "C:\\Program Files\\Git\\usr\\bin\\bash.exe",
  "C:\\Program Files (x86)\\Git\\bin\\bash.exe",
];

/**
 * Resolve the shell binary passed to `pi.exec` for shell jobs.
 *
 * - `PI_SCHEDULE_SHELL` env (absolute path) wins outright — lets users force a
 *   specific binary without a release.
 * - On Windows, prefer the first existing Git Bash (see {@link WIN_GIT_BASH_PATHS}).
 * - Otherwise fall back to `"bash"` (POSIX; the historic behavior).
 */
export function resolveShell(): string {
  const override = process.env["PI_SCHEDULE_SHELL"];
  if (override) return override;
  if (process.platform !== "win32") return "bash";
  for (const candidate of WIN_GIT_BASH_PATHS) {
    try {
      if (existsSync(candidate)) return candidate;
    } catch {
      // permission / race — try the next candidate
    }
  }
  return "bash";
}

const FIRE_ON_REASONS = new Set(["startup", "new", "resume"]);
const ERROR_NOTIFY_COOLDOWN_MS = 5 * 60_000;

export interface RunnerOptions {
  store: ScheduleStore;
  pi: ExtensionAPI;
  ledger?: RunLedger;
  locks?: JobLockManager;
  privilege?: PrivilegeGuard;
  hasInitialPrompt?: () => boolean;
  now?: () => Date;
  tickMs?: number;
  /** Delivery wait budget for in-flight compaction (default {@link COMPACTION_WAIT_MAX_MS}). */
  compactionWaitMs?: number;
  /** Poll interval for the compaction wait (default {@link COMPACTION_POLL_MS}). */
  compactionPollMs?: number;
  /** Send-probe cadence while parked on the compaction flag (default {@link COMPACTION_PROBE_MS}). */
  compactionProbeMs?: number;
}

export class ScheduleRunner {
  private timer: ReturnType<typeof setInterval> | null = null;
  private cwd = process.cwd();
  private waveActive = false;
  /** Serializes waves so run_now waits instead of silently no-oping. */
  private waveChain: Promise<unknown> = Promise.resolve();
  private lastErrorNotifyAt = 0;
  private lastErrorNotifyKey = "";
  /** True between session_before_compact and session_compact (hint only). */
  private compacting = false;
  private readonly hasInitialPrompt: (() => boolean) | undefined;
  private readonly now: () => Date;
  private readonly tickMs: number;
  private readonly compactionWaitMs: number;
  private readonly compactionPollMs: number;
  private readonly compactionProbeMs: number;
  private readonly ledger: RunLedger;
  private readonly locks: JobLockManager;
  private readonly privilege: PrivilegeGuard;

  constructor(private readonly opts: RunnerOptions) {
    this.hasInitialPrompt = opts.hasInitialPrompt;
    this.now = opts.now ?? (() => new Date());
    this.tickMs = opts.tickMs ?? TICK_MS;
    this.compactionWaitMs = opts.compactionWaitMs ?? COMPACTION_WAIT_MAX_MS;
    this.compactionPollMs = opts.compactionPollMs ?? COMPACTION_POLL_MS;
    this.compactionProbeMs = opts.compactionProbeMs ?? COMPACTION_PROBE_MS;

    const paths = opts.store.pathsInfo();
    this.ledger = opts.ledger ?? new RunLedger(paths.runsFile);
    this.locks = opts.locks ?? new JobLockManager(paths.lockDir);
    this.privilege = opts.privilege ?? new PrivilegeGuard();
  }

  /** Bind session lifecycle + privilege hooks. Call once from extension factory. */
  attach(): void {
    const { pi } = this.opts;
    this.privilege.attach(pi);

    pi.on("session_start", async (event, ctx) => {
      this.cwd = ctx.cwd;
      this.stopTicker();
      this.compacting = false; // fresh session cannot be mid-compaction

      if (FIRE_ON_REASONS.has(event.reason)) {
        if (shouldSkipDueOnSessionStart(event.reason, this.hasInitialPrompt)) {
          this.startTicker(ctx);
          return;
        }
        await this.fireDue(ctx, { source: "session_start" });
      }

      this.startTicker(ctx);
    });

    // Track compaction so scheduled wakes wait it out instead of crashing
    // into pi's "Cannot submit a prompt while compaction is in progress".
    // Hint only: the thrown error is the authoritative backstop (race window
    // before session_before_compact fires, or a missed event).
    pi.on("session_before_compact", () => {
      this.compacting = true;
    });
    pi.on("session_compact", () => {
      this.compacting = false;
    });

    pi.on("session_shutdown", () => {
      this.stopTicker();
      this.privilege.clear();
      this.compacting = false;
    });
  }

  /**
   * Process due jobs (or forced job ids).
   * Auto waves drop if another auto wave is active.
   * run_now serializes on the wave chain (never silent no-op).
   */
  async fireDue(
    ctx: ExtensionContext,
    meta: { source: FireSource; jobIds?: string[] },
  ): Promise<ScheduledJob[]> {
    if (meta.source === "run_now") {
      const result = this.waveChain.then(() => this.runWave(ctx, meta));
      this.waveChain = result.then(
        () => undefined,
        () => undefined,
      );
      return result;
    }

    if (this.waveActive) return [];
    return this.runWave(ctx, meta);
  }

  private async runWave(
    ctx: ExtensionContext,
    meta: { source: FireSource; jobIds?: string[] },
  ): Promise<ScheduledJob[]> {
    if (this.waveActive && meta.source !== "run_now") return [];
    this.waveActive = true;

    try {
      const now = this.now();
      let candidates: ScheduledJob[];

      if (meta.jobIds && meta.jobIds.length > 0) {
        candidates = meta.jobIds
          .map((id) => this.opts.store.get(id, this.cwd))
          .filter((j): j is ScheduledJob => Boolean(j));
      } else {
        if (meta.source === "tick" && !ctx.isIdle()) {
          return [];
        }
        candidates = this.opts.store.dueJobs(this.cwd, now);
      }

      if (candidates.length === 0) return [];

      const maxFires =
        meta.source === "session_start"
          ? LIMITS.maxFiresPerSessionStart
          : meta.source === "tick"
            ? LIMITS.maxFiresPerTick
            : candidates.length;

      const updated: ScheduledJob[] = [];
      let attempts = 0; // ok + error count toward cap

      for (const job of candidates) {
        const forced = meta.source === "run_now";
        const result = await this.processOne(ctx, job, {
          source: meta.source,
          forced,
          deliverAs: attempts === 0 ? undefined : "followUp",
          allowFire: forced || attempts < maxFires,
        });
        if (result) {
          updated.push(result);
          if (result.lastStatus === "ok" || result.lastStatus === "error") {
            attempts += 1;
          }
        }
      }

      return updated;
    } catch (err) {
      if (meta.source === "run_now") throw err;
      if (err instanceof StoreError) {
        this.emitError(ctx, `store error: ${err.message}`, `store:${err.message}`);
        return [];
      }
      const message = err instanceof Error ? err.message : String(err);
      this.emitError(ctx, `runner error: ${message}`, `runner:${message}`);
      return [];
    } finally {
      this.waveActive = false;
    }
  }

  private emitError(ctx: ExtensionContext, detail: string, key: string): void {
    const now = Date.now();
    if (
      key === this.lastErrorNotifyKey &&
      now - this.lastErrorNotifyAt < ERROR_NOTIFY_COOLDOWN_MS
    ) {
      return;
    }
    this.lastErrorNotifyAt = now;
    this.lastErrorNotifyKey = key;

    const msg = `[pi-schedule] ${detail}`;
    if (ctx.hasUI) ctx.ui.notify(msg, "error");
    else console.error(msg);
  }

  private alreadyDelivered(job: ScheduledJob, key: string): boolean {
    // Durable primary: store row survives ledger window eviction.
    if (job.lastIdempotencyKey === key && job.lastStatus === "ok") return true;
    return this.ledger.wasDelivered(key);
  }

  private recordBestEffort(
    args: Parameters<typeof buildRun>[0],
  ): void {
    this.ledger.append(buildRun(args));
  }

  /** One submission attempt — the original synchronous send. */
  private sendNow(
    body: string,
    ctx: ExtensionContext,
    deliverAs?: "followUp" | "steer",
  ): void {
    if (deliverAs || !ctx.isIdle()) {
      this.opts.pi.sendUserMessage(body, {
        deliverAs: deliverAs ?? "followUp",
      });
    } else {
      this.opts.pi.sendUserMessage(body);
    }
  }

  /** Bounded sleep used by the compaction wait loop. */
  private sleep(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }

  /**
   * Submit the agent message, waiting out in-flight context compaction.
   *
   * pi rejects prompt submission while compaction runs, and a scheduled wake
   * can easily land in that window (long shell poll + full context). The
   * send attempt itself is the probe — the throw is raised at the top of
   * pi's prompt(), before events/expansion/enqueue, so a doomed attempt is
   * side-effect-free — and the session-event flag only paces retries:
   *
   * - attempt immediately (covers a stale flag: compaction already ended or
   *   was cancelled without a session_compact event);
   * - while the throw says compaction is running, sleep a poll tick and
   *   retry as soon as session_compact flips the flag (≤ one poll tick);
   * - every compactionProbeMs, retry the send even while the flag is still
   *   set: a CANCELLED compaction never emits session_compact, so only a
   *   probe can discover it (≤ one probe interval of delay);
   * - bounded by compactionWaitMs so a stuck compaction degrades to the
   *   normal delivery-error path (advance + ledger), never a hung wave.
   */
  private async sendAgentMessage(
    body: string,
    ctx: ExtensionContext,
    deliverAs?: "followUp" | "steer",
  ): Promise<void> {
    const deadline = Date.now() + this.compactionWaitMs;

    for (;;) {
      try {
        this.sendNow(body, ctx, deliverAs);
        this.compacting = false; // flag can go stale (cancelled compaction)
        return;
      } catch (err) {
        if (!isCompactionBusyError(err)) throw err;
        if (Date.now() >= deadline) throw err;

        // Authoritative: pi is compacting. Park on the event flag for quick
        // resume, but cap the park at the probe cadence / deadline so a
        // probe send happens even if session_compact never fires.
        const parkUntil = Math.min(
          Date.now() + this.compactionProbeMs,
          deadline,
        );
        do {
          await this.sleep(this.compactionPollMs);
        } while (this.compacting && Date.now() < parkUntil);
      }
    }
  }

  private async deliver(
    ctx: ExtensionContext,
    job: ScheduledJob,
    opts: {
      runId: string;
      source: FireSource;
      forced: boolean;
      deliverAs?: "followUp" | "steer";
    },
  ): Promise<{ detail?: string; wokeAgent: boolean; lastShell?: ShellRunResult }> {
    const action: JobAction = job.action ?? DEFAULT_JOB_ACTION;

    if (action === "notify") {
      const msg = notifyLabel(job);
      if (ctx.hasUI) ctx.ui.notify(msg, "info");
      else console.log(msg);
      this.opts.pi.sendMessage?.(
        {
          customType: "pi-schedule",
          content: msg,
          display: true,
          details: { jobId: job.id, action: "notify", runId: opts.runId },
        },
        { triggerTurn: false },
      );
      return { detail: "notify", wokeAgent: false };
    }

    if (action === "message") {
      const body = job.prompt.trim() || job.name;
      if (this.opts.pi.sendMessage) {
        this.opts.pi.sendMessage(
          {
            customType: "pi-schedule",
            content: body,
            display: true,
            details: { jobId: job.id, action: "message", runId: opts.runId },
          },
          { triggerTurn: false },
        );
      } else {
        // No custom-message channel on this pi build — surface to the console
        // so the message is still delivered somewhere rather than throwing
        // (notify/shell already treat sendMessage as optional via `?.`).
        console.log(notifyLabel(job));
      }
      return { detail: "message", wokeAgent: false };
    }

    if (action === "shell") {
      return this.deliverShell(ctx, job, opts);
    }

    // prompt (default)
    const body = buildFirePrompt({
      job,
      runId: opts.runId,
      source: opts.source,
      forced: opts.forced,
    });
    await this.sendAgentMessage(body, ctx, opts.deliverAs);
    return { detail: "prompt", wokeAgent: true };
  }

  private async deliverShell(
    ctx: ExtensionContext,
    job: ScheduledJob,
    opts: {
      runId: string;
      source: FireSource;
      forced: boolean;
      deliverAs?: "followUp" | "steer";
    },
  ): Promise<{ detail?: string; wokeAgent: boolean; lastShell?: ShellRunResult }> {
    const command = job.command?.trim();
    if (!command) {
      throw new Error(`shell job "${job.name}" has no command`);
    }

    // Global jobs have no projectPath, so they run in the session cwd — a
    // relative command is therefore session-dependent. Prefer absolute
    // commands or a project-scoped job for a deterministic cwd.
    const cwd = job.projectPath ?? ctx.cwd ?? this.cwd;
    const timeoutMs = job.timeoutMs ?? DEFAULT_SHELL_TIMEOUT_MS;

    if (ctx.hasUI) {
      ctx.ui.notify(
        `[pi-schedule] running shell "${job.name}": ${command}`,
        "info",
      );
    }

    const execResult = await this.opts.pi.exec(resolveShell(), ["-lc", command], {
      cwd,
      timeout: timeoutMs,
    });

    const lastShell: ShellRunResult = {
      ok: execResult.code === 0 && execResult.killed !== true,
      command,
      cwd,
      timeoutMs,
      code: execResult.code,
      killed: Boolean(execResult.killed),
      stdout: truncateOutput(execResult.stdout),
      stderr: truncateOutput(execResult.stderr),
    };

    this.opts.pi.sendMessage?.(
      {
        customType: "pi-schedule",
        content: `Shell "${job.name}" exit ${lastShell.code}${lastShell.killed ? " (killed)" : ""}: ${command}`,
        display: true,
        details: { jobId: job.id, action: "shell", runId: opts.runId, result: lastShell },
      },
      { triggerTurn: false },
    );

    let wokeAgent = false;
    if (shouldWakeForShell(job, lastShell)) {
      const instruction = selectShellFollowUp(job, lastShell);
      if (instruction) {
        const body = buildShellFollowUpPrompt({
          job,
          runId: opts.runId,
          source: opts.source,
          forced: opts.forced,
          result: lastShell,
          instruction,
        });
        await this.sendAgentMessage(body, ctx, opts.deliverAs);
        wokeAgent = true;
      }
    }

    const detail = `shell exit=${lastShell.code}${lastShell.killed ? " killed" : ""}${wokeAgent ? " woke" : ""}`;
    return { detail, wokeAgent, lastShell };
  }

  private async processOne(
    ctx: ExtensionContext,
    job: ScheduledJob,
    opts: {
      source: FireSource;
      forced: boolean;
      deliverAs?: "followUp" | "steer";
      allowFire: boolean;
    },
  ): Promise<ScheduledJob | null> {
    const at = this.now();
    const startedAt = at.toISOString();
    const runId = newRunId();
    const key = opts.forced
      ? `${job.id}:force:${runId}`
      : idempotencyKeyFor(job);

    const tier = job.tier ?? DEFAULT_TIER;
    const missedWindow = job.missedWindow ?? DEFAULT_MISSED_WINDOW;
    const action: JobAction = job.action ?? DEFAULT_JOB_ACTION;

    // Pre-lock idempotency (cheap).
    if (!opts.forced && this.alreadyDelivered(job, key)) {
      const advanced = this.opts.store.markAttempt(job, at, "skipped", {
        error: "idempotent_replay",
        idempotencyKey: key,
      });
      this.recordBestEffort({
        runId,
        jobId: job.id,
        jobName: job.name,
        scope: job.scope,
        projectPath: job.projectPath,
        idempotencyKey: key,
        source: opts.source,
        status: "skipped",
        startedAt,
        endedAt: this.now().toISOString(),
        detail: "idempotent_replay",
        tier,
        missedWindow,
        action,
      });
      return advanced;
    }

    if (!opts.forced) {
      const decision = decideDue(job, at, this.tickMs);
      if (decision.action === "skip") {
        const advanced = this.opts.store.markAttempt(job, at, "skipped", {
          error: decision.reason,
          idempotencyKey: key,
        });
        this.recordBestEffort({
          runId,
          jobId: job.id,
          jobName: job.name,
          scope: job.scope,
          projectPath: job.projectPath,
          idempotencyKey: key,
          source: opts.source,
          status: "skipped",
          startedAt,
          endedAt: this.now().toISOString(),
          detail: decision.reason,
          tier,
          missedWindow,
          action,
        });
        return advanced;
      }
    }

    // Over-cap: stay due, do NOT write ledger spam (busy flood).
    if (!opts.allowFire) {
      return null;
    }

    const handle = this.locks.tryAcquire(job.id);
    if (!handle) {
      // Locked: do not advance; avoid ledger spam on every tick — silent retry.
      return null;
    }

    try {
      // Re-check after lock (check-then-act fix).
      const fresh = this.opts.store.get(job.id, this.cwd) ?? job;
      const freshKey = opts.forced ? key : idempotencyKeyFor(fresh);
      const freshAction: JobAction = fresh.action ?? DEFAULT_JOB_ACTION;
      if (!opts.forced && this.alreadyDelivered(fresh, freshKey)) {
        const advanced = this.opts.store.markAttempt(fresh, at, "skipped", {
          error: "idempotent_replay_post_lock",
          idempotencyKey: freshKey,
        });
        this.recordBestEffort({
          runId,
          jobId: fresh.id,
          jobName: fresh.name,
          scope: fresh.scope,
          projectPath: fresh.projectPath,
          idempotencyKey: freshKey,
          source: opts.source,
          status: "skipped",
          startedAt,
          endedAt: this.now().toISOString(),
          detail: "idempotent_replay_post_lock",
          tier: fresh.tier ?? tier,
          missedWindow: fresh.missedWindow ?? missedWindow,
          action: freshAction,
        });
        return advanced;
      }

      const delivery = await this.deliver(ctx, fresh, {
        runId,
        source: opts.source,
        forced: opts.forced,
        deliverAs: opts.deliverAs,
      });

      // Structural tier enforcement only when an agent turn was started.
      if (delivery.wokeAgent) {
        this.privilege.enter(fresh.tier ?? tier);
      }

      // Advance store FIRST (durable). Ledger is best-effort and must not
      // prevent nextRunAt advancement if runs.jsonl is unwritable.
      const updated = this.opts.store.markAttempt(fresh, at, "ok", {
        idempotencyKey: opts.forced ? key : freshKey,
        lastShell: delivery.lastShell,
      });
      const term = terminalReason(updated, updated.runCount);
      const finalJob = term
        ? this.opts.store.terminate(updated, term, at)
        : updated;
      this.recordBestEffort({
        runId,
        jobId: fresh.id,
        jobName: fresh.name,
        scope: fresh.scope,
        projectPath: fresh.projectPath,
        idempotencyKey: opts.forced ? key : freshKey,
        source: opts.source,
        status: "delivered",
        startedAt,
        endedAt: this.now().toISOString(),
        detail:
          delivery.detail + (term ? ` terminated:${term}` : ""),
        tier: fresh.tier ?? tier,
        missedWindow: fresh.missedWindow ?? missedWindow,
        action: freshAction,
      });
      return finalJob;
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      if (ctx.hasUI) {
        ctx.ui.notify(
          `[pi-schedule] failed to fire "${job.name}": ${message}`,
          "error",
        );
      } else {
        console.error(`[pi-schedule] failed to fire "${job.name}": ${message}`);
      }
      // Advance on error so we don't hot-loop a broken delivery path. Re-read
      // the freshest job inside the lock so a tier/schedule/action change made
      // between the candidate-read and the fire is reflected in both the
      // advance and the forensic ledger (the success path already uses
      // `fresh`).
      const errorSubject = this.opts.store.get(job.id, this.cwd) ?? job;
      const errorKey = opts.forced ? key : idempotencyKeyFor(errorSubject);
      const updated = this.opts.store.markAttempt(errorSubject, at, "error", {
        error: message,
        idempotencyKey: errorKey,
      });
      const term = terminalReason(updated, updated.runCount);
      const finalJob = term
        ? this.opts.store.terminate(updated, term, at)
        : updated;
      this.recordBestEffort({
        runId,
        jobId: errorSubject.id,
        jobName: errorSubject.name,
        scope: errorSubject.scope,
        projectPath: errorSubject.projectPath,
        idempotencyKey: errorKey,
        source: opts.source,
        status: "error",
        startedAt,
        endedAt: this.now().toISOString(),
        detail: message + (term ? ` terminated:${term}` : ""),
        tier: errorSubject.tier ?? tier,
        missedWindow: errorSubject.missedWindow ?? missedWindow,
        action: errorSubject.action ?? action,
      });
      return finalJob;
    } finally {
      handle.release();
    }
  }

  private startTicker(ctx: ExtensionContext): void {
    this.stopTicker();
    this.timer = setInterval(() => {
      void this.fireDue(ctx, { source: "tick" });
    }, this.tickMs);
    this.timer.unref?.();
  }

  private stopTicker(): void {
    if (this.timer) {
      clearInterval(this.timer);
      this.timer = null;
    }
  }
}