# SPDX-License-Identifier: MPL-2.0
# Copyright (c) 2026 Jonathan D.A. Jewell (hyperpolymath) <j.d.a.jewell@open.ac.uk>

defmodule Hypatia.LearningScheduler do
  @moduledoc """
  GenServer that automatically closes the feedback loop.

  Periodically:
  1. Checks verisim-data/outcomes/ for new JSONL entries
  2. Calls OutcomeTracker.update_recipe_confidence/1 for affected recipes
  3. Checks gitbot-fleet fix-outcomes.jsonl for fleet-reported outcomes
  4. Ingests fleet outcomes back into hypatia's learning pipeline
  5. Reports confidence drift (recipes dropping or rising)

  This is the critical missing piece that makes the learning loop automatic.
  Without it, outcomes sit in JSONL files and never update recipe confidence.
  """

  use GenServer
  require Logger

  alias Hypatia.ConfidenceAnnealing
  alias Hypatia.EchidnabotObligation
  alias Hypatia.ServiceUrl

  # 5 minutes
  @poll_interval_ms 5 * 60 * 1_000
  # Action-SHA verification runs once per 24h — gh-api-heavy, doesn't need
  # per-cycle frequency. Cached so first run after 24h is fast for unchanged
  # pins.
  @action_sha_verify_interval_seconds 24 * 60 * 60
  @action_sha_cache_path "data/verified-action-shas.json"
  @annealing_state_path Application.compile_env(
                          :hypatia,
                          :annealing_state_path,
                          "data/verisim/annealing-states"
                        )

  # --- Client API ---

  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @doc "Force an immediate learning cycle (for testing/manual trigger)"
  def run_now do
    GenServer.cast(__MODULE__, :run_cycle)
  end

  @doc "Get the current state (last run, outcomes processed, etc.)"
  def status do
    GenServer.call(__MODULE__, :status)
  end

  # --- GenServer Callbacks ---

  @impl true
  def init(_opts) do
    state = %{
      last_run: nil,
      # filename => source-qualified position of the last complete read
      last_outcome_positions: %{},
      pending_confidence_recipes: MapSet.new(),
      # A restarted scheduler rechecks persisted source IDs, not just this cache.
      last_fleet_signature: nil,
      total_outcomes_processed: 0,
      total_confidence_updates: 0,
      total_annealing_reheats: 0,
      # Both keys must exist here before the cycle's map-update touches them:
      # `%{state | total_strategy_shifts: …}` raises KeyError on a missing key
      # even when the value expression uses Map.get/3 — the update itself is
      # the guard. Their absence crashed every learning cycle at line 188
      # (observed 2026-10-15: the scheduler died on each :poll and restarted,
      # so confidence updates never landed and the neural organs never trained).
      total_strategy_shifts: 0,
      proof_model_retrain_count: 0,
      started_at: DateTime.utc_now(),
      errors: []
    }

    # Schedule first poll after a short delay (let app boot)
    Process.send_after(self(), :poll, 5_000)
    Logger.info("LearningScheduler started. Polling every #{div(@poll_interval_ms, 1000)}s.")
    {:ok, state}
  end

  @impl true
  def handle_info(:poll, state) do
    new_state = run_learning_cycle(state)
    Process.send_after(self(), :poll, @poll_interval_ms)
    {:noreply, new_state}
  end

  @impl true
  def handle_cast(:run_cycle, state) do
    new_state = run_learning_cycle(state)
    {:noreply, new_state}
  end

  @impl true
  def handle_call(:status, _from, state) do
    {:reply, state, state}
  end

  # --- Core Learning Cycle ---

  defp run_learning_cycle(state) do
    Logger.info("LearningScheduler: running feedback cycle...")

    {new_positions, outcomes_count, recipe_ids} =
      ingest_new_outcomes(state.last_outcome_positions)

    {fleet_outcomes_count, fleet_signature} =
      ingest_fleet_outcomes(state.last_fleet_signature)

    {confidence_updates, pending_confidence_recipes} =
      update_affected_recipes(MapSet.union(recipe_ids, state.pending_confidence_recipes))

    report_confidence_drift()

    # Report annealing state across all tracked recipes
    annealing_report = report_annealing_state()

    # N4: Detect strategy-shift events and re-queue candidates.
    shift_events = detect_and_requeue_strategy_shifts()

    # N5: Retrain ProverRecommender from VeriSimDB proof_attempts.
    # Reads the growing proof_attempts table (written by echidnabot's
    # VeriSimWriter) and re-fits per-class RBF networks. This closes the
    # echidna → VeriSimDB → Hypatia learning loop: new attempts from
    # echidnabot immediately influence the next prover hint recommendation.
    proof_model_updated = retrain_prover_recommender()

    # N6: Trigger the Neural Coordinator's learning cycle so the organs
    # (ESN/RBF) actually train on the outcomes just ingested. Without this
    # the Coordinator's networks run on init-defaults forever and
    # pattern_analyzer defensively ignores their (untrained) output -- the
    # root cause of the token-cost/scattergun symptom (issue #274).
    neural_cycled = trigger_neural_cycle()

    # M18 — pull findings from configured cross-org peers and import them
    # under each peer's policy gates. No-op when :cross_org_policies is
    # unset (the default). Best-effort: a failing peer never breaks the
    # local learning cycle.
    cross_org_imports = poll_cross_org_peers()

    # Daily action-SHA verification. Companion to the static
    # `:known_fake_action_sha` rule — that rule blocks the 25 known fakes
    # from the 2026-05-30 audit, this catches future fabrications via
    # `gh api commits/<sha>`. Cache at `data/verified-action-shas.json`
    # makes subsequent runs fast (only verifies new pins). Spawned as a
    # subprocess so the mix task's `exit({:shutdown, N})` can't crash this
    # GenServer. Async — does not block the learning cycle.
    maybe_verify_action_shas()

    now = DateTime.utc_now()
    total = outcomes_count + fleet_outcomes_count

    if total > 0 do
      Logger.info(
        "LearningScheduler: processed #{total} new outcomes, " <>
          "updated #{confidence_updates} recipes, " <>
          "annealing: #{annealing_report.nascent} nascent, " <>
          "#{annealing_report.adolescent} adolescent, " <>
          "#{annealing_report.mature} mature, " <>
          "#{annealing_report.veteran} veteran, " <>
          "#{annealing_report.reheated} reheated"
      )
    end

    if length(shift_events) > 0 do
      Logger.info("LearningScheduler: detected #{length(shift_events)} strategy shifts")
    end

    if proof_model_updated do
      Logger.info("LearningScheduler: prover recommender retrained from VeriSimDB proof_attempts")
    end

    if cross_org_imports > 0 do
      Logger.info(
        "LearningScheduler: cross_org imported #{cross_org_imports} accepted finding(s) " <>
          "from #{length(Hypatia.VCL.CrossOrg.peers())} peer(s)"
      )
    end

    if neural_cycled do
      Logger.info(
        "LearningScheduler: neural learning cycle triggered " <>
          "(ESN/RBF train on #{total} new + accumulated outcomes)"
      )
    end

    %{
      state
      | last_run: now,
        last_outcome_positions: new_positions,
        pending_confidence_recipes: pending_confidence_recipes,
        last_fleet_signature: fleet_signature,
        total_outcomes_processed: state.total_outcomes_processed + total,
        total_confidence_updates: state.total_confidence_updates + confidence_updates,
        total_annealing_reheats: state.total_annealing_reheats + annealing_report.reheated,
        total_strategy_shifts: Map.get(state, :total_strategy_shifts, 0) + length(shift_events),
        proof_model_retrain_count:
          Map.get(state, :proof_model_retrain_count, 0) + if(proof_model_updated, do: 1, else: 0)
    }
  end

  # --- Action-SHA verification (companion to :known_fake_action_sha rule) ----

  # Runs `mix hypatia.verify_action_shas` at most once per 24 hours. The mix
  # task itself maintains a SHA→{real|fake} cache so re-runs only call
  # `gh api` for new pins.
  #
  # Spawned as a subprocess so the mix task's `exit({:shutdown, 2})` on
  # fakes-found cannot crash this GenServer. Fully async (Task.start,
  # no return value) — the learning cycle never waits on it.
  defp maybe_verify_action_shas do
    if action_sha_verify_stale?() do
      Task.start(fn -> run_action_sha_verification() end)
    end
  end

  defp action_sha_verify_stale? do
    case File.stat(@action_sha_cache_path) do
      {:ok, %{mtime: mtime}} ->
        # mtime is an erlang :calendar.datetime tuple in UTC
        cache_secs = :calendar.datetime_to_gregorian_seconds(mtime)
        now_secs = :calendar.datetime_to_gregorian_seconds(:calendar.universal_time())
        now_secs - cache_secs > @action_sha_verify_interval_seconds

      {:error, _} ->
        # No cache yet → run on first cycle
        true
    end
  rescue
    e ->
      Logger.warning("LearningScheduler: action_sha cache stat failed: #{inspect(e)}")
      false
  end

  defp run_action_sha_verification do
    Logger.info("LearningScheduler: running daily action-SHA verification...")

    try do
      case System.cmd("mix", ["hypatia.verify_action_shas", "--quiet"], stderr_to_stdout: true) do
        {_output, 0} ->
          Logger.info("LearningScheduler: action_sha verify clean (zero fakes)")

        {output, 2} ->
          # Exit 2 = fakes-found. Surface the count and a clipped sample.
          Logger.warning(
            "LearningScheduler: action_sha verify FOUND FAKES — " <>
              String.slice(output, 0, 500)
          )

        {output, code} ->
          Logger.warning(
            "LearningScheduler: action_sha verify exit=#{code} — " <>
              String.slice(output, 0, 200)
          )
      end
    rescue
      e ->
        Logger.warning("LearningScheduler: action_sha verify crashed: #{inspect(e)}")
    end
  end

  # --- N5: ProverRecommender retraining from VeriSimDB proof_attempts --------

  # Fetches proof_attempts from VeriSimDB and re-fits ProverRecommender's
  # per-class RBF networks. Returns true if training succeeded, false if
  # VeriSimDB was unconfigured (HYPATIA_VERISIM_URL unset), unreachable or
  # returned too few samples.
  #
  # Minimum sample threshold (50) prevents overfitting on early sparse data.
  # The existing in-memory models continue serving recommendations until
  # a successful retrain replaces them.
  defp retrain_prover_recommender do
    try do
      case Hypatia.Neural.ProverRecommender.train_from_verisim() do
        {:ok, models} ->
          sample_size = Map.get(models, :sample_size, 0)

          if sample_size >= 50 do
            # Store the retrained models in the ProverRecommender agent/GenServer.
            # ProverRecommender.update_models/1 swaps the live model atomically.
            :ok = Hypatia.Neural.ProverRecommender.update_models(models)
            true
          else
            Logger.debug(
              "LearningScheduler: prover_recommender skip (only #{sample_size} samples, need 50)"
            )

            false
          end

        {:error, reason} ->
          Logger.debug(
            "LearningScheduler: prover_recommender retrain skipped -- #{inspect(reason)}"
          )

          false
      end
    rescue
      e ->
        Logger.warning("LearningScheduler: prover_recommender retrain error: #{inspect(e)}")
        false
    end
  end

  # --- N6: Neural Coordinator learning-cycle trigger -------------------

  # Casts :force_cycle to the Neural Coordinator so it trains ESN on the
  # accumulated confidence trajectories and RBF on the pattern registry,
  # then persists all neural states. Cast is fire-and-forget (training
  # runs inside the Coordinator process); we only confirm the process is
  # alive. Non-fatal: the scheduler must keep running even if the neural
  # layer is down.
  defp trigger_neural_cycle do
    case Process.whereis(Hypatia.Neural.Coordinator) do
      nil ->
        false

      _pid ->
        Hypatia.Neural.Coordinator.force_cycle()
        true
    end
  rescue
    e ->
      Logger.warning("LearningScheduler: neural cycle trigger failed: #{inspect(e)}")
      false
  end

  # --- M18: Cross-org peer polling -------------------------------------
  #
  # For every peer configured under :hypatia, :cross_org_policies, pull
  # recent findings through their /graphql endpoint and apply the peer
  # policy gates (severity allowlist, age cap, dismissal override, drift
  # threshold). Returns the total accepted-finding count for logging.
  #
  # All failure modes are isolated: an unreachable peer / bad response /
  # parse error never breaks the local learning cycle. The CrossOrg
  # module itself returns {:error, reason} for any of these; we just
  # tally and continue.
  defp poll_cross_org_peers do
    case Hypatia.VCL.CrossOrg.peers() do
      [] ->
        0

      peers ->
        Enum.reduce(peers, 0, fn policy, acc ->
          peer_id = Map.get(policy, :peer_id)

          case Hypatia.VCL.CrossOrg.pull_findings(peer_id: peer_id, limit: 100) do
            {:ok, accepted, stats} ->
              Logger.info(
                "LearningScheduler: cross_org peer #{peer_id} accepted=#{stats.accepted} " <>
                  "dropped_severity=#{stats.dropped_severity} dropped_age=#{stats.dropped_age} " <>
                  "dropped_dismissed=#{stats.dropped_dismissed} drift_alerts=#{stats.drift_alerts}"
              )

              acc + length(accepted)

            {:error, reason} ->
              Logger.warning("LearningScheduler: cross_org peer #{peer_id} skipped: #{reason}")
              acc
          end
        end)
    end
  rescue
    e ->
      Logger.warning("LearningScheduler: cross_org poll crashed: #{inspect(e)}")
      0
  end

  # --- N4: Strategy-shift detection + re-queueing ----------------------

  # Runs StrategyDrift.check_all_shifts/1 on each tick. For each shift
  # event, enqueues the candidate failed-attempt IDs via echidnabot's
  # submitProofObligation mutation with the new top prover as hint.
  # Logs but does not block on re-queueing errors. With HYPATIA_VERISIM_URL
  # unset no class can shift, so this returns [] without a network call.
  defp detect_and_requeue_strategy_shifts do
    try do
      events = Hypatia.Rules.StrategyDrift.check_all_shifts()

      Enum.each(events, fn {:shift, class, old_top, new_top, candidates} ->
        Logger.info(
          "strategy shift: class=#{class} #{old_top} → #{new_top} " <>
            "(#{length(candidates)} candidates to re-queue)"
        )

        requeue_candidates(class, new_top, candidates)
      end)

      events
    rescue
      e ->
        Logger.warning("StrategyDrift check failed: #{inspect(e)}")
        []
    end
  end

  @doc """
  Re-queue failed proof attempts with echidnabot after a strategy shift.

  Sends one `submitProofObligation` per attempt id (at most 20 per call) with
  `new_top` as the prover hint, and returns `:ok`. Failures are logged and
  never raised: the scheduler must keep running when echidnabot is
  unreachable.

  `HYPATIA_ECHIDNABOT_URL` is echidnabot's base URL, the same meaning
  `Hypatia.FleetDispatcher` gives it, and `/graphql` is appended. It has no
  default: when it is unset or blank, nothing is sent, the dropped candidates
  are logged, and the call returns `:ok` (see `Hypatia.ServiceUrl`).
  `new_top` goes through
  `Hypatia.EchidnabotObligation.normalise_prover_hint/1`, so a name with no
  echidnabot `ProverKind` (such as `"lean4"`) sends no prover instead of an
  invalid enum value.
  """
  @spec requeue_candidates(String.t(), String.t() | nil, [String.t()]) :: :ok
  def requeue_candidates(_class, _new_top, []), do: :ok

  def requeue_candidates(class, new_top, candidates) do
    case ServiceUrl.echidnabot() do
      {:ok, base_url} ->
        send_requeues(EchidnabotObligation.graphql_url(base_url), class, new_top, candidates)

      {:error, :not_configured} ->
        Logger.info(
          "LearningScheduler: HYPATIA_ECHIDNABOT_URL unset -- not re-queueing " <>
            "#{length(candidates)} attempts for class=#{class}"
        )

        :ok
    end
  end

  # Send at most 20 re-queues to echidnabot's GraphQL endpoint
  # `echidnabot_url` and log how many it accepted.
  defp send_requeues(echidnabot_url, class, new_top, candidates) do
    success_count =
      candidates
      # rate-limit: max 20 re-queues per tick
      |> Enum.take(20)
      |> Enum.count(fn attempt_id ->
        submit_requeue(echidnabot_url, class, new_top, attempt_id)
      end)

    if success_count > 0 do
      Logger.info(
        "  re-queued #{success_count}/#{length(candidates)} attempts via echidnabot (capped at 20/tick)"
      )
    end

    :ok
  end

  # Send one best-effort submitProofObligation for a failed attempt and
  # report whether echidnabot accepted it. The document is static; the class,
  # attempt id and prover travel as GraphQL variables built by
  # EchidnabotObligation, so no value can change the document.
  defp submit_requeue(url, class, prover, attempt_id) do
    variables =
      EchidnabotObligation.variables(
        "hyperpolymath/requeue",
        "requeue-of-#{attempt_id}",
        "strategy-shift class=#{class}",
        prover,
        inline: true
      )

    body = Jason.encode!(%{"query" => EchidnabotObligation.mutation(), "variables" => variables})

    case http_post_json(url, body, 5_000) do
      {:ok, response_body} ->
        case Jason.decode(response_body) do
          {:ok, %{"data" => %{"submitProofObligation" => %{"success" => true}}}} -> true
          _ -> false
        end

      {:error, _} ->
        false
    end
  end

  defp http_post_json(url, body, timeout_ms) do
    Application.ensure_all_started(:inets)
    Application.ensure_all_started(:ssl)

    case :httpc.request(
           :post,
           {String.to_charlist(url), [{~c"accept", ~c"application/json"}], ~c"application/json",
            body},
           [{:timeout, timeout_ms}],
           []
         ) do
      {:ok, {{_, status, _}, _headers, resp_body}} when status in 200..299 ->
        {:ok, List.to_string(resp_body)}

      {:ok, {{_, status, _}, _, _}} ->
        {:error, {:http_status, status}}

      {:error, reason} ->
        {:error, {:transport, reason}}
    end
  end

  # --- Ingest new outcomes from verisim-data/outcomes/*.jsonl ---

  # E0 repair (#984, #995): the local JSONL reader must not acknowledge
  # incomplete records, must quarantine malformed complete rows, must
  # detect file truncation/rotation, and must use a stable event identity
  # so the same event on replay credits once while two distinct attempts
  # with identical payloads credit twice.

  @doc false
  def local_source_id(file, line_number) when is_binary(file) and is_integer(line_number) do
    # Append-only line identity: disambiguates distinct identical attempts
    # (same recipe_id, same outcome bytes) while making replay of the same
    # delivery detectable. A future producer event_id supersedes this.
    :crypto.hash(:sha256, "local-jsonl-v1:" <> file <> ":" <> Integer.to_string(line_number))
    |> Base.encode16(case: :lower)
  end

  # Normalise the offset/tail view of a position entry. Legacy entries
  # may be bare integers or maps without a source_path; ingestion must
  # separately qualify their source before reusing an offset.
  @doc false
  def normalise_position(pos) when is_integer(pos) and pos >= 0,
    do: %{offset: pos, incomplete_tail: ""}

  def normalise_position(%{offset: o, incomplete_tail: t})
      when is_integer(o) and o >= 0 and is_binary(t),
      do: %{offset: o, incomplete_tail: t}

  def normalise_position(_), do: %{offset: 0, incomplete_tail: ""}

  # Keep filename keys compatible with existing state, but never reuse
  # an offset from another runtime store. Source-less legacy cursors
  # conservatively replay once rather than skip an unidentified source.
  defp position_for_source(%{source_path: source_path} = position, path)
       when source_path == path,
       do: normalise_position(position)

  defp position_for_source(_position, _path), do: normalise_position(0)

  @doc false
  def ingest_new_outcomes(last_positions) do
    outcomes_dir = Hypatia.Paths.outcomes() |> Path.expand()

    case File.ls(outcomes_dir) do
      {:ok, files} ->
        jsonl_files = Enum.filter(files, &String.ends_with?(&1, ".jsonl"))

        Enum.reduce(jsonl_files, {last_positions, 0, MapSet.new()}, fn file,
                                                                       {positions, count, recipes} ->
          path = Path.join(outcomes_dir, file)
          pos = position_for_source(Map.get(positions, file), path)

          case File.stat(path) do
            {:ok, %{size: size}} when size < pos.offset ->
              # File truncated or rotated — reset to zero and re-scan.
              Logger.warning(
                "LearningScheduler: outcome file #{file} truncated " <>
                  "(size #{size} < offset #{pos.offset}), resetting"
              )

              # The stat already invalidated the old cursor. Persist the
              # path-bound reset in the accumulator before attempting I/O,
              # so a failed read cannot restore that cursor and skip a
              # replacement prefix once the file grows beyond its old size.
              reset = %{offset: 0, incomplete_tail: "", source_path: path}

              ingest_local_outcome_file(
                path,
                file,
                reset,
                {Map.put(positions, file, reset), count, recipes}
              )

            {:ok, %{size: size}} when size == pos.offset and pos.incomplete_tail == "" ->
              # No new bytes and nothing buffered — nothing to do.
              {positions, count, recipes}

            {:ok, _stat} ->
              ingest_local_outcome_file(path, file, pos, {positions, count, recipes})

            {:error, reason} ->
              Logger.warning(
                "LearningScheduler: cannot stat outcome file #{file}: #{inspect(reason)}"
              )

              {positions, count, recipes}
          end
        end)

      {:error, :enoent} ->
        # Outcomes directory does not exist yet — nothing to ingest.
        {last_positions, 0, MapSet.new()}

      {:error, reason} ->
        Logger.warning("LearningScheduler: cannot list outcomes directory: #{inspect(reason)}")

        {last_positions, 0, MapSet.new()}
    end
  end

  defp ingest_local_outcome_file(path, file, pos, {positions, count, recipes} = acc) do
    read_result =
      File.open(path, [:read, :binary], fn fd ->
        with {:ok, _offset} <- :file.position(fd, pos.offset) do
          IO.binread(fd, :eof)
        end
      end)

    case read_result do
      {:ok, raw} when is_binary(raw) or raw == :eof ->
        # The offset is the start of the incomplete tail, not the end of
        # the previous read. Re-read those bytes from the file exactly once;
        # prepending the saved tail would duplicate a partially written row.
        full_data = if raw == :eof, do: "", else: raw
        lines = String.split(full_data, "\n", trim: false)
        incomplete_tail = List.last(lines)
        complete_lines = Enum.drop(lines, -1)

        # Acknowledge only newline-terminated bytes. Retain the trailing
        # fragment as diagnostic state without moving the read position.
        new_offset = pos.offset + byte_size(full_data) - byte_size(incomplete_tail)

        {new_count, new_recipes, quarantined} =
          process_local_complete_lines(complete_lines, path, file, pos.offset)

        if quarantined > 0 do
          Logger.warning(
            "LearningScheduler: quarantined #{quarantined} malformed outcome row(s) in #{file}"
          )
        end

        new_position = %{
          offset: new_offset,
          incomplete_tail: incomplete_tail,
          source_path: path
        }

        {Map.put(positions, file, new_position), count + new_count,
         MapSet.union(recipes, new_recipes)}

      {:ok, {:error, reason}} ->
        local_read_failed(file, reason, acc)

      {:error, reason} ->
        local_read_failed(file, reason, acc)
    end
  end

  defp local_read_failed(file, reason, acc) do
    Logger.warning("LearningScheduler: cannot read outcome file #{file}: #{inspect(reason)}")
    acc
  end

  # Process a batch of complete JSONL lines from a local outcome file.
  # `base_offset` is the byte offset of the first line in the batch.
  # Returns `{local_count, recipe_ids, quarantined_count}`.
  defp process_local_complete_lines(complete_lines, path, file, base_offset) do
    # Use the same resolved file as the read, including after a runtime
    # data-directory change, so diagnostic line numbers refer to that file.
    start_line = local_line_start(path, base_offset)

    complete_lines
    |> Enum.with_index(start_line)
    |> Enum.reduce({0, MapSet.new(), 0}, fn {line, line_number}, {count, recipes, quarantined} ->
      if line == "" do
        # Empty line between records — skip without counting.
        {count, recipes, quarantined}
      else
        case Jason.decode(line) do
          {:ok, record} when is_map(record) ->
            # Imported fleet entries are already handled by the fleet
            # poll. Do not re-count them on the local poll.
            if Map.get(record, "origin") == "gitbot-fleet" do
              {count, recipes, quarantined}
            else
              recipe_ids =
                record
                |> Map.get("recipe_id")
                |> List.wrap()
                |> Enum.reject(&is_nil/1)

              {count + 1, MapSet.union(recipes, MapSet.new(recipe_ids)), quarantined}
            end

          {:ok, _not_a_map} ->
            Logger.warning(
              "LearningScheduler: quarantined non-map JSON row #{line_number} in #{file}"
            )

            {count, recipes, quarantined + 1}

          {:error, reason} ->
            Logger.warning(
              "LearningScheduler: quarantined malformed outcome row #{line_number} in #{file}: #{inspect(reason)}"
            )

            {count, recipes, quarantined + 1}
        end
      end
    end)
  end

  # Compute the physical line number for a batch starting at
  # `base_offset`. On a cold start (offset 0) this is 1.
  # Otherwise count newlines in the file prefix to find the first line
  # that begins at `base_offset`.
  defp local_line_start(_path, 0), do: 1

  defp local_line_start(path, base_offset) do
    case File.open(path, [:read, :binary]) do
      {:ok, fd} ->
        prefix =
          if base_offset > 0 do
            :file.position(fd, 0)
            IO.read(fd, base_offset)
          else
            ""
          end

        File.close(fd)
        # Every newline in the prefix starts a new line; the batch begins
        # on the line after the last newline at or before base_offset.
        prefix_newlines =
          prefix
          |> to_string()
          |> String.split("\n")
          |> length()
          |> Kernel.-(1)
          |> max(0)

        prefix_newlines + 1

      {:error, _} ->
        1
    end
  end

  # --- Fleet outcomes: verified evidence only, restart-safe replay ---

  @doc false
  def classify_fleet_record(record) when is_map(record) do
    recipe_id = Map.get(record, "recipe_id") || Map.get(record, "pattern")
    repo = Map.get(record, "repo")
    file = Map.get(record, "file")
    file = if is_binary(file) and file != "", do: file, else: "."

    # A legacy "outcome: success" means only that the fixer exited zero.
    # Require the postcondition verdict AND supporting numeric evidence.
    verdict = Map.get(record, "verdict")
    exit_code = Map.get(record, "exit_code")
    before_hits = Map.get(record, "before_hits")
    after_hits = Map.get(record, "after_hits")

    classified =
      case {verdict, Map.get(record, "outcome")} do
        {"verified", "success"}
        when is_integer(before_hits) and before_hits > 0 and
               after_hits == 0 ->
          {:success, "verified"}

        {"failed", "failure"} when is_integer(exit_code) and exit_code != 0 ->
          {:failure, "failed"}

        {"incomplete", "failure"} when is_integer(after_hits) and after_hits > 0 ->
          {:failure, "incomplete"}

        _ ->
          :unqualified
      end

    cond do
      Map.get(record, "bot") == "hypatia" ->
        :skip

      not (is_binary(recipe_id) and recipe_id != "" and is_binary(repo) and repo != "") ->
        :skip

      classified == :unqualified ->
        :skip

      true ->
        {outcome, verification} = classified
        {:ok, recipe_id, repo, file, outcome, verification}
    end
  end

  def classify_fleet_record(_record), do: :skip

  @doc false
  def fleet_source_id(line, line_number) when is_binary(line) and is_integer(line_number) do
    # Append-only line identity disambiguates distinct identical attempts.
    # A future producer event_id will supersede this provisional locator.
    :crypto.hash(:sha256, "fleet-jsonl-v1:" <> Integer.to_string(line_number) <> ":" <> line)
    |> Base.encode16(case: :lower)
  end

  @doc false
  def ingest_fleet_outcomes(previous_signature) do
    fleet_file =
      Path.join([
        Path.expand(Hypatia.Paths.fleet()),
        "shared-context/learning/fix-outcomes.jsonl"
      ])

    case File.read(fleet_file) do
      {:ok, content} ->
        signature = :crypto.hash(:sha256, content)

        if signature == previous_signature do
          {0, signature}
        else
          case Hypatia.OutcomeTracker.imported_fleet_ids() do
            {:ok, known_ids} ->
              case import_fleet_lines(content, known_ids) do
                {:ok, imported} ->
                  {imported, signature}

                {:error, reason} ->
                  Logger.warning(
                    "LearningScheduler: fleet import not acknowledged: #{inspect(reason)}"
                  )

                  {0, previous_signature}
              end

            {:error, reason} ->
              Logger.warning("LearningScheduler: cannot read imported IDs: #{inspect(reason)}")
              {0, previous_signature}
          end
        end

      {:error, :enoent} ->
        {0, previous_signature}

      {:error, reason} ->
        Logger.warning("LearningScheduler: cannot read fleet file: #{inspect(reason)}")
        {0, previous_signature}
    end
  end

  defp import_fleet_lines(content, known_ids) do
    lines = String.split(content, "\n", trim: false)
    final_line = length(lines)

    lines
    |> Enum.with_index(1)
    |> Enum.reduce_while({:ok, known_ids, 0}, fn {line, line_number}, {:ok, seen, count} ->
      if String.trim(line) == "" do
        {:cont, {:ok, seen, count}}
      else
        case Jason.decode(line) do
          {:ok, record} ->
            case classify_fleet_record(record) do
              {:ok, recipe_id, repo, file, outcome, verification} ->
                id = fleet_source_id(line, line_number)

                if MapSet.member?(seen, id) do
                  {:cont, {:ok, seen, count}}
                else
                  case Hypatia.OutcomeTracker.record_fleet_outcome(
                         recipe_id,
                         repo,
                         file,
                         outcome,
                         id,
                         verification
                       ) do
                    {:ok, _record} ->
                      {:cont, {:ok, MapSet.put(seen, id), count + 1}}

                    {:error, reason} ->
                      {:halt, {:error, {:write_failed, line_number, reason}}}
                  end
                end

              :skip ->
                {:cont, {:ok, seen, count}}
            end

          {:error, reason} ->
            if line_number == final_line and not String.ends_with?(content, "\n") do
              # The producer may still be writing the last record. Retry it.
              {:halt, {:error, {:incomplete_fleet_tail, line_number, reason}}}
            else
              # A permanently malformed historical row must not starve later
              # valid events. It remains in the append-only source as evidence.
              Logger.warning(
                "LearningScheduler: quarantined malformed fleet row #{line_number}: #{inspect(reason)}"
              )

              {:cont, {:ok, seen, count}}
            end
        end
      end
    end)
    |> case do
      {:ok, _seen, count} -> {:ok, count}
      error -> error
    end
  end

  # --- Update confidence for all affected recipes ---

  defp update_affected_recipes(recipe_ids) do
    Enum.reduce(recipe_ids, {0, MapSet.new()}, fn recipe_id, {count, pending} ->
      case Hypatia.OutcomeTracker.update_recipe_confidence(recipe_id) do
        {:ok, _confidence} ->
          {count + 1, pending}

        :no_outcomes ->
          {count, pending}

        error ->
          Logger.warning(
            "LearningScheduler: confidence update deferred for #{recipe_id}: #{inspect(error)}"
          )

          {count, MapSet.put(pending, recipe_id)}
      end
    end)
  end

  # --- Report any recipes with concerning confidence drift ---

  defp report_confidence_drift do
    recipes_dir = Hypatia.Paths.recipes() |> Path.expand()

    case File.ls(recipes_dir) do
      {:ok, files} ->
        files
        |> Enum.filter(&String.ends_with?(&1, ".json"))
        |> Enum.each(fn file ->
          path = Path.join(recipes_dir, file)

          with {:ok, content} <- File.read(path),
               {:ok, recipe} <- Jason.decode(content) do
            confidence = Map.get(recipe, "confidence", 0.5)
            recipe_id = Map.get(recipe, "id", file)

            cond do
              confidence <= 0.3 ->
                Logger.warning(
                  "DRIFT ALERT: Recipe #{recipe_id} confidence dropped to #{confidence} -- " <>
                    "consider disabling auto-execute"
                )

              confidence >= 0.95 ->
                Logger.info(
                  "Recipe #{recipe_id} at high confidence #{confidence} -- auto-execute eligible"
                )

              true ->
                :ok
            end
          end
        end)

      {:error, _} ->
        :ok
    end
  end

  # --- Annealing State Reporting ---

  # Scans all persisted annealing state files and produces a summary
  # of stage distribution and drift/reheat counts.
  defp report_annealing_state do
    annealing_dir = Path.expand(@annealing_state_path)

    stage_counts = %{nascent: 0, adolescent: 0, mature: 0, veteran: 0}
    reheated = 0

    case File.ls(annealing_dir) do
      {:ok, files} ->
        Enum.reduce(
          Enum.filter(files, &String.ends_with?(&1, ".json")),
          %{stages: stage_counts, reheated: reheated},
          fn file, acc ->
            path = Path.join(annealing_dir, file)

            case File.read(path) do
              {:ok, content} ->
                case Jason.decode(content) do
                  {:ok, data} ->
                    stage = String.to_existing_atom(Map.get(data, "stage", "nascent"))

                    drift =
                      ConfidenceAnnealing.drift_detected?(
                        data
                        |> Map.get("recent_outcomes", [])
                        |> Enum.map(&String.to_existing_atom/1)
                      )

                    stages = Map.update!(acc.stages, stage, &(&1 + 1))
                    reheated = if drift, do: acc.reheated + 1, else: acc.reheated
                    %{acc | stages: stages, reheated: reheated}

                  {:error, _} ->
                    acc
                end

              {:error, _} ->
                acc
            end
          end
        )
        |> then(fn acc ->
          Map.merge(acc.stages, %{reheated: acc.reheated})
        end)

      {:error, _} ->
        Map.merge(stage_counts, %{reheated: 0})
    end
  end
end