},

  buyer: nil

# Continual learning is off until an operator admits the named buyer, the base

# models, and the exact training code digest. The lane refuses rather than

# training on data or a model nobody admitted.

config :openagents, OpenAgents.ContinualLearning,

  enabled: false,

  buyer_ref: nil,

  buyer_class: "openagents_training",

  runtime_classes: ["standard", "strong"],

  admitted_base_models: %{},

  admitted_custody: ["openagents_managed"],

  maximum_rounds: 8,

  maximum_datasets: 4,

  wall_clock_ms: 900_000,

  maximum_state_bytes: 65_536,

  concurrency_limit: 1,

  training_code_digest: nil,

  trainer: OpenAgents.ContinualLearning.Trainer.Reference,

  evaluator: OpenAgents.ContinualLearning.Evaluator.Reference,

  class_watts: %{"standard" => 350, "strong" => 700, "batch" => 250},

  round_cost_usd_cents: %{"standard" => 2, "strong" => 4, "batch" => 1},

  settlement_unit: "usd_cents",

  outcome_repository: "OpenAgentsInc/openagents.com",

  outcome_issue_number: 86

# Import environment specific config. This must remain at the bottom

# of this file so it overrides the configuration defined above.

import_config "#{config_env()}.exs"

         buyer: capacity_buyer

       )

continual_learning_config = Application.get_env(:openagents, OpenAgents.ContinualLearning, [])

# The admitted base models are a JSON object of model reference to exact model

# digest. A model that is not listed here cannot be trained from, so widening

# the lane is an explicit operator act.

admitted_base_models =

  case optional_text.("OPENAGENTS_CONTINUAL_LEARNING_BASE_MODELS_JSON") do

    nil ->

      Keyword.get(continual_learning_config, :admitted_base_models, %{})

    encoded ->

      case Jason.decode(encoded) do

        {:ok, models} when is_map(models) ->

          models

        _invalid ->

          raise "environment variable OPENAGENTS_CONTINUAL_LEARNING_BASE_MODELS_JSON must be a JSON object"

      end

  end

config :openagents,

       OpenAgents.ContinualLearning,

       Keyword.merge(continual_learning_config,

         enabled: System.get_env("OPENAGENTS_CONTINUAL_LEARNING_ENABLED") == "true",

         buyer_ref: optional_text.("OPENAGENTS_CONTINUAL_LEARNING_BUYER_REF"),

         training_code_digest:

           optional_text.("OPENAGENTS_CONTINUAL_LEARNING_TRAINING_CODE_DIGEST"),

         admitted_base_models: admitted_base_models

       )

if config_env() == :dev do

  config :openagents, :openai_api_key, optional_text.("OPENAI_API_KEY")

  config :openagents, :openrouter_api_key, optional_text.("OPENROUTER_API_KEY")

