/**
 * Minions — BullMQ-inspired Postgres-native job queue for GBrain.
 *
 * Usage:
 *   const queue = new MinionQueue(engine);
 *   const job = await queue.add('sync', { full: true });
 *
 *   const worker = new MinionWorker(engine);
 *   worker.register('sync', async (job) => {
 *     await runSync(engine, job.data);
 *     return { pages_synced: 42 };
 *   });
 *   await worker.start();
 */

// --- Status & Type Unions ---

export type MinionJobStatus =
  | 'waiting'
  | 'active'
  | 'completed'
  | 'failed'
  | 'delayed'
  | 'dead'
  | 'cancelled'
  | 'waiting-children'
  | 'paused';

export type BackoffType = 'fixed' | 'exponential';

export type ChildFailPolicy = 'fail_parent' | 'remove_dep' | 'ignore' | 'continue';

// --- Job Record ---

export interface MinionJob {
  id: number;
  name: string;
  queue: string;
  status: MinionJobStatus;
  priority: number;
  data: Record<string, unknown>;

  // Retry
  max_attempts: number;
  attempts_made: number;
  attempts_started: number;
  backoff_type: BackoffType;
  backoff_delay: number;
  backoff_jitter: number;

  // Stall detection
  stalled_counter: number;
  max_stalled: number;
  lock_token: string | null;
  lock_until: Date | null;

  // Scheduling
  delay_until: Date | null;

  // Dependencies
  parent_job_id: number | null;
  on_child_fail: ChildFailPolicy;

  // Token accounting
  tokens_input: number;
  tokens_output: number;
  tokens_cache_read: number;

  // v7: subagent + parity
  depth: number;
  max_children: number | null;
  timeout_ms: number | null;
  timeout_at: Date | null;
  remove_on_complete: boolean;
  remove_on_fail: boolean;
  idempotency_key: string | null;

  // v12: scheduler polish — quiet-hours gate + deterministic stagger
  quiet_hours: Record<string, unknown> | null;
  stagger_key: string | null;

  // Results
  result: Record<string, unknown> | null;
  progress: unknown | null;
  error_text: string | null;
  stacktrace: string[];

  // Timestamps
  created_at: Date;
  started_at: Date | null;
  finished_at: Date | null;
  updated_at: Date;
}

// --- Input Types ---

export interface MinionJobInput {
  name: string;
  data?: Record<string, unknown>;
  queue?: string;
  priority?: number;
  max_attempts?: number;
  backoff_type?: BackoffType;
  backoff_delay?: number;
  backoff_jitter?: number;
  /**
   * Per-job override for how many stall windows are tolerated before the
   * queue dead-letters the job. When omitted, the schema column DEFAULT
   * applies (bumped 1 → 3 in v0.14, now 5 as of v0.13.1's audit). Clamped
   * to [1, 100] on insert. For long-running handlers (LLM loops etc.) that
   * should survive a worker kill mid-run, set max_stalled: 3+.
   */
  max_stalled?: number;
  delay?: number; // ms delay before eligible
  parent_job_id?: number;
  on_child_fail?: ChildFailPolicy;

  // v7: subagent + parity
  /** Cap on live (non-terminal) children of THIS job. NULL/undefined = unlimited. */
  max_children?: number;
  /** Wall-clock per-job deadline in ms. Set on claim → timeout_at. Terminal on expire (no retry). */
  timeout_ms?: number;
  /** DELETE row on successful completion (after token rollup + child_done insert). */
  remove_on_complete?: boolean;
  /** DELETE row on terminal failure (after parent failure hook). */
  remove_on_fail?: boolean;
  /** Override the queue's maxSpawnDepth for THIS submission only. */
  max_spawn_depth?: number;
  /** Global dedup key. Same key returns the existing job, no second row created. */
  idempotency_key?: string;
  /** Submission backpressure: cap waiting jobs with this name before inserting a new row. */
  maxWaiting?: number;

