/**
 * Workflow run state persistence for pause/resume support.
 */

import {
  existsSync,
  mkdirSync,
  readdirSync,
  readFileSync,
  renameSync,
  rmSync,
  unlinkSync,
  writeFileSync,
} from "node:fs";
import { basename, dirname, join } from "node:path";
import type { AgentContextWindowStats } from "./agent.js";
import type { AgentHistoryEntry } from "./agent-history.js";
import type { ConductorRunStatus } from "./conductor-types.js";
import type { WorkflowErrorCode } from "./errors.js";
import type { FoundationHostCapability } from "./foundation-host.js";
import type { HarnessSelection } from "./harness-selector.js";
import { parseHarnessSelection, serializeHarnessSelection } from "./harness-selector.js";
import type { JournalEntry, WorkflowRunOptions } from "./workflow.js";
import { workflowProjectPaths } from "./workflow-paths.js";

export type RunStatus = "pending" | "running" | "paused" | "completed" | "failed" | "aborted";

/**
 * Operator/model control actions on a run (issue #136 §1). Each carries an
 * authorization reason and an originator, persisted so the navigator and a
 * cold start can show who did what and why. The engine `RunStatus` is the
 * source of truth for lifecycle; this is the control-surface audit trail.
 */
export type ControlAction = "pause" | "stop" | "resume" | "status" | "steer";

/** Who authorized a control action. `operator` = user/TUI; `system` = the
 *  manager's usage-limit auto-resume or stale-run reconciliation. */
export type ControlAuthorizer = "operator" | "system";

/** A single control-action event, persisted into a bounded audit log. */
export interface ControlActionEvent {
  action: ControlAction;
  at: string;
  authorizer: ControlAuthorizer;
  /** Why the action was taken, e.g. "usage_limit", "operator_request", "cap_exceeded". */
  reason?: string;
  /** Provider reset hint captured for usage-limit pauses (verbatim). */
  resetHint?: string;
}

export interface PersistedAgentState {
  id: number;
  label: string;
  phase?: string;
  prompt: string;
  status: "queued" | "running" | "done" | "error" | "skipped";
  result?: unknown;
  error?: string;
  errorCode?: WorkflowErrorCode;
  recoverable?: boolean;
  history?: AgentHistoryEntry[];
  startedAt?: string;
  endedAt?: string;
  /** Tokens used by this agent, when known. */
  tokens?: number;
  /** Context-window occupancy stats for this agent, when known. */
  contextWindow?: AgentContextWindowStats;
  /** The model this agent ran on (provider/id), when known. */
  model?: string;
}