defmodule OpenAgents.ContinualLearning do

  @moduledoc """

  Bounded continual-learning jobs over verified licensed datasets.

  One named internal buyer starts a job that names a versioned objective, an

  admitted base model, exact licensed dataset references, an evaluation corpus,

  a budget, a runtime class, and a stopping policy. Admission resolves every

  dataset through `OpenAgents.ArtifactCatalog`, so a job holds the exact

  artifact, provenance, license, and listing digests it trained on, and a

  removed listing, an expired license, a license that does not admit training,

  or a buyer class the listing was not licensed to refuses before any capacity

  is spent. Fleet admission is `OpenAgents.Capacity.match/2`: this lane adds no

  second scheduler, and the run itself is an ordinary `work_jobs` row of kind

  `continual_learning` driven by `OpenAgents.Work.ContinualLearningServer`.

  Every round writes a durable checkpoint before it is counted, so resume and

  replay are different acts: a resume continues the surviving checkpoint chain

  under the same admission digest, and a replay is a new job that starts from

  round zero. A lost checkpoint refuses the resume instead of retraining

  silently.

  Evaluation is graded through `OpenAgents.AcceptedOutcome`, under the admitted

  evaluator policy, so a failed, unevidenced, or non-independent evaluation

  cannot produce a qualified artifact. A qualified artifact binds the exact base

  model, dataset, code, configuration, checkpoint, and evaluation digests, and

  the job's settlement-ready receipt names the buyer, the unit, the amount, and

  the treasury policy without moving money.

  See `INVARIANTS.md`, CONTINUAL-001.

  """

  import Ecto.Query

  alias OpenAgents.AcceptedOutcome

  alias OpenAgents.Accounts

  alias OpenAgents.Accounts.User

  alias OpenAgents.ArtifactCatalog

  alias OpenAgents.Capacity

  alias OpenAgents.ContinualLearning.Artifact

  alias OpenAgents.ContinualLearning.Bounds

  alias OpenAgents.ContinualLearning.Checkpoint

  alias OpenAgents.ContinualLearning.Job

  alias OpenAgents.ContinualLearning.Receipt

  alias OpenAgents.Provenance.Canonical

  alias OpenAgents.Repo

  alias OpenAgents.Settlement

  alias OpenAgents.Work

  @active_statuses ~w(queued running)

  @evaluation_purpose "evaluation"

  @training_purpose "delivery"

  # ── admission ──────────────────────────────────────────────────────────────

  @doc """

  Admits and starts one continual-learning job for the named buyer.

  Returns `{:ok, job}` with a queued job whose run is already supervised, or a

  typed refusal.

  """

  @spec start(User.t(), map()) :: {:ok, Job.t()} | {:error, term()}

  def start(%User{} = user, attributes) when is_map(attributes) do

    with :ok <- feature_enabled(),

         :ok <- operator(user),

         {:ok, buyer_ref} <- buyer_ref(attributes),

         {:ok, buyer_class} <- buyer_class(),

         {:ok, objective} <- objective(attributes),

         {:ok, objective_version} <- objective_version(attributes),

         {:ok, base_model} <- base_model(attributes),

         {:ok, training_code_digest} <- training_code_digest(),

         {:ok, configuration} <- configuration(attributes),

         {:ok, runtime_class} <- runtime_class(attributes),

         {:ok, conversation_id} <- identifier(attributes, :conversation_id),

         {:ok, owner_visitor_id} <- identifier(attributes, :owner_visitor_id),

         {:ok, datasets} <- datasets(attributes, buyer_ref, buyer_class, runtime_class),

         {:ok, evaluation} <- evaluation(attributes, buyer_ref, buyer_class, runtime_class),

         {:ok, budget} <- budget(attributes),

         {:ok, stopping_policy} <- stopping_policy(attributes),

         :ok <- concurrency(),

         {:ok, capacity_receipt} <- capacity(user, runtime_class, budget, stopping_policy) do

      admission = %{

        buyer_ref: buyer_ref,

        buyer_class: buyer_class,

        objective: objective,

        objective_version: objective_version,

        base_model_ref: base_model.ref,

        base_model_digest: base_model.digest,

        training_code_digest: training_code_digest,

        configuration: configuration,

        configuration_digest: Canonical.digest!(configuration),

        datasets: datasets,

        evaluation: evaluation,

        budget: budget,

        runtime_class: runtime_class,

        capacity_receipt: capacity_receipt,

        stopping_policy: stopping_policy,

        replay_of_id: Map.get(attributes, :replay_of_id)

      }

      with {:ok, job} <- insert_job(admission),

           {:ok, started} <-

             launch(job, conversation_id, owner_visitor_id, "admission") do

        {:ok, started}

      end

    end

  end

  def start(_user, _attributes), do: {:error, :operator_required}

  @doc "One job the buyer may read, or a typed refusal."

  @spec get(User.t(), String.t()) :: {:ok, Job.t()} | {:error, term()}

  def get(%User{} = user, id) when is_binary(id) do

    with :ok <- operator(user), do: fetch(id)

  end

  def get(_user, _id), do: {:error, :operator_required}

  @doc "The buyer's most recent jobs, newest first, bounded."

  @spec list(User.t(), pos_integer()) :: {:ok, [Job.t()]} | {:error, term()}

  def list(%User{} = user, limit \\ 50) do

    with :ok <- operator(user) do

      bounded = min(max(limit, 1), 200)

      buyer_ref = Bounds.buyer_ref()

      {:ok,

       Job

       |> where([job], job.buyer_ref == ^buyer_ref)

       |> order_by([job], desc: job.inserted_at)

       |> limit(^bounded)

       |> Repo.all()}

    end

  end

  # ── lifecycle ──────────────────────────────────────────────────────────────

  @doc """

  Cancels one active job.

  The durable row reaches `cancelled` here, so the round loop stops at its next

  boundary even when the worker is already gone.

  """

  @spec cancel(User.t(), String.t()) :: {:ok, Job.t()} | {:error, term()}

  def cancel(%User{} = user, id) when is_binary(id) do

    with :ok <- operator(user),

         {:ok, job} <- fetch(id) do

      cond do

        job.status == "cancelled" ->

          {:ok, job}

        Job.terminal?(job) ->

          {:error, :not_cancellable}

        true ->

          if job.work_job_id, do: Work.cancel_job(job.work_job_id)

          terminalize(job, "cancelled", "cancelled")

      end

    end

  end

  def cancel(_user, _id), do: {:error, :operator_required}

  @doc """

  Resumes one interrupted or budget-exhausted job from its surviving checkpoint.

  A resume is not a replay: the job keeps its admission digest, its receipt

  chain, and its checkpoints, and continues at the next round. It refuses when

  the checkpoint is gone, when a dataset's license no longer admits the job, or

  when the fleet cannot admit the runtime class again.

  """

  @spec resume(User.t(), String.t(), map()) :: {:ok, Job.t()} | {:error, term()}

  def resume(%User{} = user, id, attributes \\ %{}) do

    with :ok <- feature_enabled(),

         :ok <- operator(user),

         {:ok, job} <- fetch(id),

         :ok <- resumable(job),

         {:ok, checkpoint} <- surviving_checkpoint(job),

         :ok <- rounds_remaining(job),

         :ok <- budget_remaining(job),

         :ok <- reverify_datasets(job),

         {:ok, capacity_receipt} <-

           capacity(user, job.runtime_class, job.budget, job.stopping_policy),

         {:ok, conversation_id, owner_visitor_id} <- previous_surface(job, attributes),

         {:ok, resumed} <- mark_resumed(job, checkpoint, capacity_receipt) do

      launch(resumed, conversation_id, owner_visitor_id, "resume")

    end

  end

  @doc """

  Replays one job as a new job under the same admitted inputs.

  A replay re-resolves every licensed dataset and the fleet again, starts at

  round zero, and records the job it replays, so a reproducibility check never

  reuses the original job's checkpoints.

  """

  @spec replay(User.t(), String.t(), map()) :: {:ok, Job.t()} | {:error, term()}

  def replay(%User{} = user, id, attributes) when is_map(attributes) do

    with :ok <- operator(user),

         {:ok, job} <- fetch(id),

         {:ok, conversation_id, owner_visitor_id} <- previous_surface(job, attributes) do

      start(

        user,

        replay_attributes(job, %{

          conversation_id: conversation_id,

          owner_visitor_id: owner_visitor_id

        })

      )

    end

  end

  @doc "Records one append-only receipt for a job."

  @spec record_receipt(Job.t(), String.t(), map()) :: {:ok, Receipt.t()} | {:error, term()}

  def record_receipt(%Job{} = job, kind, payload) when is_binary(kind) and is_map(payload) do

    sequence = Repo.aggregate(from(r in Receipt, where: r.job_id == ^job.id), :count) + 1

    body = Map.put(payload, "recorded_at", DateTime.to_iso8601(DateTime.utc_now()))

    %Receipt{job_id: job.id}

    |> Receipt.changeset(%{

      kind: kind,

      sequence: sequence,

      receipt_ref: "continual-learning-#{kind}:#{job.id}:#{sequence}",

      payload: body,

      digest: Canonical.digest!(Map.put(body, "job_id", job.id))

    })

    |> Repo.insert()

  end

  @doc "The bounded evidence export for one job."

  @spec export_evidence(User.t(), String.t()) :: {:ok, map()} | {:error, term()}

  def export_evidence(%User{} = user, id) when is_binary(id) do

    with :ok <- operator(user),

         {:ok, job} <- fetch(id) do

      {:ok,

       %{

         "schema" => "openagents.continual_learning_evidence.v1",

         "exported_at" => DateTime.utc_now(),

         "job" => projection(job),

         "checkpoints" => Enum.map(checkpoints(job), &checkpoint_projection/1),

         "receipts" => Enum.map(receipts(job), &receipt_projection/1),

         "artifact" => artifact_projection(artifact(job))

       }}

    end

  end

  @doc "The ordered checkpoint chain of one job."

  @spec checkpoints(Job.t()) :: [Checkpoint.t()]

  def checkpoints(%Job{} = job) do

    Checkpoint

    |> where([checkpoint], checkpoint.job_id == ^job.id)

    |> order_by([checkpoint], asc: checkpoint.round)

    |> Repo.all()

  end

  @doc "The ordered receipts of one job."

  @spec receipts(Job.t()) :: [Receipt.t()]

  def receipts(%Job{} = job) do

    Receipt

    |> where([receipt], receipt.job_id == ^job.id)

    |> order_by([receipt], asc: receipt.sequence)

    |> Repo.all()

  end

  @doc "The terminal artifact of one job, or `nil`."

  @spec artifact(Job.t()) :: Artifact.t() | nil

  def artifact(%Job{} = job), do: Repo.get_by(Artifact, job_id: job.id)

  @doc "The latest checkpoint of one job, or `nil`."

  @spec latest_checkpoint(Job.t()) :: Checkpoint.t() | nil

  def latest_checkpoint(%Job{} = job) do

    Checkpoint

    |> where([checkpoint], checkpoint.job_id == ^job.id)

    |> order_by([checkpoint], desc: checkpoint.round)

    |> limit(1)

    |> Repo.one()

  end

  @doc "How many continual-learning jobs are queued or running right now."

  @spec active_count() :: non_neg_integer()

  def active_count do

    Repo.aggregate(from(job in Job, where: job.status in ^@active_statuses), :count)

  end

  @doc "Reloads one job by id."

  @spec fetch(String.t()) :: {:ok, Job.t()} | {:error, :not_found}

  def fetch(id) when is_binary(id) do

    case Ecto.UUID.cast(id) do

      {:ok, uuid} ->

        case Repo.get(Job, uuid) do

          nil -> {:error, :not_found}

          job -> {:ok, job}

        end

      :error ->

        {:error, :not_found}

    end

  end

  @doc "Moves a job's lifecycle fields."

  @spec update_lifecycle(Job.t(), map()) :: {:ok, Job.t()} | {:error, term()}

  def update_lifecycle(%Job{} = job, attributes) when is_map(attributes) do

    job

    |> Job.lifecycle_changeset(attributes)

    |> Repo.update()

  end

  @doc """

  Terminalizes a job once. An already-terminal job is returned unchanged, so a

  cancel racing the round loop cannot rewrite the first terminal state.

  """

  @spec terminalize(Job.t(), String.t(), String.t() | nil) :: {:ok, Job.t()} | {:error, term()}

  def terminalize(%Job{} = job, status, error_code) do

    Repo.transaction(fn ->

      locked =

        Job

        |> where([row], row.id == ^job.id)

        |> lock("FOR UPDATE")

        |> Repo.one()

      cond do

        is_nil(locked) ->

          Repo.rollback(:not_found)

        Job.terminal?(locked) ->

          locked

        true ->

          locked

          |> Job.lifecycle_changeset(%{

            status: status,

            error_code: error_code,

            completed_at: DateTime.utc_now()

          })

          |> Repo.update()

          |> case do

            {:ok, updated} -> updated

            {:error, reason} -> Repo.rollback(reason)

          end

      end

    end)

  end

  @doc "The public projection of one job."

  @spec projection(Job.t()) :: map()

  def projection(%Job{} = job) do

    %{

      "id" => job.id,

      "buyer_ref" => job.buyer_ref,

      "buyer_class" => job.buyer_class,

      "objective" => job.objective,

      "objective_version" => job.objective_version,

      "base_model_ref" => job.base_model_ref,

      "base_model_digest" => job.base_model_digest,

      "training_code_digest" => job.training_code_digest,

      "configuration_digest" => job.configuration_digest,

      "datasets" => job.datasets,

      "evaluation" => job.evaluation,

      "budget" => job.budget,

      "runtime_class" => job.runtime_class,

      "capacity_receipt" => job.capacity_receipt,

      "stopping_policy" => job.stopping_policy,

      "admission_digest" => job.admission_digest,

      "status" => job.status,

      "error_code" => job.error_code,

      "rounds_completed" => job.rounds_completed,

      "resume_count" => job.resume_count,

      "usage" => job.usage,

      "work_job_id" => job.work_job_id,

      "replay_of_id" => job.replay_of_id,

      "started_at" => job.started_at,

      "completed_at" => job.completed_at

    }

  end

  # ── dataset admission ──────────────────────────────────────────────────────

  @doc """

  Resolves one licensed dataset reference into its exact binding.

  The listing must be available, licensed to the job's buyer class, licensed

  for the requested use, and licensed for the custody the runtime class

  provides, and the buyer must already hold an admitted acceptance receipt.

  """

  @spec bind_dataset(map(), String.t(), String.t(), String.t(), String.t()) ::

          {:ok, map()} | {:error, term()}

  def bind_dataset(reference, purpose, buyer_ref, buyer_class, runtime_class)

      when is_map(reference) do

    with {:ok, listing_id} <- reference_field(reference, "listing_id"),

         {:ok, acceptance_ref} <- reference_field(reference, "acceptance_ref"),

         {:ok, access} <- authorize(listing_id, purpose, buyer_ref, acceptance_ref),

         {:ok, listing} <- available_listing(listing_id),

         :ok <- licensed_buyer_class(listing, buyer_class),

         :ok <- licensed_use(listing, purpose),

         :ok <- licensed_custody(listing, runtime_class) do

      {:ok,

       %{

         "listing_id" => listing.id,

         "acceptance_ref" => acceptance_ref,

         "purpose" => purpose,

         "source_ref_digest" => Canonical.sha256(access.source_ref),

         "artifact_digest" => listing.artifact_digest,

         "provenance_digest" => listing.provenance_digest,

         "license_digest" => listing.license_digest,

         "listing_digest" => listing.listing_digest,

         "license_contract_ref" => listing.license_contract_ref,

         "license_expires_at" => DateTime.to_iso8601(listing.license_expires_at),

         "record_count" => listing.record_count

       }}

    end

  end

  defp datasets(attributes, buyer_ref, buyer_class, runtime_class) do

    references = Map.get(attributes, :datasets)

    cond do

      not is_list(references) or references == [] ->

        {:error, :datasets_required}

      length(references) > Bounds.maximum_datasets() ->

        {:error, :too_many_datasets}

      true ->

        bind_all(references, @training_purpose, buyer_ref, buyer_class, runtime_class)

    end

  end

  defp bind_all(references, purpose, buyer_ref, buyer_class, runtime_class) do

    Enum.reduce_while(references, {:ok, []}, fn reference, {:ok, bound} ->

      case bind_dataset(reference, purpose, buyer_ref, buyer_class, runtime_class) do

        {:ok, binding} -> {:cont, {:ok, bound ++ [binding]}}

        {:error, reason} -> {:halt, {:error, reason}}

      end

    end)

  end

  defp authorize(listing_id, purpose, buyer_ref, acceptance_ref) do

    case ArtifactCatalog.authorize_source_access(listing_id, %{

           purpose: purpose,

           buyer_ref: buyer_ref,

           acceptance_ref: acceptance_ref

         }) do

      {:ok, access} ->

        {:ok, access}

      # The catalog distinguishes a listing that is gone from one whose license

      # window closed, and the refusal has to keep that distinction.

      {:error, reason} when reason in [:not_found, :listing_removed, :stale_license] ->

        {:error, {:dataset_unavailable, reason}}

      {:error, reason} ->

        {:error, {:dataset_not_authorized, reason}}

    end

  end

  defp available_listing(listing_id) do

    case ArtifactCatalog.get_public_listing(listing_id) do

      {:ok, listing} -> {:ok, listing}

      {:error, reason} -> {:error, {:dataset_unavailable, reason}}

    end

  end

  defp licensed_buyer_class(listing, buyer_class) do

    if listing.buyer_class == buyer_class,

      do: :ok,

      else: {:error, {:dataset_buyer_class_mismatch, listing.id}}

  end

  defp licensed_use(listing, purpose) do

    terms = listing.license_terms || %{}

    allowed = List.wrap(terms["allowed_uses"])

    use_name = if purpose == @evaluation_purpose, do: "evaluation", else: "training"

    cond do

      terms["opt_in"] != true -> {:error, {:consent_missing, listing.id}}

      use_name not in allowed -> {:error, {:use_not_licensed, listing.id, use_name}}

      true -> :ok

    end

  end

  defp licensed_custody(listing, runtime_class) do

    location = data_location(runtime_class)

    licensed = List.wrap((listing.license_terms || %{})["data_locations"])

    cond do

      location not in Bounds.admitted_custody() ->

        {:error, {:unsupported_custody, location}}

      licensed != [] and location not in licensed ->

        {:error, {:unsupported_custody, listing.id}}

      true ->

        :ok

    end

  end

  defp reverify_datasets(%Job{} = job) do

    bindings = job.datasets ++ List.wrap(get_in(job.evaluation, ["corpus"]))

    Enum.reduce_while(bindings, :ok, fn binding, :ok ->

      case bind_dataset(

             binding,

             binding["purpose"],

             job.buyer_ref,

             job.buyer_class,

             job.runtime_class

           ) do

        {:ok, rebound} ->

          if rebound["license_digest"] == binding["license_digest"] and

               rebound["artifact_digest"] == binding["artifact_digest"] do

            {:cont, :ok}

          else

            {:halt, {:error, {:dataset_moved, binding["listing_id"]}}}

          end

        {:error, reason} ->

          {:halt, {:error, reason}}

      end

    end)

  end

  # ── evaluation admission ───────────────────────────────────────────────────

  defp evaluation(attributes, buyer_ref, buyer_class, runtime_class) do

    case Map.get(attributes, :evaluation) do

      evaluation when is_map(evaluation) ->

        admit_evaluation(evaluation, buyer_ref, buyer_class, runtime_class)

      _missing ->

        {:error, :evaluation_required}

    end

  end

  defp admit_evaluation(evaluation, buyer_ref, buyer_class, runtime_class) do

    with {:ok, corpus} <-

           corpus(evaluation, buyer_ref, buyer_class, runtime_class),

         {:ok, verifier} <- verifier(evaluation),

         {:ok, criteria} <- acceptance_criteria(evaluation),

         {:ok, target} <- target_metric(evaluation) do

      {:ok,

       %{

         "corpus" => corpus,

         "corpus_digest" => Canonical.digest!(Enum.map(corpus, & &1["artifact_digest"])),

         "verifier" => verifier,

         "separation_required" => evaluation[:separation_required] == true,

         "acceptance_criteria" => criteria,

         "target_metric" => target.metric,

         "target_value" => target.value,

         "policy_version" => Map.get(evaluation, :policy_version, 1)

       }}

    end

  end

  defp corpus(evaluation, buyer_ref, buyer_class, runtime_class) do

    references = Map.get(evaluation, :corpus)

    cond do

      not is_list(references) or references == [] ->

        {:error, :evaluation_corpus_required}

      length(references) > Bounds.maximum_datasets() ->

        {:error, :too_many_datasets}

      true ->

        bind_all(references, @evaluation_purpose, buyer_ref, buyer_class, runtime_class)

    end

  end

  defp verifier(evaluation) do

    verifier = Map.get(evaluation, :verifier)

    separation = evaluation[:separation_required] == true

    cond do

      not is_map(verifier) or not is_binary(verifier[:id]) ->

        {:error, :evaluator_required}

      verifier[:admitted] != true ->

        {:error, :evaluator_not_admitted}

      separation and verifier[:independent_of_producer] != true ->

        {:error, :evaluator_not_independent}

      true ->

        {:ok,

         %{

           "id" => verifier[:id],

           "admitted" => true,

           "independent_of_producer" => verifier[:independent_of_producer] == true,

           "policy_digest" => Canonical.digest!(%{"verifier" => verifier[:id]})

         }}

    end

  end

  defp acceptance_criteria(evaluation) do

    criteria = List.wrap(Map.get(evaluation, :acceptance_criteria))

    if criteria != [] and Enum.all?(criteria, &(is_binary(&1) and String.trim(&1) != "")) do

      {:ok, criteria}

    else

      {:error, :acceptance_criteria_required}

    end

  end

  defp target_metric(evaluation) do

    metric = Map.get(evaluation, :target_metric)

    value = Map.get(evaluation, :target_value)

    if is_binary(metric) and metric != "" and is_number(value) do

      {:ok, %{metric: metric, value: value}}

    else

      {:error, :evaluation_target_required}

    end

  end

  # ── other admission checks ─────────────────────────────────────────────────

  defp feature_enabled do

    if Bounds.enabled?(), do: :ok, else: {:error, :continual_learning_disabled}

  end

  defp operator(user) do

    if Accounts.admin?(user), do: :ok, else: {:error, :operator_required}

  end

  defp buyer_ref(attributes) do

    admitted = Bounds.buyer_ref()

    requested = Map.get(attributes, :buyer_ref)

    cond do

      not is_binary(admitted) or admitted == "" -> {:error, :buyer_not_configured}

      requested != admitted -> {:error, :buyer_not_admitted}

      true -> {:ok, admitted}

    end

  end

  defp buyer_class do

    case Bounds.buyer_class() do

      value when is_binary(value) and value != "" -> {:ok, value}

      _missing -> {:error, :buyer_not_configured}

    end

  end

  defp objective(attributes) do

    case Map.get(attributes, :objective) do

      value when is_binary(value) ->

        trimmed = String.trim(value)

        if trimmed != "" and byte_size(trimmed) <= 2_000,

          do: {:ok, trimmed},

          else: {:error, :objective_invalid}

      _missing ->

        {:error, :objective_invalid}

    end

  end

  defp objective_version(attributes) do

    case Map.get(attributes, :objective_version) do

      value when is_integer(value) and value > 0 -> {:ok, value}

      _invalid -> {:error, :objective_version_invalid}

    end

  end

  defp base_model(attributes) do

    admitted = Bounds.admitted_base_models()

    requested = Map.get(attributes, :base_model_ref)

    digest = Map.get(attributes, :base_model_digest)

    case Map.fetch(admitted, requested) do

      {:ok, admitted_digest} when is_binary(digest) and digest != admitted_digest ->

        {:error, :base_model_digest_mismatch}

      {:ok, admitted_digest} ->

        {:ok, %{ref: requested, digest: admitted_digest}}

      :error ->

        {:error, :base_model_not_admitted}

    end

  end

  defp training_code_digest do

    case Bounds.training_code_digest() do

      value when is_binary(value) -> {:ok, value}

      _missing -> {:error, :training_code_not_pinned}

    end

  end

  defp configuration(attributes) do

    case Map.get(attributes, :configuration, %{}) do

      value when is_map(value) ->

        if byte_size(Jason.encode!(value)) <= 8_192,

          do: {:ok, value},

          else: {:error, :configuration_too_large}

      _invalid ->

        {:error, :configuration_invalid}

    end

  end

  defp runtime_class(attributes) do

    requested = Map.get(attributes, :runtime_class)

    if is_binary(requested) and requested in Bounds.runtime_classes(),

      do: {:ok, requested},

      else: {:error, :runtime_class_not_admitted}

  end

  defp budget(attributes) do

    case Map.get(attributes, :budget) do

      %{} = budget ->

        amount = budget[:usd_cents] || budget["usd_cents"]

        if is_integer(amount) and amount > 0,

          do: {:ok, %{"unit" => "usd_cents", "amount" => amount}},

          else: {:error, :budget_invalid}

      _missing ->

        {:error, :budget_invalid}

    end

  end

  defp stopping_policy(attributes) do

    policy = Map.get(attributes, :stopping_policy)

    rounds = is_map(policy) && (policy[:maximum_rounds] || policy["maximum_rounds"])

    cond do

      not is_map(policy) ->

        {:error, :stopping_policy_required}

      not (is_integer(rounds) and rounds > 0) ->

        {:error, :stopping_policy_required}

      rounds > Bounds.maximum_rounds() ->

        {:error, :stopping_policy_exceeds_bound}

      true ->

        minimum_improvement = policy[:minimum_improvement] || policy["minimum_improvement"] || 0.0

        {:ok,

         %{

           "maximum_rounds" => rounds,

           "minimum_improvement" => minimum_improvement,

           "wall_clock_ms" => Bounds.wall_clock_ms()

         }}

    end

  end

  defp concurrency do

    if active_count() < Bounds.concurrency_limit(),

      do: :ok,

      else: {:error, :continual_learning_at_capacity}

  end

  defp capacity(user, runtime_class, budget, stopping_policy) do

    requirement = %{

      "quantity" => 1,

      "isolation" => isolation(runtime_class),

      "egress" => "policy_broker",

      "data_location" => data_location(runtime_class),

      "target" => "openagents_managed",

      "tools" => ["shell"],

      "duration_seconds" => duration_seconds(stopping_policy),

      "budget" => %{"currency" => "usd_cents", "amount" => budget["amount"]}

    }

    case Capacity.match(user, requirement) do

      {:ok, match} ->

        candidate = Enum.find(match["candidates"], &(&1["class"] == runtime_class))

        if candidate do

          {:ok,

           %{

             "schema" => match["schema"],

             "matched_at" => match["generated_at"],

             "requirement" => match["requirement"],

             "class" => candidate["class"],

             "rank" => candidate["rank"],

             "evidence" => candidate["evidence"],

             "estimate" => candidate["estimate"]

           }}

        else

          {:error, {:capacity_unavailable, runtime_class}}

        end

      {:error, %{"error" => %{"code" => code}}} ->

        {:error, {:capacity_unavailable, code}}

    end

  end

  defp isolation("strong"), do: "managed_strong"

  defp isolation(_class), do: "managed_standard"

  defp data_location(_class), do: "openagents_managed"

  defp duration_seconds(stopping_policy) do

    stopping_policy

    |> Map.get("wall_clock_ms", Bounds.wall_clock_ms())

    |> div(1_000)

    |> max(1)

  end

  defp identifier(attributes, key) do

    case Map.get(attributes, key) do

      value when is_binary(value) and value != "" -> {:ok, value}

      _missing -> {:error, :"#{key}_required"}

    end

  end

  # A dataset reference arrives either from the JSON API (string keys) or from

  # an internal caller (atom keys); both name the same admitted listing.

  defp reference_field(reference, "listing_id"),

    do: reference_value(reference, "listing_id", :listing_id)

  defp reference_field(reference, "acceptance_ref"),

    do: reference_value(reference, "acceptance_ref", :acceptance_ref)

  defp reference_value(reference, string_key, atom_key) do

    case Map.get(reference, string_key) || Map.get(reference, atom_key) do

      value when is_binary(value) and value != "" -> {:ok, value}

      _missing -> {:error, {:dataset_reference_invalid, string_key}}

    end

  end

  # ── insertion and launch ───────────────────────────────────────────────────

  defp insert_job(admission) do

    digest =

      Canonical.digest!(%{

        "buyer_ref" => admission.buyer_ref,

        "objective" => admission.objective,

        "objective_version" => admission.objective_version,

        "base_model_digest" => admission.base_model_digest,

        "training_code_digest" => admission.training_code_digest,

        "configuration_digest" => admission.configuration_digest,

        "dataset_digests" => Enum.map(admission.datasets, & &1["artifact_digest"]),

        "license_digests" => Enum.map(admission.datasets, & &1["license_digest"]),

        "evaluation_corpus_digest" => admission.evaluation["corpus_digest"],

        "verifier_policy_digest" => admission.evaluation["verifier"]["policy_digest"],

        "runtime_class" => admission.runtime_class,

        "stopping_policy" => admission.stopping_policy,

        "budget" => admission.budget

      })

    Repo.transaction(fn ->

      changeset =

        Job.admission_changeset(%Job{}, Map.put(admission, :admission_digest, digest))

      with {:ok, job} <- Repo.insert(changeset),

           {:ok, _receipt} <-

             record_receipt(job, "admission", %{

               "admission_digest" => job.admission_digest,

               "buyer_ref" => job.buyer_ref,

               "buyer_class" => job.buyer_class,

               "objective_version" => job.objective_version,

               "base_model_ref" => job.base_model_ref,

               "base_model_digest" => job.base_model_digest,

               "training_code_digest" => job.training_code_digest,

               "configuration_digest" => job.configuration_digest,

               "datasets" => job.datasets,

               "evaluation" => Map.drop(job.evaluation, ["corpus"]),

               "evaluation_corpus" => job.evaluation["corpus"],

               "runtime_class" => job.runtime_class,

               "capacity_receipt" => job.capacity_receipt,

               "budget" => job.budget,

               "stopping_policy" => job.stopping_policy,

               "replay_of_id" => job.replay_of_id

             }) do

        job

      else

        {:error, reason} -> Repo.rollback(reason)

      end

    end)

  end

  defp launch(%Job{} = job, conversation_id, owner_visitor_id, cause) do

    case Work.start_continual_learning(%{

           conversation_id: conversation_id,

           owner_visitor_id: owner_visitor_id,

           surface: "text",

           goal: job.objective,

           delegation: %{

             "continual_learning_job_id" => job.id,

             "admission_digest" => job.admission_digest,

             "cause" => cause,

             "resume_count" => job.resume_count

           },

           authority_snapshot: %{

             "buyer_ref" => job.buyer_ref,

             "buyer_class" => job.buyer_class,

             "runtime_class" => job.runtime_class,

             "base_model_ref" => job.base_model_ref,

             "base_model_digest" => job.base_model_digest,

             "training_code_digest" => job.training_code_digest,

             "verifier_id" => job.evaluation["verifier"]["id"]

           },

           budget_snapshot: Bounds.snapshot(job.runtime_class)

         }) do

      {:ok, work_job} ->

        update_lifecycle(job, %{work_job_id: work_job.id})

      {:error, reason} ->

        _refusal = record_receipt(job, "refusal", %{"reason" => inspect(reason)})

        _terminal = terminalize(job, "failed", "worker_start_failed")

        {:error, reason}

    end

  end

  defp resumable(%Job{} = job) do

    if Job.resumable?(job), do: :ok, else: {:error, :not_resumable}

  end

  defp surviving_checkpoint(%Job{} = job) do

    case latest_checkpoint(job) do

      nil -> {:error, :checkpoint_missing}

      %Checkpoint{lost: true} -> {:error, :checkpoint_lost}