  // v12: scheduler polish
  /**
   * Quiet-hours window evaluated at claim time. Jobs whose current wall-clock
   * falls inside the window are deferred (delay +15m) or skipped per policy.
   * Example: `{start:22,end:7,tz:"America/Los_Angeles",policy:"defer"}`.
   */
  quiet_hours?: { start: number; end: number; tz: string; policy?: 'skip' | 'defer' };
  /**
   * Deterministic stagger key. When multiple jobs share a key (same cron fire),
   * their claim order is decorrelated by hash-based minute-offset. Optional.
   */
  stagger_key?: string;
}

/** Constructor options for MinionQueue (v7). */
export interface MinionQueueOpts {
  /** Max parent→child→grandchild depth. Default 5. Enforced on add() with parent_job_id. */
  maxSpawnDepth?: number;
  /** Max attachment size in bytes. Default 5 MiB. */
  maxAttachmentBytes?: number;
}

export interface MinionWorkerOpts {
  queue?: string;
  concurrency?: number; // default 1
  lockDuration?: number; // ms, default 30000
  stalledInterval?: number; // ms, default 30000
  maxStalledCount?: number; // default 1
  pollInterval?: number; // ms, default 5000 (for PGLite fallback)
  /** RSS threshold in MB. When exceeded, worker triggers graceful shutdown
   *  so a supervisor can respawn it. 0 or undefined = disabled. */
  maxRssMb?: number;
  /** Optional injection point for RSS readback. Defaults to
   *  `() => process.memoryUsage().rss`. Tests inject deterministic sequences. */
  getRss?: () => number;
  /** Periodic RSS check interval in ms, default 60000. Catches the freeze
   *  case where all concurrency slots are wedged with zero job completions
   *  so the per-job check never fires. */
  rssCheckInterval?: number;
  /** Self-health-check interval in ms. 0 = disabled. Default: 60000 (1 minute).
   *  Automatically disabled when running under a supervisor (GBRAIN_SUPERVISED=1).
   *  Provides DB liveness probes and stall detection for bare `gbrain jobs work`
   *  deployments managed by external process managers (systemd, Docker, cron). */
  healthCheckInterval?: number;
  /** Stall detection: ms of continuous idle (waiting>0, inFlight=0, no completions)
   *  before emitting the first warning. Default: 300000 (5 minutes). */
  stallWarnAfterMs?: number;
  /** Stall detection: ms of continuous idle before emitting `'unhealthy'` with
   *  reason='stalled'. Default: 600000 (10 minutes). Must be > stallWarnAfterMs. */
  stallExitAfterMs?: number;
  /** DB liveness probe: number of consecutive failed `SELECT 1` probes before
   *  emitting `'unhealthy'` with reason='db_dead'. Default: 3. */
  dbFailExitAfter?: number;
  /** Per-probe wall-clock timeout in ms. A `SELECT 1` that hangs longer than
   *  this counts as a failure (fed into dbFailExitAfter). Without this, a
   *  hung probe would wedge the recursive setTimeout chain forever and
   *  silently disable the health monitor. Default: 10000 (10 seconds). */
  dbProbeTimeoutMs?: number;
}

// --- Job Context (passed to handlers) ---

export interface MinionJobContext {
  id: number;
  name: string;
  data: Record<string, unknown>;
  attempts_made: number;
  /** AbortSignal for cooperative cancellation (fires on timeout, cancel, pause, or lock loss). */
  signal: AbortSignal;
  /** Absolute wall-clock deadline (epoch ms) from the claim-time `timeout_at` stamp,
   *  or null when the job has no per-job timeout. This is the DB's ground truth —
   *  the same instant handleTimeouts() dead-letters against — so handlers that
   *  spawn bounded sub-work (e.g. autopilot-cycle's subagent phases) can budget
   *  from the REMAINING time instead of a fixed constant that may exceed it. */
  deadlineAtMs: number | null;
  /** AbortSignal that fires only on worker process SIGTERM/SIGINT. Handlers sensitive
   *  to deploy restarts (e.g. the shell handler, which must run a SIGTERM → 5s → SIGKILL
   *  sequence on its child) listen to this in addition to `signal`. Most handlers can
   *  ignore it — workers give them the full 30s cleanup race to finish naturally. */
  shutdownSignal: AbortSignal;
  /** Update structured progress (not just 0-100). */
  updateProgress(progress: unknown): Promise<void>;
  /** Accumulate token usage for this job. */
  updateTokens(tokens: TokenUpdate): Promise<void>;
  /** Append a log message or transcript entry to the job's stacktrace array. */
  log(message: string | TranscriptEntry): Promise<void>;
  /** Check if the lock is still held (for long-running jobs). */
  isActive(): Promise<boolean>;
  /** Read unread inbox messages (marks as read). */
  readInbox(): Promise<InboxMessage[]>;
}

