{:ok, Thread.t()} | {:error, :thread_terminal | Ecto.Changeset.t()}

  def record_event(%Thread{} = thread, event_type, payload)

      when is_binary(event_type) and is_map(payload) do

    case record_events(thread, [%{event_type: event_type, payload: payload}]) do

      {:ok, updated, _events} -> {:ok, updated}

      {:error, {_index, %Ecto.Changeset{} = changeset}} -> {:error, changeset}

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

    end

  end

  @doc """

  Append a batch of events to a thread's transcript, all or nothing.

  One transaction, insertion order preserved: either every entry lands, in the

  order given, or nothing does. A transcript with a hole in the middle

  describes a session that never happened, so one invalid entry rolls the whole

  batch back and the refusal names its position as `{index, changeset}`.

  The terminal refusal covers the whole batch for the same reason

  `record_event/3` refuses at all, and each committed event is broadcast as

  `{:thread_event, event}` in order after the transaction, exactly as a single

  append is, so a subscriber cannot tell a batch from the same events posted

  one at a time.

  The batch's size is the caller's to bound (`maximum_event_batch/0` is what

  the public route enforces); this function bounds only its shape.

  """

  @spec record_events(Thread.t(), [%{event_type: String.t(), payload: map()}]) ::

          {:ok, Thread.t(), [Event.t()]}

          | {:error,

             :thread_terminal | Ecto.Changeset.t() | {non_neg_integer(), Ecto.Changeset.t()}}

  def record_events(%Thread{} = thread, entries) when is_list(entries) and entries != [] do

    now = DateTime.utc_now()

    Repo.transaction(fn ->

      case locked(thread.id) do

        %Thread{status: "open"} = current ->

          with {:ok, event} <- insert_event(current, event_type, payload, now),

          with {:ok, events} <- insert_events(current, entries, now),

               {:ok, updated} <-

                 current |> Thread.event_count_changeset(current.event_count + 1) |> Repo.update() do

            {updated, event}

                 current

                 |> Thread.event_count_changeset(current.event_count + length(events))

                 |> Repo.update() do

            {updated, events}

          else

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

          end

      end

    end)

    |> case do

      {:ok, {updated, event}} ->

        Phoenix.PubSub.broadcast(OpenAgents.PubSub, topic(updated.id), {:thread_event, event})

        {:ok, updated}

      {:ok, {updated, events}} ->

        for event <- events do

          Phoenix.PubSub.broadcast(OpenAgents.PubSub, topic(updated.id), {:thread_event, event})

        end

        {:ok, updated, events}

      {:error, reason} ->

        {:error, reason}

    end

  end

  @doc "How many events one batch append may carry."

  @spec maximum_event_batch() :: pos_integer()

  def maximum_event_batch, do: setting(:maximum_thread_event_batch, 100)

  defp insert_events(thread, entries, now) do

    entries

    |> Enum.with_index()

    |> Enum.reduce_while({:ok, []}, fn {entry, index}, {:ok, inserted} ->

      case insert_event(thread, entry.event_type, entry.payload, now) do

        {:ok, event} -> {:cont, {:ok, [event | inserted]}}

        {:error, changeset} -> {:halt, {:error, {index, changeset}}}

      end

    end)

    |> case do

      {:ok, inserted} -> {:ok, Enum.reverse(inserted)}

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

    end

  end

  @doc """

  Subscribe to a thread's transcript appends.

    # and its own code — never a silent substitution.

    "model_unavailable" => {503, "The model's provider is not configured on this deployment"},

    "thread_terminal" => {422, "This thread is terminal and its transcript is closed"},

    # A transcript writer meets two refusals that cannot change: a closed

    # thread, and an event the server has called invalid. `thread_terminal`

    # already carries a code; these give the other refusals of the append route

    # the same property, so a client drops or splits without parsing prose.

    "event_invalid" => {422, "The event could not be recorded"},

    "event_batch_too_large" => {422, "The batch carries more events than the maximum"},

    # Spending the account's inference credit is not a rate limit: no amount of

    # waiting or revoking makes the same call succeed, so it is the payment

    # status and its own code.

  """

  @spec changeset(Plug.Conn.t(), Changeset.t(), keyword()) :: Plug.Conn.t()

  def changeset(conn, %Changeset{} = changeset, opts \\ []) do

    validation_failed(conn, Changeset.traverse_errors(changeset, &translate/1), opts)

    validation_failed(conn, changeset_errors(changeset), opts)

  end

  @doc """

  A changeset's errors as the envelope's field-to-messages map.

  For a route that refuses a changeset under its own code rather than the

  generic `validation_failed`: translate here, then pass the map to `refuse/3`

  as `:errors`.

  """

  @spec changeset_errors(Changeset.t()) :: map()

  def changeset_errors(%Changeset{} = changeset) do

    Changeset.traverse_errors(changeset, &translate/1)

  end

  @doc """

  end

  @doc """

  Append one event to a thread's transcript.

  Append to a thread's transcript: one event, or a batch of them.

  Append-only and bounded: the payload is capped by the database, and a

  terminal thread refuses, because a transcript that keeps growing after the

  report was written is not the transcript the report describes.

  One route serves both shapes — `{"event_type": ..., "payload": ...}` appends

  one event, `{"events": [...]}` appends a batch — because there is one door to

  a transcript and the batch is the same act performed fewer round trips at a

  time. A batch lands all-or-nothing in one transaction, in order, capped at

  `OpenAgents.Threads.maximum_event_batch/0`, and the created events come back

  in order so a client learns every id it just wrote.

  A refused event carries the stable code `event_invalid` beside the field

  errors, symmetric with `thread_terminal`, so a client tells a drop-only

  refusal from a retry-safe one without parsing prose.

  """

  def record(conn, %{"thread_id" => thread_id} = params) do

    with_thread(conn, thread_id, fn thread ->

      case event_parameters(params) do

        {:ok, event_type, payload} -> append(conn, thread, event_type, payload)

        {:refused, field, message} -> ApiError.validation_failed(conn, %{field => [message]})

      case Map.fetch(params, "events") do

        {:ok, events} -> record_batch(conn, thread, events)

        :error -> record_single(conn, thread, params)

      end

    end)

  end

  defp record_single(conn, thread, params) do

    case event_parameters(params) do

      {:ok, event_type, payload} -> append(conn, thread, event_type, payload)

      {:refused, field, message} -> event_invalid(conn, %{field => [message]})

    end

  end

  defp record_batch(conn, thread, events) do

    case batch_parameters(events) do

      {:ok, entries} -> append_batch(conn, thread, entries)

      {:refused, field, message} -> event_invalid(conn, %{field => [message]})

      {:oversized, count, cap} -> batch_too_large(conn, count, cap)

    end

  end

  def show(conn, %{"thread_id" => thread_id}) do

    with_thread(conn, thread_id, fn thread -> render_thread(conn, :ok, thread) end)

  end

    end)

  end

  # The created event is the point of the 201: its id is the cursor a client

  # continues from, and a writer that never learns it cannot dedup its own

  # append against a later read. The thread rides along for the count.

  defp append(conn, thread, event_type, payload) do

    case Threads.record_event(thread, event_type, payload) do

      {:ok, updated} ->

    case Threads.record_events(thread, [%{event_type: event_type, payload: payload}]) do

      {:ok, updated, [event]} ->

        conn

        |> put_extension_header()

        |> put_status(:created)

        |> json(%{"thread" => thread_view(updated)})

        |> json(%{"event" => event_view(event), "thread" => thread_view(updated)})

      {:error, :thread_terminal} ->

        sentence =

          "This thread is #{thread.status} and its transcript is closed. " <>

            "Open another thread to record more work."

        thread_terminal(conn, thread)

        ApiError.refuse(conn, "thread_terminal",

          message: sentence,

          errors: %{"thread" => [sentence]}

        )

      {:error, {_index, changeset}} ->

        event_invalid(conn, ApiError.changeset_errors(changeset))

      {:error, changeset} ->

        ApiError.changeset(conn, changeset)

      {:error, %Ecto.Changeset{} = changeset} ->

        event_invalid(conn, ApiError.changeset_errors(changeset))

    end

  end

  defp append_batch(conn, thread, entries) do

    case Threads.record_events(thread, entries) do

      {:ok, updated, events} ->

        conn

        |> put_extension_header()

        |> put_status(:created)

        |> json(%{"events" => Enum.map(events, &event_view/1), "thread" => thread_view(updated)})

      {:error, :thread_terminal} ->

        thread_terminal(conn, thread)

      {:error, {index, changeset}} ->

        errors =

          changeset

          |> ApiError.changeset_errors()

          |> Map.new(fn {field, messages} -> {"events[#{index}].#{field}", messages} end)

        event_invalid(conn, errors)

      {:error, %Ecto.Changeset{} = changeset} ->

        event_invalid(conn, ApiError.changeset_errors(changeset))

    end

  end

  defp thread_terminal(conn, thread) do

    sentence =

      "This thread is #{thread.status} and its transcript is closed. " <>

        "Open another thread to record more work."

    ApiError.refuse(conn, "thread_terminal",

      message: sentence,

      errors: %{"thread" => [sentence]}

    )

  end

  defp event_invalid(conn, errors) do

    ApiError.refuse(conn, "event_invalid", errors: errors)

  end

  defp batch_too_large(conn, count, cap) do

    sentence =

      "This batch carries #{count} events and the maximum is #{cap}. " <>

        "Split it and post the parts in order."

    ApiError.refuse(conn, "event_batch_too_large",

      message: sentence,

      errors: %{"events" => [sentence]}

    )

  end

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

  defp open(conn, objective, options) do

    end

  end

  # The whole batch is parsed before anything is appended, so a refusal names

  # the entry by its position and leaves nothing behind. An empty batch is

  # refused rather than answered 201: a client that posted nothing and read

  # "created" would believe something landed.

  defp batch_parameters(events) when is_list(events) do

    cap = Threads.maximum_event_batch()

    cond do

      events == [] ->

        {:refused, "events", "A batch appends at least one event."}

      length(events) > cap ->

        {:oversized, length(events), cap}

      true ->

        events

        |> Enum.with_index()

        |> Enum.reduce_while({:ok, []}, fn {event, index}, {:ok, entries} ->

          case batch_entry(event, index) do

            {:ok, entry} -> {:cont, {:ok, [entry | entries]}}

            {:refused, _field, _message} = refusal -> {:halt, refusal}

          end

        end)

        |> case do

          {:ok, entries} -> {:ok, Enum.reverse(entries)}

          {:refused, _field, _message} = refusal -> refusal

        end

    end

  end

  defp batch_parameters(_events) do

    {:refused, "events", "The events key carries an array of events."}

  end

  defp batch_entry(event, index) when is_map(event) do

    case event_parameters(event) do

      {:ok, event_type, payload} ->

        {:ok, %{event_type: event_type, payload: payload}}

      {:refused, field, message} ->

        {:refused, "events[#{index}].#{field}", message}

    end

  end

  defp batch_entry(event, index) do

    {:refused, "events[#{index}]", "#{inspect(event)} is not an object."}

  end

  defp event_type(%{"event_type" => event_type}) when is_binary(event_type) do

    if String.trim(event_type) == "" do

      {:refused, "event_type", "The event type names what happened and cannot be blank."}

      "schema" => event.schema,

      "event_type" => event.event_type,

      "payload" => event.payload,

      "emitted_at" => stamp(event.emitted_at)

      "emitted_at" => stamp(event.emitted_at),

      "inserted_at" => stamp(event.inserted_at)

    }

  end

    refute_receive {:thread_event, _event}

  end

  test "a committed batch broadcasts each event once, in order" do

    user = github_user("thread-broadcast-batch")

    {:ok, thread} = Threads.open(user, "Broadcast the batch")

    :ok = Threads.subscribe(thread)

    {:ok, _updated, events} =

      Threads.record_events(thread, [

        %{event_type: "turn.user", payload: %{"text" => "first"}},

        %{event_type: "tool.ran", payload: %{"tool" => "bash"}},

        %{event_type: "turn.assistant", payload: %{"text" => "third"}}

      ])

    # A subscriber cannot tell a batch from the same events posted one at a

    # time: one message per event, in the order they landed, none repeated.

    for event <- events do

      assert_receive {:thread_event, %Event{} = received}

      assert received.id == event.id

      assert received.event_type == event.event_type

    end

    refute_receive {:thread_event, _event}

  end

  test "a batch that rolls back broadcasts nothing" do

    user = github_user("thread-broadcast-rollback")

    {:ok, thread} = Threads.open(user, "Rolled back batches stay silent")

    :ok = Threads.subscribe(thread)

    assert {:error, {1, %Ecto.Changeset{}}} =

             Threads.record_events(thread, [

               %{event_type: "turn.user", payload: %{"text" => "valid"}},

               %{event_type: String.duplicate("x", 81), payload: %{}}

             ])

    # The first entry was inserted and then rolled back with the second, so a

    # subscriber must never have heard about it.

    refute_receive {:thread_event, _event}

    assert Threads.list_events(thread) |> Enum.map(& &1.event_type) == ["thread.opened"]

  end

  test "another thread's subscriber hears nothing" do

    user = github_user("thread-broadcast-scope")

    {:ok, mine} = Threads.open(user, "Mine")

    {:list_events, 2} => :thread_struct,

    {:list_for_user, 1} => :scoped_by_owner,

    {:list_for_user, 2} => :scoped_by_owner,

    {:maximum_event_batch, 0} => :no_thread,

    {:maximum_open_per_account, 0} => :no_thread,

    {:mint_grant, 1} => :returns_plaintext_token,

    {:open, 2} => :scoped_by_owner,

    {:open_count, 1} => :scoped_by_owner,

    {:reap_expired, 1} => :scoped_by_owner,

    {:record_event, 3} => :thread_struct,

    {:record_events, 2} => :thread_struct,

    {:subscribe, 1} => :thread_struct

  }

      assert body["thread"]["event_count"] > 0

    end

    test "returns the created event, whose id is the cursor", %{authenticated: conn, id: id} do

      body =

        conn

        |> post(~p"/api/v3/threads/#{id}/events", %{

          "event_type" => "turn.user",

          "payload" => %{"text" => "echo me back"}

        })

        |> json_response(201)

      # A writer that never learns its event's id cannot continue from it or

      # dedup its own append against a later read, so the 201 carries the event

      # rather than only the thread it landed on.

      assert is_integer(body["event"]["id"])

      assert body["event"]["event_type"] == "turn.user"

      assert body["event"]["payload"] == %{"text" => "echo me back"}

      assert is_binary(body["event"]["inserted_at"])

      read = conn |> get(~p"/api/v3/threads/#{id}/events") |> json_response(200)

      assert List.last(read["events"])["id"] == body["event"]["id"]

    end

    test "refuses an event with no type", %{authenticated: conn, id: id} do

      body =

        conn

        |> post(~p"/api/v3/threads/#{id}/events", %{"payload" => %{"text" => "x"}})

        |> json_response(422)

      # The code is the machine's half of the refusal, symmetric with

      # `thread_terminal`: a client drops the event without parsing prose.

      assert body["code"] == "event_invalid"

      assert body["errors"]["event_type"] != nil

    end

      assert body["code"] == "thread_terminal"

    end

    test "appends a batch in order and returns the created events", %{

      authenticated: conn,

      id: id

    } do

      before = conn |> get(~p"/api/v3/threads/#{id}") |> json_response(200)

      body =

        conn

        |> post(~p"/api/v3/threads/#{id}/events", %{

          "events" => [

            %{"event_type" => "turn.user", "payload" => %{"text" => "first"}},

            %{"event_type" => "tool.ran", "payload" => %{"tool" => "bash"}},

            %{"event_type" => "turn.assistant", "payload" => %{"text" => "third"}}

          ]

        })

        |> json_response(201)

      # A tool-heavy turn no longer costs one round trip per event, and the

      # created events come back in the order they landed so the writer learns

      # every id it just wrote.

      assert Enum.map(body["events"], & &1["event_type"]) ==

               ["turn.user", "tool.ran", "turn.assistant"]

      ids = Enum.map(body["events"], & &1["id"])

      assert ids == Enum.sort(ids)

      assert Enum.all?(body["events"], &is_binary(&1["inserted_at"]))

      assert body["thread"]["event_count"] == before["thread"]["event_count"] + 3

      read = conn |> get(~p"/api/v3/threads/#{id}/events") |> json_response(200)

      assert Enum.take(read["events"], -3) |> Enum.map(& &1["id"]) == ids

    end

    test "a batch with one invalid event records nothing", %{authenticated: conn, id: id} do

      before = conn |> get(~p"/api/v3/threads/#{id}") |> json_response(200)

      # The second entry passes the route's parse — its type is non-blank — and

      # is refused by the database's 80-character ceiling, so the refusal

      # proves the transaction rolled the first entry back with it.

      body =

        conn

        |> post(~p"/api/v3/threads/#{id}/events", %{

          "events" => [

            %{"event_type" => "turn.user", "payload" => %{"text" => "landed?"}},

            %{"event_type" => String.duplicate("x", 81), "payload" => %{}}

          ]

        })

        |> json_response(422)

      assert body["code"] == "event_invalid"

      assert body["errors"]["events[1].event_type"] != nil

      after_refusal = conn |> get(~p"/api/v3/threads/#{id}") |> json_response(200)

      assert after_refusal["thread"]["event_count"] == before["thread"]["event_count"]

    end

    test "a batch entry with no type is refused naming its position", %{

      authenticated: conn,

      id: id

    } do

      body =

        conn

        |> post(~p"/api/v3/threads/#{id}/events", %{

          "events" => [

            %{"event_type" => "turn.user"},

            %{"payload" => %{"text" => "no type"}}

          ]

        })

        |> json_response(422)

      assert body["code"] == "event_invalid"

      assert body["errors"]["events[1].event_type"] != nil

    end

    test "an empty batch is refused rather than answered created", %{

      authenticated: conn,

      id: id

    } do

      body =

        conn

        |> post(~p"/api/v3/threads/#{id}/events", %{"events" => []})

        |> json_response(422)

      assert body["code"] == "event_invalid"

      assert body["errors"]["events"] != nil

    end

    test "a batch over the cap is refused with its own code", %{authenticated: conn, id: id} do

      previous = Application.get_env(:openagents, :maximum_thread_event_batch)

      Application.put_env(:openagents, :maximum_thread_event_batch, 2)

      on_exit(fn ->

        Application.put_env(:openagents, :maximum_thread_event_batch, previous)

      end)

      body =

        conn

        |> post(~p"/api/v3/threads/#{id}/events", %{

          "events" =>

            for index <- 1..3 do

              %{"event_type" => "turn.user", "payload" => %{"index" => index}}

            end

        })

        |> json_response(422)

      # Over the cap is not an invalid event — every entry may be well formed —

      # so it carries its own code, and the sentence names the split.

      assert body["code"] == "event_batch_too_large"

      assert body["message"] =~ "2"

      assert [sentence] = body["errors"]["events"]

      assert sentence =~ "3 events"

    end

    test "refuses a batch to a revoked thread as one refusal", %{authenticated: conn, id: id} do

      conn |> delete(~p"/api/v3/threads/#{id}") |> json_response(200)

      body =

        conn

        |> post(~p"/api/v3/threads/#{id}/events", %{

          "events" => [

            %{"event_type" => "turn.user", "payload" => %{"text" => "late"}},

            %{"event_type" => "turn.assistant", "payload" => %{"text" => "later"}}

          ]

        })

        |> json_response(422)

      assert body["code"] == "thread_terminal"

    end

    test "does not read another account's transcript", %{authenticated: conn, id: id} do

      conn

      |> post(~p"/api/v3/threads/#{id}/events", %{"event_type" => "turn.user"})