export interface PersistedRunState {
  runId: string;
  workflowName: string;
  script: string;
  args?: unknown;
  /** The pi session this run belongs to. Runs persist on disk across sessions but
   * the navigator shows only the current session's runs (undefined = legacy/global). */
  sessionId?: string;
  status: RunStatus;
  /** Optional conductor-level semantic status, layered on top of the engine
   *  `status` above. Older persisted runs may omit this; loaders must not reject
   *  runs that lack it. When present it is round-tripped verbatim on save/load. */
  semanticStatus?: ConductorRunStatus;
  /** Why a paused run is paused (e.g. "usage_limit" when a provider quota was hit). */
  pauseReason?: string;
  /** Provider reset hint for a usage-limit pause, e.g. "Resets in ~3h" (verbatim). */
  resetHint?: string;
  /**
   * Bounded usage-limit resume accounting (issue #136 §5). Increments on each
   * resume of a usage_limit-paused run; when it exceeds `maxResumeAttempts` the
   * run settles into a terminal `failed`/`needs-human` state with an actionable
   * reason instead of silently looping. Absent on old runs (treated as 0).
   */
  resumeAttempts?: number;
  /** The cap this run was started/resumed under. Absent -> runtime default. */
  maxResumeAttempts?: number;
  /** ISO timestamp of the last resume, so an operator can see how long a run has
   *  been cycling on a usage limit. */
  lastResumeAt?: string;
  /**
   * Bounded control-action audit log (issue #136 §1). Older persisted runs omit
   * this; loaders must not reject runs that lack it. Kept small (last N events)
   * so it never dominates the persisted record.
   */
  controlActions?: ControlActionEvent[];
  /**
   * Terminal exhaustion marker set when usage-limit resume attempts exceeded the
   * cap. The run is `failed` with `pauseReason: "usage_exhausted"` and a
   * `needs-human` semantic status, so the navigator shows an actionable terminal
   * state instead of a silent infinite-retry loop.
   */
  usageExhausted?: boolean;
  /** Snapshot of the routing/context/tool-policy captured at run start, so resume
   *  keeps the original snapshot unless an explicit change invalidates the right
   *  suffix (issue #136 §2). The harness selection already carries routing; this
   *  is the steering override applied after start. */
  steeringSnapshot?: { contextMode?: string; harnessType?: string; harnessConfig?: string };
  phases: string[];
  currentPhase?: string;
  agents: PersistedAgentState[];
  logs: string[];
  result?: unknown;
  startedAt: string;
  updatedAt: string;
  completedAt?: string;
  durationMs?: number;
  /** Absolute path to this persisted run-state JSON, when recorded by WorkflowManager. */
  runStatePath?: string;
  /** Effective run-wide wall-clock timeout (ms) captured at start, so resume
   *  keeps the original explicit/settings value. null disables the timeout;
   *  absent (old runs) means the runtime default still applies. */
  workflowTimeoutMs?: number | null;
  /** Effective run-level hard per-agent context cap captured at start. */
  agentMaxContextTokens?: number | null;
  /** Effective run-level context reserve override captured at start. */
  agentContextReserveTokens?: number | null;
  /** Effective run-level compaction policy captured at start. */
  compactionPolicy?: WorkflowRunOptions["compactionPolicy"];
  /** Effective run-level loop-guard policy captured at start/resume. */
  loopGuard?: WorkflowRunOptions["loopGuard"];
  /** Snapshot of the harness selection detected at run start.
   *
   *  Persisted as a canonical serialized string (via serializeHarnessSelection)
   *  so the on-disk snapshot is deterministic and resume can reuse it instead of
   *  re-running detection. The field also tolerates a plain `HarnessSelection`
   *  object form for backward compatibility with older persisted runs. Either
   *  form is validated back through parseHarnessSelection() on load.
   *  Optional — older persisted runs may omit it, in which case resume falls
   *  back to a fresh selectHarness() call. */
  harnessSelection?: HarnessSelection | string;
  tokenUsage?: {
    input: number;
    output: number;
    total: number;
    cost?: number;
    cacheRead?: number;
    cacheWrite?: number;
  };
  /** Cached agent results and replay metadata, keyed by deterministic call index. */
  journal?: JournalEntry[];
  /**
   * Run-level isolation worktree (when the run was launched with
   * `isolation: { worktree: true }`/`worktreeRequired`). Persisted so a paused
   * run keeps its worktree across a resume (edits live in the worktree, not the
   * primary checkout); resume reuses it via `reuseWorktree`.
   */
  worktree?: { cwd: string; branch?: string; repoRoot?: string; workspaceId?: string };
  /** Persisted herdr pane id for a pane-spawn run, so resume can recreate the
   *  pane handle and keep driving the pane's lifecycle/finalization. Absent on
   *  non-pane-spawn and older runs. */
  paneId?: string;
  /** Package-authorized host callbacks captured when the saved command started. */
  hostCapabilities?: FoundationHostCapability[];
  /** Stable package execution-policy identity used by resume hashes. */
  hostCapabilityPolicyKey?: string;
  /** Initial isolated-worktree HEAD used to validate committed and uncommitted paths. */
  hostEditScopeBaseRef?: string;
}

export interface RunPersistence {
  /** Save current run state. */
  save(state: PersistedRunState): void;
  /** Load a persisted run by ID. */
  load(runId: string): PersistedRunState | null;
  /** List all persisted runs. */
  list(): PersistedRunState[];
  /** Delete a persisted run. */
  delete(runId: string): boolean;
  /**
   * Acquire an exclusive cross-process lease for a run. Returns null when another
   * live process owns the run; stale/corrupt lock files are removed and retried.
   */
  acquireRunLease(runId: string): RunLease | null;
  /** Release a lease previously returned by acquireRunLease(). */
  releaseRunLease(lease: RunLease): void;
  /** Get runs directory path. */
  getRunsDir(): string;
  /**
   * Delete completed runs exceeding the retention policy. Only runs whose status
   * is exactly "completed" are ever considered; active/paused/failed/aborted
   * runs are never removed, and artifact files still referenced by a surviving
   * run are never deleted. Returns the run IDs that were removed.
   */
  pruneCompletedRuns(config?: RunStateRetentionConfig): string[];
}

export interface RunLease {
  runId: string;
  token: string;
}

interface LockFile {
  runId: string;
  runPath: string;
  pid: number;
  startedAt: string;
  token: string;
}