export type MinionHandler = (job: MinionJobContext) => Promise<unknown>;

// --- Inbox Message ---

export interface InboxMessage {
  id: number;
  job_id: number;
  sender: string;
  payload: unknown;
  sent_at: Date;
  read_at: Date | null;
}

export function rowToInboxMessage(row: Record<string, unknown>): InboxMessage {
  return {
    id: row.id as number,
    job_id: row.job_id as number,
    sender: row.sender as string,
    payload: typeof row.payload === 'string' ? JSON.parse(row.payload) : row.payload,
    sent_at: new Date(row.sent_at as string),
    read_at: row.read_at ? new Date(row.read_at as string) : null,
  };
}

// --- Child-done inbox message (auto-posted on every terminal transition) ---

/**
 * Posted into the parent's inbox when a child reaches a terminal state.
 *
 * Pre-v0.15: only success paths (completeJob) emitted this. Failed/dead/
 * cancelled children produced no payload, which stranded aggregator-style
 * parents that needed to wait for N children regardless of outcome.
 *
 * v0.15: failJob, cancelJob, and handleTimeouts also emit child_done with
 * the appropriate `outcome`, so the aggregator handler can count "N children
 * resolved" without worrying about which rail each one took.
 *
 * Backwards compatible: old ChildDoneMessage consumers only read child_id,
 * job_name, and result (non-null on success). Outcome and error are additive.
 */
export type ChildOutcome = 'complete' | 'failed' | 'dead' | 'cancelled' | 'timeout';

export interface ChildDoneMessage {
  type: 'child_done';
  child_id: number;
  job_name: string;
  result: unknown;
  /**
   * Terminal outcome. When absent (from a pre-v0.15 writer that didn't set
   * it), consumers should treat the message as 'complete' — the legacy writer
   * only emitted on success paths.
   */
  outcome?: ChildOutcome;
  /** Set when outcome !== 'complete'. Mirrors minion_jobs.error_text. */
  error?: string | null;
}

// --- Attachments (v7) ---

/** Caller-supplied attachment payload. content is base64-encoded bytes. */
export interface AttachmentInput {
  filename: string;
  content_type: string;
  /** Base64-encoded file bytes. Validated server-side. */
  content_base64: string;
}

/** Persisted attachment row (without inline bytes; use getAttachment to fetch). */
export interface Attachment {
  id: number;
  job_id: number;
  filename: string;
  content_type: string;
  storage_uri: string | null;
  size_bytes: number;
  sha256: string;
  created_at: Date;
}

export function rowToAttachment(row: Record<string, unknown>): Attachment {
  return {
    id: row.id as number,
    job_id: row.job_id as number,
    filename: row.filename as string,
    content_type: row.content_type as string,
    storage_uri: (row.storage_uri as string) || null,
    size_bytes: row.size_bytes as number,
    sha256: row.sha256 as string,
    created_at: new Date(row.created_at as string),
  };
}

// --- Token Update ---

export interface TokenUpdate {
  input?: number;
  output?: number;
  cache_read?: number;
}

// --- Structured Progress (convention, not enforced) ---

export interface AgentProgress {
  step: number;
  total: number;
  message: string;
  tokens_in: number;
  tokens_out: number;
  last_tool: string;
  started_at: string;
}

// --- Transcript Entry ---