/**
 * On-disk reference to a payload that was spilled out of the run-state JSON to
 * keep the persisted file within the documented size bound. Only the save path
 * ever writes these markers; {@link rehydrateState} resolves them back into the
 * original value on load so every consumer of `load()`/`list()` sees the real
 * payload. The marker uses a namespaced discriminator (`__spilledArtifact`) and
 * an exact key set so it cannot be mistaken for a legitimate workflow result.
 */
export interface SpilledArtifactRef {
  readonly __spilledArtifact: true;
  /** Path relative to the run's runsDir, e.g. `<runId>/artifacts/result-3.json`. */
  readonly path: string;
  /** Original byte size of the pretty-JSON payload, for auditing/regression. */
  readonly bytes: number;
  /** Payload class: "history" | "result". */
  readonly kind: ArtifactKind;
}

export type ArtifactKind = "history" | "result";

function isSpilledArtifactRef(value: unknown): value is SpilledArtifactRef {
  if (typeof value !== "object" || value === null) return false;
  const keys = Object.keys(value);
  if (keys.length !== 4) return false;
  const v = value as Record<string, unknown>;
  return (
    v.__spilledArtifact === true &&
    typeof v.path === "string" &&
    typeof v.bytes === "number" &&
    (v.kind === "history" || v.kind === "result")
  );
}

/**
 * Retention/cleanup policy for persisted run state. Only COMPLETED runs are ever
 * eligible for automatic cleanup; active/paused/failed/aborted runs and any
 * artifact still referenced by a surviving run are never removed.
 */
export interface RunStateRetentionConfig {
  /**
   * Per-payload byte bound (pretty-JSON, the on-disk form). A single
   * agent-history, journal-result, or top-level result payload larger than this
   * is spilled to an artifact file and replaced inline by a {@link SpilledArtifactRef}.
   * Identity fields (journal hash/index/label/usage/model/timestamps) are never
   * spilled, so deterministic resume longest-prefix semantics are unaffected.
   * Default {@link DEFAULT_RUN_STATE_PAYLOAD_BOUND_BYTES}.
   */
  payloadBoundBytes?: number;
  /** Remove completed runs whose `updatedAt` is older than this (ms). 0 = no age limit. */
  completedMaxAgeMs?: number;
  /** Keep at most this many most-recent completed runs; older ones are pruned. 0 = no count limit. */
  completedMaxCount?: number;
}

/** Default per-payload spill bound: 32 KiB (pretty-JSON bytes). */
export const DEFAULT_RUN_STATE_PAYLOAD_BOUND_BYTES = 32 * 1024;

/**
 * Documented target upper bound for a completed run's persisted JSON once
 * history is de-duplicated and large payloads spill to artifacts. Spilled
 * artifact files are lazy-loaded and not counted toward this bound.
 */
export const TARGET_RUN_STATE_JSON_BOUND_BYTES = 256 * 1024;

const ARTIFACTS_SUBDIR = "artifacts";

/** Resolve a payload-bound override to a positive integer, falling back to the default. */
export function normalizePayloadBound(value: unknown): number {
  if (typeof value === "number" && Number.isFinite(value) && value > 0) return Math.floor(value);
  return DEFAULT_RUN_STATE_PAYLOAD_BOUND_BYTES;
}

/** Parse an ISO timestamp to epoch ms, or null if unparseable (used by retention). */
function safeEpochMs(iso: string | undefined): number | null {
  if (typeof iso !== "string") return null;
  const ms = Date.parse(iso);
  return Number.isFinite(ms) ? ms : null;
}

/**
 * Persisted run IDs become filenames and lock names, so keep them deliberately
 * boring: generated ids already use lowercase base36 + hyphen. Rejecting dots,
 * slashes, backslashes, controls, and absolute-looking values closes traversal
 * through every run-state/lease/resume/delete path.
 */
export const RUN_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/;

export function isValidRunId(runId: unknown): runId is string {
  return typeof runId === "string" && RUN_ID_PATTERN.test(runId);
}

export function assertValidRunId(runId: unknown): asserts runId is string {
  if (!isValidRunId(runId)) {
    throw new Error(`Invalid workflow runId: ${JSON.stringify(runId)}`);
  }
}

/**
 * Filesystem operations used by run persistence.
 * Exposed for testing – pass overrides to inject mock implementations.
 */