export type TranscriptEntry =
  | { type: 'log'; message: string; ts: string }
  | { type: 'tool_call'; tool: string; args_size: number; result_size: number; ts: string }
  | { type: 'llm_turn'; model: string; tokens_in: number; tokens_out: number; ts: string }
  | { type: 'error'; message: string; stack?: string; ts: string };

// --- Errors ---

/** Throw this from a handler to skip all retry logic and go straight to 'dead'. */
export class UnrecoverableError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'UnrecoverableError';
  }
}

// --- Row Mapping ---

export function rowToMinionJob(row: Record<string, unknown>): MinionJob {
  return {
    id: row.id as number,
    name: row.name as string,
    queue: row.queue as string,
    status: row.status as MinionJobStatus,
    priority: row.priority as number,
    data: (typeof row.data === 'string' ? JSON.parse(row.data) : row.data ?? {}) as Record<string, unknown>,
    max_attempts: row.max_attempts as number,
    attempts_made: row.attempts_made as number,
    attempts_started: row.attempts_started as number,
    backoff_type: row.backoff_type as BackoffType,
    backoff_delay: row.backoff_delay as number,
    backoff_jitter: row.backoff_jitter as number,
    stalled_counter: row.stalled_counter as number,
    max_stalled: row.max_stalled as number,
    lock_token: (row.lock_token as string) || null,
    lock_until: row.lock_until ? new Date(row.lock_until as string) : null,
    delay_until: row.delay_until ? new Date(row.delay_until as string) : null,
    parent_job_id: (row.parent_job_id as number | null) ?? null,
    on_child_fail: row.on_child_fail as ChildFailPolicy,
    tokens_input: (row.tokens_input as number) ?? 0,
    tokens_output: (row.tokens_output as number) ?? 0,
    tokens_cache_read: (row.tokens_cache_read as number) ?? 0,
    depth: (row.depth as number) ?? 0,
    max_children: (row.max_children as number) ?? null,
    timeout_ms: (row.timeout_ms as number) ?? null,
    timeout_at: row.timeout_at ? new Date(row.timeout_at as string) : null,
    remove_on_complete: row.remove_on_complete === true,
    remove_on_fail: row.remove_on_fail === true,
    idempotency_key: (row.idempotency_key as string) || null,
    quiet_hours: row.quiet_hours ? (typeof row.quiet_hours === 'string' ? JSON.parse(row.quiet_hours) : row.quiet_hours) as Record<string, unknown> : null,
    stagger_key: (row.stagger_key as string) || null,
    result: row.result ? (typeof row.result === 'string' ? JSON.parse(row.result) : row.result) as Record<string, unknown> : null,
    progress: row.progress ? (typeof row.progress === 'string' ? JSON.parse(row.progress) : row.progress) : null,
    error_text: (row.error_text as string) || null,
    stacktrace: row.stacktrace ? (typeof row.stacktrace === 'string' ? JSON.parse(row.stacktrace) : row.stacktrace) as string[] : [],
    created_at: new Date(row.created_at as string),
    started_at: row.started_at ? new Date(row.started_at as string) : null,
    finished_at: row.finished_at ? new Date(row.finished_at as string) : null,
    updated_at: new Date(row.updated_at as string),
  };
}

// ---------------------------------------------------------------------------
// Subagent runtime (v0.15+)
// ---------------------------------------------------------------------------

/**
 * Input payload for the 'subagent' handler. Shape is intentionally narrow —
 * tool registry and provider config resolve via handler-side defaults + env,
 * not per-job data, so restart/replay uses the same behavior.
 */