export type FsLayer = {
  existsSync: typeof existsSync;
  mkdirSync: typeof mkdirSync;
  readdirSync: typeof readdirSync;
  readFileSync: typeof readFileSync;
  renameSync: typeof renameSync;
  unlinkSync: typeof unlinkSync;
  writeFileSync: typeof writeFileSync;
  rmSync?: typeof rmSync;
};

/** Shared formula for a run-state JSON file path: runsDir/runId.json.
 *  Used by both RunPersistence (primaryRunPath) and WorkflowManager.runStatePathFor
 *  so the <recovery> log link can never drift from where the run state is actually
 *  written. */
export function runStateJsonPath(runsDir: string, runId: string): string {
  assertValidRunId(runId);
  return join(runsDir, `${runId}.json`);
}

// ─── run-state size bounding: spill / rehydrate / compact ─────────────────────

/** Pretty-JSON byte length of a value (the on-disk form used by save()). */
export function persistedByteLength(value: unknown): number {
  return Buffer.byteLength(JSON.stringify(value, null, 2), "utf-8");
}

interface FsWriteOps {
  existsSync: typeof existsSync;
  mkdirSync: typeof mkdirSync;
  writeFileSync: typeof writeFileSync;
  renameSync: typeof renameSync;
}

interface FsReadOps {
  existsSync: typeof existsSync;
  readFileSync: typeof readFileSync;
}

/**
 * Relative path (from runsDir) of an artifact for a given run/payload.
 * Shape: `<runId>/artifacts/<kind>-<index>.json`.
 */
export function artifactRelPath(runId: string, kind: ArtifactKind, index: number): string {
  assertValidRunId(runId);
  return join(runId, ARTIFACTS_SUBDIR, `${kind}-${index}.json`);
}

/**
 * If `value` exceeds `bound` pretty-JSON bytes, write it atomically to an
 * artifact file and return a {@link SpilledArtifactRef}; otherwise return the
 * value unchanged. `index` disambiguates multiple payloads of the same kind.
 * Spill failures are non-fatal: on error the original value is kept inline so
 * the run state stays loadable (size bounding is best-effort, like the .bak).
 */
function spillIfTooLarge(
  value: unknown,
  bound: number,
  runId: string,
  kind: ArtifactKind,
  index: number,
  runsDir: string,
  fs: FsWriteOps,
): unknown {
  if (value === undefined) return undefined;
  const size = persistedByteLength(value);
  if (size <= bound) return value;
  const rel = artifactRelPath(runId, kind, index);
  const abs = join(runsDir, rel);
  try {
    fs.mkdirSync(dirname(abs), { recursive: true });
    // Atomic write: tmp + rename on the same filesystem.
    const tmp = `${abs}.tmp`;
    fs.writeFileSync(tmp, JSON.stringify(value, null, 2));
    fs.renameSync(tmp, abs);
  } catch {
    // Keep inline on spill failure; the run state still loads.
    return value;
  }
  return { __spilledArtifact: true, path: rel, bytes: size, kind } satisfies SpilledArtifactRef;
}

/**
 * Resolve a {@link SpilledArtifactRef} back into its original value by reading
 * the artifact file. On any read/parse failure the ref is returned as-is so the
 * caller still has the audit metadata (best-effort, never throws).
 */
function resolveArtifactRef(ref: SpilledArtifactRef, runsDir: string, fs: FsReadOps): unknown {
  // The ref path is relative to runsDir and confined under <runId>/artifacts.
  const abs = join(runsDir, ref.path);
  try {
    if (!fs.existsSync(abs)) return ref;
    return JSON.parse(fs.readFileSync(abs, "utf-8"));
  } catch {
    return ref;
  }
}

/**
 * Walk a loaded {@link PersistedRunState} and resolve every spilled artifact
 * reference back into its in-memory value, mutating in place. This is the single
 * rehydration point: both `load()` and `list()` route through it, so every
 * consumer (resume, telemetry, UI, /workflows) sees the original payloads.
 *
 * It also fills `agents[].history` from the journal for runs saved after the
 * history de-duplication (the journal is the on-disk source of truth; agent
 * rows no longer carry history). Legacy runs that still have `agents[].history`
 * are left untouched.
 */
function rehydrateState(state: PersistedRunState, runsDir: string, fs: FsReadOps): PersistedRunState {
  // Top-level result.
  if (isSpilledArtifactRef(state.result)) {
    state.result = resolveArtifactRef(state.result, runsDir, fs);
  }
  // Journal entries: result + history.
  if (Array.isArray(state.journal)) {
    for (const entry of state.journal) {
      if (isSpilledArtifactRef(entry.result)) entry.result = resolveArtifactRef(entry.result, runsDir, fs);
      if (isSpilledArtifactRef(entry.history))
        entry.history = resolveArtifactRef(entry.history, runsDir, fs) as AgentHistoryEntry[];
    }
  }
  // Agents: history may be a spilled ref — resolve it back into the in-memory
  // value. (History lives on-disk in agents[]; the journal holds no history for
  // new saves, and hydrateJournalHistory() in workflow-manager copies agent ->
  // journal history at resume time for entries that lack it.)
  if (Array.isArray(state.agents)) {
    for (const agent of state.agents) {
      if (isSpilledArtifactRef(agent.history)) {
        agent.history = resolveArtifactRef(agent.history, runsDir, fs) as AgentHistoryEntry[];
      }
    }
  }
  return state;
}

/**
 * Copy `journal[].history` onto matching `agents[].history` when an agent lacks
 * history, mirroring {@link hydrateJournalHistory} in workflow-manager but
 * operating on the loaded state so on-disk consumers (telemetry, UI agent
 * detail) see history without the duplicate copy on disk.
 *
 * Pairing is by label + done status, in journal/agent order, so a resumed run
 * with replayed (non-live) agents still recovers history for display/telemetry.
 * Runs whose journal lacks history (pre-change shape) keep whatever agent
 * history they already have — fully backward compatible.
 */
export function hydrateAgentHistoryFromJournal(state: PersistedRunState): void {
  const journal = state.journal;
  const agents = state.agents;
  if (!Array.isArray(journal) || !Array.isArray(agents)) return;
  const ordered = [...journal].sort((a, b) => a.index - b.index);
  let nextAgentIndex = 0;
  for (const entry of ordered) {
    const history = entry.history;
    if (!Array.isArray(history) || history.length === 0) continue;
    // Skip checkpoint entries (no label/model/usage/tokens).
    const isAgentEntry = Boolean(entry.label || entry.model || entry.usage || entry.tokens !== undefined);
    if (!isAgentEntry) continue;
    const label = entry.label;
    // Find the next done agent with matching label that lacks history.
    let found = -1;
    for (let i = nextAgentIndex; i < agents.length; i++) {
      const a = agents[i];
      if (a.status !== "done") continue;
      if (label && a.label !== label) continue;
      if (a.history && a.history.length > 0) continue; // already has history
      found = i;
      break;
    }
    if (found < 0) {
      // Fall back to any later done agent without history (label-less entries).
      for (let i = nextAgentIndex; i < agents.length; i++) {
        const a = agents[i];
        if (a.status !== "done") continue;
        if (a.history && a.history.length > 0) continue;
        found = i;
        break;
      }
    }
    if (found < 0) continue;
    agents[found].history = history;
    nextAgentIndex = found + 1;
  }
}

/**
 * Produce a size-bounded copy of `state` for persistence: drop the duplicated
 * `agents[].history` (the journal is the on-disk source of truth) and spill any
 * oversized journal-result / journal-history / top-level-result payloads to
 * artifact files under `<runsDir>/<runId>/artifacts/`. Returns the compacted
 * state plus the set of artifact relative paths written.
 *
 * Identity fields (journal index/hash/label/usage/model/timestamps, agent
 * status/tokens/contextWindow, run status/phases/logs) are NEVER spilled, so
 * deterministic resume longest-prefix semantics and recovery links are intact.
 */
export function compactStateForSave(
  state: PersistedRunState,
  bound: number,
  runsDir: string,
  fs: FsWriteOps,
): { compact: PersistedRunState; artifacts: string[] } {
  const artifacts: string[] = [];
  const compact: PersistedRunState = { ...state };

  // Top-level result.
  const resultOut = spillIfTooLarge(state.result, bound, state.runId, "result", 0, runsDir, fs);
  compact.result = resultOut;
  if (isSpilledArtifactRef(resultOut)) artifacts.push(resultOut.path);

  // Journal: spill oversized result per entry. History is NOT persisted in the
  // journal — it lives once, in agents[], and hydrateJournalHistory() (in
  // workflow-manager) copies agent -> journal history on resume for the entries
  // that lack it. This keeps a single on-disk history copy while preserving the
  // legacy/recovery fallback (older runs with history only in agents still work).
  if (Array.isArray(state.journal)) {
    compact.journal = state.journal.map((entry, i) => {
      const out: JournalEntry = { ...entry };
      const r = spillIfTooLarge(entry.result, bound, state.runId, "result", i + 1, runsDir, fs);
      out.result = r;
      if (isSpilledArtifactRef(r)) artifacts.push(r.path);
      // Drop the duplicated journal history (agents[] holds it).
      out.history = undefined;
      return out;
    });
  }

  // Agents: keep history (the single on-disk source), spilling oversized copies
  // to artifacts so the inline JSON stays bounded. Every other field is kept so
  // UI/telemetry/resume identifiers are unchanged.
  if (Array.isArray(state.agents)) {
    compact.agents = state.agents.map((agent, i) => {
      const out = { ...agent } as PersistedAgentState;
      const h = spillIfTooLarge(agent.history, bound, state.runId, "history", i + 1, runsDir, fs);
      out.history = h as PersistedAgentState["history"];
      if (isSpilledArtifactRef(h)) artifacts.push(h.path);
      return out;
    });
  }

  return { compact, artifacts };
}