export interface SubagentHandlerData {
  /** Top-level user turn kicking off the loop. */
  prompt: string;
  /** Optional subagent definition path (skills/subagents/*.md or plugin). */
  subagent_def?: string;
  /** Anthropic model id. Defaults to sonnet at handler resolution time. */
  model?: string;
  /** Max assistant turns before the loop fails with stop_reason='max_turns'. */
  max_turns?: number;
  /**
   * Per-turn max output tokens (#2778). Resolution: this field →
   * `agent.max_output_tokens` config → 8192 default. The pre-#2778
   * hardcoded 4096 made pages >~12KB unwritable via put_page.
   */
  max_tokens?: number;
  /**
   * Whitelist of tool names the agent may call. MUST be a subset of the
   * derived registry names — invalid entries are rejected at tool-dispatch
   * time, not silently ignored. Empty array = no tools.
   */
  allowed_tools?: string[];
  /** System prompt override. When omitted, the handler builds one. */
  system?: string;
  /** Template variables for subagent_def. Arbitrary JSON-serializable. */
  input_vars?: Record<string, unknown>;
  /**
   * Connected-gbrains brain id (v0.19+, PR 0 plumbing only).
   *
   * CURRENT BEHAVIOR: stamped onto every tool-call's `OperationContext.
   * brainId` but NOT yet used to select an engine at dispatch time.
   * `gbrain agent run` does not yet accept a `--brain` flag that would
   * populate this field — all subagent jobs submitted by the CLI today
   * default to the host engine. The field + handler acceptance exist so
   * PR 1 can add the registry lookup + CLI flag in a single commit.
   *
   * FUTURE (PR 1): setting `brain_id: "yc-media"` at job submission will
   * cause every tool call from the subagent to run against the yc-media
   * engine via BrainRegistry.getBrain() at buildOpContext time.
   */
  brain_id?: string;
  /**
   * Trusted-workspace allow-list for put_page (v0.23 dream cycle).
   *
   * When set, the subagent's put_page calls are bounded to slugs matching
   * any of these prefix globs (e.g. ["wiki/personal/reflections/*",
   * "wiki/originals/*"]). When unset/empty, the legacy
   * `wiki/agents/<subagentId>/...` namespace check applies.
   *
   * Trust comes from PROTECTED_JOB_NAMES gating subagent submission — MCP
   * cannot reach this field. Only cycle.ts (synthesize/patterns phases)
   * and direct CLI submitters set it.
   */
  allowed_slug_prefixes?: string[];
  /**
   * Brain source the subagent's tool calls are scoped to (#1586).
   *
   * When set, every tool-call `OperationContext.sourceId` uses this value
   * instead of the legacy 'default', so put_page writes land in the cycle's
   * resolved source. Same trust story as `allowed_slug_prefixes`:
   * PROTECTED_JOB_NAMES gates subagent submission, so only cycle.ts and
   * direct CLI submitters can set it. Validated via `validateSourceId` at
   * tool-registry build time.
   */
  source_id?: string;
  /**
   * v0.41 Approach C: opt out of the auto-generated tool-usage preamble
   * that `buildSystemPrompt()` splices into `system`. Default behavior
   * (omitted or false) prepends a deterministic preamble listing each
   * tool's name + usage_hint. Set to `true` to keep `system` byte-for-byte
   * as provided.
   *
   * Use when the caller has hand-tuned a complete system prompt for a
   * specific subagent (no benefit from auto-generated guidance, prompt
   * cache hits ride entirely on the caller-supplied prefix).
   */
  system_no_tool_preamble?: boolean;
  /**
   * v0.41 E6 — opt OUT of classifier-gated auto-resubmit on terminal
   * failures. Default behavior (omitted or false) runs the
   * `RECOVERABLE_CLUSTERS` self-fix path when the failure classifies as
   * one of {`prompt_too_long`, `tool_schema_mismatch`, `malformed_json`}.
   * Set true to disable per-job (useful for graders / probes where a
   * retry would muddy the signal).
   */
  no_self_fix?: boolean;
  /**
   * v0.41 E6 — internal marker set by `submitSelfFixChild` so the
   * chain-depth walker can count self-fix ancestors. Counter starts at
   * 0 on a fresh user-submitted job; increments by 1 per chain hop.
   */
  is_self_fix_child?: boolean;
  /**
   * v0.41 E6 — which classifier bucket triggered this self-fix child.
   * Read by audit + diagnostic surfaces (jobs get / dashboard).
   */
  self_fix_cluster?: string;
}