export function createRunPersistence(
  cwd: string,
  fsOverride?: Partial<FsLayer>,
  retention?: RunStateRetentionConfig,
): RunPersistence {
  const _existsSync = fsOverride?.existsSync ?? existsSync;
  const _mkdirSync = fsOverride?.mkdirSync ?? mkdirSync;
  const _readdirSync = fsOverride?.readdirSync ?? readdirSync;
  const _readFileSync = fsOverride?.readFileSync ?? readFileSync;
  const _renameSync = fsOverride?.renameSync ?? renameSync;
  const _unlinkSync = fsOverride?.unlinkSync ?? unlinkSync;
  const _writeFileSync = fsOverride?.writeFileSync ?? writeFileSync;
  const _rmSync = fsOverride?.rmSync ?? rmSync;

  const payloadBound = normalizePayloadBound(retention?.payloadBoundBytes);

  const paths = workflowProjectPaths(cwd);
  const runsDir = paths.runsDir;
  const legacyRunsDir = paths.legacyRunsDir;

  const ensureDir = () => {
    if (!_existsSync(runsDir)) {
      _mkdirSync(runsDir, { recursive: true });
    }
  };

  const runPath = runStateJsonPath; // shared formula (also used by WorkflowManager.runStatePathFor)
  const primaryRunPath = (runId: string) => runPath(runsDir, runId);
  const legacyRunPath = (runId: string) => runPath(legacyRunsDir, runId);
  const lockPath = (dir: string, runId: string) => {
    assertValidRunId(runId);
    return join(dir, `${runId}.lock`);
  };
  const primaryLockPath = (runId: string) => lockPath(runsDir, runId);
  const legacyLockPath = (runId: string) => lockPath(legacyRunsDir, runId);
  const candidateRunPaths = (runId: string) => [primaryRunPath(runId), legacyRunPath(runId)];

  const pidIsAlive = (pid: number): boolean => {
    if (!Number.isInteger(pid) || pid <= 0) return false;
    try {
      process.kill(pid, 0);
      return true;
    } catch (err) {
      if ((err as { code?: string }).code === "EPERM") return true;
      return false;
    }
  };

  const readLockAt = (path: string): LockFile | null => {
    try {
      return JSON.parse(_readFileSync(path, "utf-8")) as LockFile;
    } catch {
      return null;
    }
  };

  const readLock = (runId: string): LockFile | null =>
    isValidRunId(runId) ? readLockAt(primaryLockPath(runId)) : null;

  const removeStaleLegacyLock = (runId: string): boolean => {
    if (!isValidRunId(runId)) return false;
    const lock = legacyLockPath(runId);
    const existing = readLockAt(lock);
    if (existing?.runId === runId && pidIsAlive(existing.pid)) return false;
    try {
      if (_existsSync(lock)) _unlinkSync(lock);
    } catch {
      return false;
    }
    return true;
  };

  const readStateFile = (path: string, expectedRunId?: string): PersistedRunState | null => {
    try {
      const state = JSON.parse(_readFileSync(path, "utf-8")) as PersistedRunState;
      const fileRunId = basename(path).replace(/\.json(?:\.bak)?$/, "");
      if (!isValidRunId(state.runId)) return null;
      if (expectedRunId !== undefined && state.runId !== expectedRunId) return null;
      if (isValidRunId(fileRunId) && state.runId !== fileRunId) return null;
      // Resolve any spilled artifact references back into the in-memory values so
      // every consumer of load()/list() sees the original payloads. Also fills
      // agents[].history from the journal for runs saved with history de-duped.
      return rehydrateState(state, runsDir, { existsSync: _existsSync, readFileSync: _readFileSync });
    } catch {
      return null;
    }
  };

  /** Best-effort recursive removal of a run's artifact directory. */
  const removeArtifacts = (runId: string): void => {
    if (!isValidRunId(runId)) return;
    // Only the primary runs layout uses artifact dirs; legacy layout never spilled.
    const dir = join(runsDir, runId, ARTIFACTS_SUBDIR);
    try {
      if (_existsSync(dir)) _rmSync?.(dir, { recursive: true, force: true });
    } catch {
      // best-effort
    }
  };

  return {
    save(state: PersistedRunState) {
      assertValidRunId(state.runId);
      ensureDir();
      state.updatedAt = new Date().toISOString();
      const path = primaryRunPath(state.runId);
      // Compact for save: drop duplicated agent history and spill oversized
      // payloads to artifact files. The on-disk JSON stays within the documented
      // bound while load() rehydrates the original values transparently.
      const { compact } = compactStateForSave(state, payloadBound, runsDir, {
        existsSync: _existsSync,
        mkdirSync: _mkdirSync,
        writeFileSync: _writeFileSync,
        renameSync: _renameSync,
      });
      const json = JSON.stringify(compact, null, 2);
      // Atomic write: a crash mid-write can't corrupt the live file (tmp+rename is
      // atomic on the same filesystem). A .bak from the previous good save is the
      // recovery fallback if the primary is somehow truncated.
      _writeFileSync(`${path}.tmp`, json);
      _renameSync(`${path}.tmp`, path);
      try {
        _writeFileSync(`${path}.bak`, json);
      } catch {
        // backup is best-effort; the primary write already succeeded
      }
    },

    load(runId: string): PersistedRunState | null {
      if (!isValidRunId(runId)) return null;
      // Try the primary, then the .bak — so a corrupt primary doesn't lose the run.
      for (const path of candidateRunPaths(runId)) {
        for (const candidate of [path, `${path}.bak`]) {
          if (!_existsSync(candidate)) continue;
          const state = readStateFile(candidate, runId);
          if (state) return state;
        }
      }
      return null;
    },

    list(): PersistedRunState[] {
      const byRunId = new Map<string, PersistedRunState>();
      for (const dir of [runsDir, legacyRunsDir]) {
        try {
          if (!_existsSync(dir)) continue;
          const files = _readdirSync(dir).filter((f) => f.endsWith(".json"));
          for (const file of files) {
            const state = readStateFile(join(dir, file));
            if (state && !byRunId.has(state.runId)) byRunId.set(state.runId, state);
          }
        } catch {
          // Skip unreadable directories; another storage location may still work.
        }
      }
      return [...byRunId.values()].sort((a, b) => new Date(b.updatedAt).getTime() - new Date(a.updatedAt).getTime());
    },

    delete(runId: string): boolean {
      if (!isValidRunId(runId)) return false;
      let deleted = false;
      try {
        for (const path of candidateRunPaths(runId)) {
          const dir = path === primaryRunPath(runId) ? runsDir : legacyRunsDir;
          // Best-effort cleanup of the sidecar files alongside the primary.
          for (const sidecar of [`${path}.bak`, `${path}.tmp`, lockPath(dir, runId)]) {
            try {
              if (_existsSync(sidecar)) _unlinkSync(sidecar);
            } catch {
              // ignore sidecar cleanup failures
            }
          }
          try {
            if (_existsSync(path)) {
              _unlinkSync(path);
              deleted = true;
            }
          } catch {
            // ignore per-file cleanup failures
          }
        }
        // Remove any spilled artifact directory for the run (primary layout only).
        if (deleted) removeArtifacts(runId);
        return deleted;
      } catch {
        return deleted;
      }
    },

    acquireRunLease(runId: string): RunLease | null {
      if (!isValidRunId(runId)) return null;
      ensureDir();
      const path = primaryRunPath(runId);
      const lock = primaryLockPath(runId);
      if (!removeStaleLegacyLock(runId)) return null;
      for (let attempt = 0; attempt < 2; attempt++) {
        const token = `${process.pid}-${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`;
        const payload: LockFile = {
          runId,
          runPath: path,
          pid: process.pid,
          startedAt: new Date().toISOString(),
          token,
        };
        try {
          _writeFileSync(lock, JSON.stringify(payload, null, 2), { flag: "wx" });
          return { runId, token };
        } catch (err) {
          const code = (err as { code?: string }).code;
          if (code !== "EEXIST") throw err;
          const existing = readLock(runId);
          if (existing && existing.runPath === path && pidIsAlive(existing.pid)) {
            return null;
          }
          try {
            _unlinkSync(lock);
          } catch {
            return null;
          }
        }
      }
      return null;
    },

    releaseRunLease(lease: RunLease): void {
      if (!isValidRunId(lease.runId)) return;
      try {
        const existing = readLock(lease.runId);
        if (existing?.token === lease.token) _unlinkSync(primaryLockPath(lease.runId));
      } catch {
        // Best-effort cleanup only.
      }
    },

    getRunsDir(): string {
      return runsDir;
    },

    pruneCompletedRuns(config?: RunStateRetentionConfig): string[] {
      const maxAgeMs =
        typeof config?.completedMaxAgeMs === "number" && config.completedMaxAgeMs > 0 ? config.completedMaxAgeMs : 0;
      const maxCount =
        typeof config?.completedMaxCount === "number" && config.completedMaxCount > 0
          ? Math.floor(config.completedMaxCount)
          : 0;
      if (maxAgeMs === 0 && maxCount === 0) return []; // no policy configured -> no-op

      // List only COMPLETED runs. list() already rehydrates and sorts by
      // updatedAt descending; reuse it but filter strictly on status.
      const completed = this.list().filter((r) => r.status === "completed");
      const now = Date.now();
      const removed: string[] = [];
      // Determine the set of run IDs to keep by count (most-recent first), then
      // additionally drop anything older than the age cutoff. A run is removed
      // only if it is BOTH completed AND past the applicable limit.
      const keepByCount = new Set(maxCount > 0 ? completed.slice(0, maxCount).map((r) => r.runId) : null);
      for (const run of completed) {
        let tooOld = false;
        if (maxAgeMs > 0) {
          const updated = safeEpochMs(run.updatedAt);
          if (updated !== null && now - updated > maxAgeMs) tooOld = true;
        }
        const overCount = maxCount > 0 && !keepByCount.has(run.runId);
        // A run is pruned if it exceeds EITHER limit (intersection of the policy:
        // when both are set, the more restrictive applicable limit wins per run).
        if (tooOld || overCount) {
          if (this.delete(run.runId)) removed.push(run.runId);
        }
      }
      return removed;
    },
  };
}