/**
 * Input for the 'subagent_aggregator' handler. Claims AFTER all children
 * resolve and aggregates their results into a brain page.
 */
export interface AggregatorHandlerData {
  /** The subagent child job ids this aggregator is waiting on. */
  children_ids: number[];
  /**
   * Optional template for the synthesis prompt. When omitted, the handler
   * uses a generic "summarize these N results" prompt.
   */
  aggregate_prompt_template?: string;
  /**
   * Target slug for the aggregated brain page. When present, a trusted-CLI
   * put_page (viaSubagent=false) writes the final aggregation there.
   */
  output_slug?: string;
}

/** Tool execution context passed to every ToolDef.execute. */
export interface ToolCtx {
  /** Engine for DB-backed tools (brain_query, put_page, etc.). */
  engine: import('../engine.ts').BrainEngine;
  /** The subagent job id (used for audit + put_page namespace enforcement). */
  jobId: number;
  /** Always true for LLM-invoked tools — matches MCP trust boundary. */
  remote: true;
  /** Fired on cooperative abort (timeout, lock loss, cancel, SIGTERM). */
  signal?: AbortSignal;
}

/**
 * A tool the subagent can call. Names match Anthropic's constraint
 * `^[a-zA-Z0-9_-]{1,64}$` — no dots. The input_schema is the JSONSchema
 * shipped to the Anthropic Messages API verbatim; ToolDef is the single
 * Anthropic-compatible envelope, not an MCP McpToolDef (those have a
 * different shape — ".inputSchema" vs ".input_schema").
 *
 * `idempotent: true` is required for the two-phase replay path: on resume,
 * a 'pending' row can be re-executed. Non-idempotent tools need a separate
 * resume policy and are not supported in v0.15.
 */
export interface ToolDef {
  name: string;
  description: string;
  input_schema: Record<string, unknown>;
  idempotent: boolean;
  execute(input: unknown, ctx: ToolCtx): Promise<unknown>;
  /**
   * v0.41 Approach C: one-line hint surfaced verbatim in the subagent
   * system prompt's tool preamble. Tells the model WHEN to use this tool.
   * The `description` tells the model HOW; the `usage_hint` tells WHEN.
   *
   * Field-report case: a `shell` tool sat in the registry and the subagent
   * never used it because nothing told the model "to write a file, use
   * shell." Per-tool hints surface that directly. Plugin authors get this
   * affordance for free.
   *
   * Optional — omitted tools just don't get a hint line. Must be a single
   * line (no embedded newlines) so the rendered preamble stays scannable.
   */
  usage_hint?: string;
}

/**
 * Anthropic content-block subset we persist in subagent_messages.content_blocks.
 * This is structural — we don't gatekeep on unknown block types (future SDK
 * additions pass through). Use the string-literal discriminant on 'type'.
 */
export type ContentBlock =
  | { type: 'text'; text: string; [k: string]: unknown }
  | { type: 'tool_use'; id: string; name: string; input: unknown; [k: string]: unknown }
  | { type: 'tool_result'; tool_use_id: string; content: unknown; is_error?: boolean; [k: string]: unknown }
  | { type: string; [k: string]: unknown };

/** Stop reason reported to the caller when the subagent loop terminates. */
export type SubagentStopReason =
  | 'end_turn'    // Anthropic says end_turn and last message has no tool_use
  | 'max_turns'   // hit max_turns budget before end_turn
  | 'max_tokens'  // final turn hit the output-token cap — result text is TRUNCATED (#2778)
  | 'refusal'     // detected via stop_reason + content shape
  | 'error';      // unrecoverable (empty response retry exhausted, etc.)

/** Terminal result payload emitted by the subagent handler. */
export interface SubagentResult {
  /** Concatenated text from the final assistant message. */
  result: string;
  /** Number of assistant turns consumed. */
  turns_count: number;
  /** Why the loop stopped. */
  stop_reason: SubagentStopReason;
  /** Rollup of tokens across all turns. */
  tokens: {
    in: number;
    out: number;
    cache_read: number;
    cache_create: number;
  };
}