/**
 * Read a persisted harness-selection snapshot back into a validated
 * `HarnessSelection`, so resume can reuse the snapshot instead of re-running
 * `selectHarness()`.
 *
 * Accepts either stored form:
 *  - canonical serialized string (current writers, via serializeHarnessSelection)
 *  - plain `HarnessSelection` object (legacy/compatible writers)
 *
 * Returns `undefined` when the field is absent (old run) or malformed, so the
 * caller falls back to a fresh `selectHarness()` call. This keeps the load path
 * backward-compatible: a persisted run without the field still loads.
 */
export function loadHarnessSelection(state: PersistedRunState): HarnessSelection | undefined {
  const raw = state.harnessSelection;
  if (raw === undefined) return undefined;
  if (typeof raw === "string") {
    // Canonical serialized string form (current writers). Parse the JSON
    // envelope, then validate the inner object through parseHarnessSelection.
    if (raw.length === 0) return undefined;
    try {
      return parseHarnessSelection(JSON.parse(raw));
    } catch {
      return undefined;
    }
  }
  // Plain object form (legacy/compatible writers): validate directly.
  return parseHarnessSelection(raw);
}

/**
 * Serialize a harness selection for persistence into the run-metadata record.
 * Returns the canonical serialized string form (via serializeHarnessSelection)
 * that round-trips through `loadHarnessSelection()`, or `undefined` when `sel`
 * is `undefined` (so the field stays absent on disk for runs without a snapshot).
 *
 * runWorkflow should assign the result onto `PersistedRunState.harnessSelection`
 * for a freshly-detected selection before `save()`.
 */
export function saveHarnessSelection(sel: HarnessSelection | undefined): string | undefined {
  return sel !== undefined ? serializeHarnessSelection(sel) : undefined;
}

/** Maximum control-action events retained in the persisted audit log. The
 *  log is a bounded ring of the most recent events so it never dominates the
 *  persisted record while still giving an operator a usable history. */
export const MAX_CONTROL_ACTION_LOG = 16;

/**
 * Generate a unique run ID.
 */
export function generateRunId(): string {
  const timestamp = Date.now().toString(36);
  const random = Math.random().toString(36).slice(2, 8);
  return `${timestamp}-${random}`;
}