feat(s012): the session process and the agent loop

One gen_statem per conversation under a DynamicSupervisor and a Registry:
idle, thinking, tool_wait, approval_wait (stub), compacting (stub), error.
A user message is a row, then a broadcast, then a model call in a Task
under the Session's own supervisor, talking back only by message; deltas
coalesce to at most 20 broadcasts a second and persist as a draft row every
500 ms or 2 KB; the draft is finalised at done (the one edit docs/05 now
names); a tool call runs the ToolRunner stub in a Task, writes a tool row,
and the loop continues under code-owned caps; a crash or provider error is
an error turn with the partial text kept; cancel persists what arrived as
interrupted; rehydrate marks a leftover draft interrupted and never resumes;
idle hibernates and stops on generic timeouts. Trinity.CorePolicy.hash/0 and
the sentinel (three families, tighten only) arrive here as the spec asks.

The 010 schema Trinity.Sessions.Session is renamed SessionRow (fix
referencing 010): docs/01 gives that name to the process. Seventeen tests,
including the two crash tests and 100 concurrent sessions with unchanged
pids; the fake provider gained global state and script sequences after
the first runs showed its process-local state was invisible to a
session's Task. Coverage 60.82%. docs/01 and docs/05 synced. G1 lines 1
to 14.

Signed-off-by: Ayla Croft <aylacroft@proton.me>

Author: HackTuah

ROADMAP.md

@@ -35,7 +35,7 @@ standards register names the rows that ask for them.
 | 003 | FIPS build leg in CI, from source | 0 Foundation | M | 000 | planned |
 | 010 | Core domain + persistence (Ecto/SQLite, schemas, Repo owner) | 1 Core loop | M | 000 | approved |
 | 011 | LLM provider layer (req_llm behind `Trinity.LLM` behaviour) | 1 Core loop | M | 010 | approved |
-| 012 | Session process + agent loop (gen_statem, DynamicSupervisor, rehydration) | 1 Core loop | L | 010, 011 | in_progress |
+| 012 | Session process + agent loop (gen_statem, DynamicSupervisor, rehydration) | 1 Core loop | L | 010, 011 | done |
 | 013 | LiveView chat UI with streaming | 1 Core loop | M | 012 | planned |
 | 020 | Tool protocol + registry | 2 Tools | M | 012 | planned |
 | 021 | Permission gate + approval UI (M2 fingerprint-bound, M7) | 2 Tools | M | 020, 013 | planned |

config/test.exs

@@ -7,6 +7,9 @@ import Config
 # The MIX_TEST_PARTITION environment variable can be used
 # to provide built-in test partitioning in CI environment.
 # Run `mix help test` for more information.
+# Slice 012: sessions hibernate and stop quickly in tests so AC8 is observable in seconds.
+config :trinity, :sessions, idle_hibernate_ms: 200, idle_stop_ms: 60_000
+
 # Slice 011: the registry in tests is the scripted fake plus a Mox mock; the live tests set
 # their own entries from the environment at runtime.
 config :trinity, :llm,

coverage.tsv

@@ -3,3 +3,4 @@ slice_id	percent	sha	date
 001	30.37	5a9c8f7	2026-09-06
 010	44.88	45ba4f0	2026-09-20
 011	51.57	ec5334a	2026-09-20
+012	60.82	019d9cf	2026-09-20

docs/01-architecture.md

@@ -16,11 +16,13 @@ Trinity.Application
 ├── Trinity.Repo                                  # Ecto (SQLite primary). Slice 010
 ├── {Phoenix.PubSub, name: Trinity.PubSub}        # all fan-out. Slice 010
 ├── Trinity.Telemetry                             # metrics + cost ledger. Slice 090
-├── {Registry, keys: :unique, name: Trinity.Registry}
-├── Trinity.LLM.Supervisor                        # provider clients, rate limiters. Slice 011
-├── Trinity.Sessions.Supervisor (DynamicSupervisor) # one Trinity.Sessions.Session per conversation. Slice 012
-│     └── Trinity.Sessions.Session (gen_statem)   # states: idle → thinking → tool_wait → approval_wait → compacting
-│           └── Trinity.Sessions.TurnTaskSupervisor (Task.Supervisor, per session) # parallel tool calls
+├── {Registry, keys: :unique, name: Trinity.Registry}   # Slice 012, as built
+├── {Task.Supervisor, name: Trinity.LLM.TaskSupervisor}  # Slice 011, as built: stream_to/3 runs here.
+│                                                 # Trinity.LLM.Supervisor (rate limiters) is not built:
+│                                                 # nothing needs a process yet (011 NOTES, follow-up)
+├── Trinity.Sessions.Supervisor (DynamicSupervisor) # one Trinity.Sessions.Session per conversation. Slice 012, as built
+│     └── Trinity.Sessions.Session (gen_statem)   # states: idle → thinking → tool_wait → approval_wait → compacting → error
+│           └── Task.Supervisor (started by the Session, linked, unnamed) # the model call and the tool calls of one turn
 ├── Trinity.Tools.Supervisor                      # tool runtime (ports, browsers). Slice 020/022
 ├── Trinity.Permissions.Gate                      # approval requests + allowlist cache. Slice 021
 ├── Trinity.Receipts.Supervisor                   # Slice 024
@@ -126,7 +128,7 @@ Every state transition is persisted before it is broadcast. A crash between pers
 - Session state = `%Session.State{}` struct, rebuilt from DB on init; in-memory only for the active turn.
 - Never block a Session on I/O: LLM streaming, tool execution, embedding happen in Tasks; Session receives messages.
 - PubSub topics: `session:<id>` (turn events), `approvals:<id>`, `gateway:<adapter>`, `system`.
-- Backpressure: stream chunks are coalesced to ≤ 20 broadcasts/sec per session (Slice 013).
+- Backpressure: stream chunks are coalesced to ≤ 20 broadcasts/sec per session (built at Slice 012: a 50 ms timer in the Session, so 013 receives coalesced deltas).
 
 ## Directory layout
 

docs/05-data-model.md

@@ -39,7 +39,10 @@ with adapter-specific `execute/1` guarded by `repo().__adapter__()`.
 | tool_call_id | string, nullable | |
 | usage | map, nullable | prompt/completion tokens, cost |
 | provider_meta | map | model, finish reason, latency |
-Append-only. Editing is a new message with `parts.supersedes`.
+Append-only. Editing is a new message with `parts.supersedes`. One edit is allowed and named (slice 012): an
+assistant row written as a draft during a turn (`parts.draft = true`, content updated every 500 ms or 2 KB) becomes
+final at the end of the turn (`draft = false`, plus `tool_calls`, `usage`, and `interrupted`, `error` or `cap` when the
+turn ended that way); the role, seq and session never change.
 
 ### messages_fts (Slice 031): SQLite `fts5(content, session_id UNINDEXED, message_id UNINDEXED)`; on Postgres a
 `tsvector` generated column on `messages`.

lib/trinity/application.ex

@@ -35,6 +35,9 @@ defmodule Trinity.Application do
           {Phoenix.PubSub, name: Trinity.PubSub},
           # Slice 011: streams to a pid run under this supervisor, never as bare tasks.
           {Task.Supervisor, name: Trinity.LLM.TaskSupervisor},
+          # Slice 012: one session process per conversation, found by id.
+          {Registry, keys: :unique, name: Trinity.Registry},
+          Trinity.Sessions.Supervisor,
           # Start to serve requests, typically the last entry
           TrinityWeb.Endpoint
         ] ++ Trinity.Smoke.children(Trinity.Smoke.argv())

lib/trinity/core_policy.ex

@@ -0,0 +1,35 @@
+# SPDX-FileCopyrightText: Sudo Apt Holdings LLC
+# SPDX-License-Identifier: Apache-2.0
+defmodule Trinity.CorePolicy do
+  @moduledoc """
+  A digest over the object code of the modules that decide what the agent may do. Slice 012
+  introduces it because AC9 (kill and reseed) asserts it; slice 024 extends the list and writes it
+  into the boot receipt. A reseeded Session is born from this hash and nothing else in process
+  state.
+  """
+
+  @modules [
+    Trinity.Sessions.Session,
+    Trinity.Sessions.Caps,
+    Trinity.Sessions.ToolRunner,
+    Trinity.Sessions.ToolRunner.Stub,
+    Trinity.Sessions.Sentinel
+  ]
+
+  @doc "The modules the hash covers, in order."
+  @spec modules() :: [module()]
+  def modules, do: @modules
+
+  @doc "SHA-256, hex, over the concatenated object code of `modules/0`."
+  @spec hash() :: String.t()
+  def hash do
+    @modules
+    |> Enum.map(fn mod ->
+      {^mod, binary, _path} = :code.get_object_code(mod)
+      binary
+    end)
+    |> IO.iodata_to_binary()
+    |> then(&:crypto.hash(:sha256, &1))
+    |> Base.encode16(case: :lower)
+  end
+end

lib/trinity/sessions.ex

@@ -8,9 +8,12 @@ defmodule Trinity.Sessions do
   exports this module alone; `Trinity.Sessions.Store` and the schemas stay inside. Slice 012
   adds the session process on top of this API and changes nothing here.
   """
-  use Boundary, deps: [Trinity], exports: []
+  # Slice 012: Sessions reaches the LLM (docs/01: Sessions depends on LLM, Repo, PubSub).
+  use Boundary,
+    deps: [Trinity, Trinity.LLM],
+    exports: [Events, Message, Persona, SessionRow, Session, Caps]
 
-  alias Trinity.Sessions.{Message, Persona, Session, Store}
+  alias Trinity.Sessions.{Message, Persona, SessionRow, Store}
 
   @type session_id :: String.t()
 
@@ -23,15 +26,15 @@ defmodule Trinity.Sessions do
   def get_persona_by_name(name), do: Store.get_persona_by_name(name)
 
   @doc "Creates a session. `persona_id` is required; `origin` and `status` come from a closed vocabulary."
-  @spec create_session(map()) :: {:ok, Session.t()} | {:error, Ecto.Changeset.t()}
+  @spec create_session(map()) :: {:ok, SessionRow.t()} | {:error, Ecto.Changeset.t()}
   def create_session(attrs), do: Store.insert_session(attrs)
 
   @doc "The session with this id, or nil."
-  @spec get_session(session_id()) :: Session.t() | nil
+  @spec get_session(session_id()) :: SessionRow.t() | nil
   def get_session(id), do: Store.get_session(id)
 
   @doc "Sessions, most recently active first. Options: `status:`, `limit:` (default 50)."
-  @spec list_sessions(keyword()) :: [Session.t()]
+  @spec list_sessions(keyword()) :: [SessionRow.t()]
   def list_sessions(opts \\ []), do: Store.list_sessions(opts)
 
   @doc """
@@ -56,8 +59,8 @@ defmodule Trinity.Sessions do
   def history(session_id, opts \\ []), do: Store.history(session_id, opts)
 
   @doc "Marks a session archived."
-  @spec archive(Session.t()) :: {:ok, Session.t()} | {:error, Ecto.Changeset.t()}
-  def archive(%Session{} = session), do: Store.update_session(session, %{status: "archived"})
+  @spec archive(SessionRow.t()) :: {:ok, SessionRow.t()} | {:error, Ecto.Changeset.t()}
+  def archive(%SessionRow{} = session), do: Store.update_session(session, %{status: "archived"})
 
   @doc "The number of messages in a session."
   @spec message_count(session_id()) :: non_neg_integer()
@@ -66,4 +69,58 @@ defmodule Trinity.Sessions do
   @doc "Every `seq` in a session, ascending. The stress test's population."
   @spec seqs(session_id()) :: [pos_integer()]
   def seqs(session_id), do: Store.seqs(session_id)
+
+  ## The process (slice 012)
+
+  alias Trinity.Sessions.{Events, Session, Supervisor}
+
+  @doc "Starts the session's process, or returns the running one. The row must exist."
+  @spec start_session(session_id()) :: {:ok, pid()} | {:error, term()}
+  def start_session(session_id), do: Supervisor.start_session(session_id)
+
+  @doc "Idempotent: the running pid, or a fresh process rehydrated from the database."
+  @spec ensure_started(session_id()) :: {:ok, pid()} | {:error, term()}
+  def ensure_started(session_id) do
+    case whereis(session_id) do
+      nil -> start_session(session_id)
+      pid -> {:ok, pid}
+    end
+  end
+
+  @doc "The session's pid, if its process is running."
+  @spec whereis(session_id()) :: pid() | nil
+  def whereis(session_id) do
+    case Registry.lookup(Trinity.Registry, session_id) do
+      [{pid, _}] -> pid
+      [] -> nil
+    end
+  end
+
+  @doc "Persists the user's message and starts a turn; refuses while a turn is in flight."
+  @spec send_user_message(session_id(), String.t()) :: {:ok, Message.t()} | {:error, term()}
+  def send_user_message(session_id, content) do
+    with {:ok, pid} <- ensure_started(session_id), do: Session.send_user_message(pid, content)
+  end
+
+  @doc "Stops the turn in flight, persisting what arrived as interrupted."
+  @spec cancel_turn(session_id()) :: :ok | {:error, term()}
+  def cancel_turn(session_id) do
+    case whereis(session_id) do
+      nil -> {:error, :not_running}
+      pid -> Session.cancel_turn(pid)
+    end
+  end
+
+  @doc "The state name and a redacted view of the process's data."
+  @spec state(session_id()) :: map() | {:error, :not_running}
+  def state(session_id) do
+    case whereis(session_id) do
+      nil -> {:error, :not_running}
+      pid -> Session.state(pid)
+    end
+  end
+
+  @doc "Subscribes the caller to the session's events on `session:<id>`."
+  @spec subscribe(session_id()) :: :ok | {:error, term()}
+  def subscribe(session_id), do: Events.subscribe(session_id)
 end

lib/trinity/sessions/caps.ex

@@ -0,0 +1,42 @@
+# SPDX-FileCopyrightText: Sudo Apt Holdings LLC
+# SPDX-License-Identifier: Apache-2.0
+defmodule Trinity.Sessions.Caps do
+  @moduledoc """
+  Code-owned caps on the agent loop (M3). Slice 012. Module attributes, not configuration: no
+  config key can raise them, and the loop function that consults them takes no cap argument.
+  Reaching a cap is a recorded outcome and a normal return to `idle`, never a crash.
+  """
+
+  @max_turns_per_message 8
+  @max_tokens_per_message 200_000
+  @max_wall_ms 600_000
+
+  @type reason :: :max_turns | :max_tokens | :max_wall_ms
+
+  @doc "The caps, for the record a cap outcome writes and for tests."
+  @spec limits() :: %{
+          max_turns: pos_integer(),
+          max_tokens: pos_integer(),
+          max_wall_ms: pos_integer()
+        }
+  def limits,
+    do: %{
+      max_turns: @max_turns_per_message,
+      max_tokens: @max_tokens_per_message,
+      max_wall_ms: @max_wall_ms
+    }
+
+  @doc """
+  `:ok` when the turn may continue, or the first cap it has reached. Takes the turn record only:
+  there is no argument through which a caller could pass a looser limit.
+  """
+  @spec check(Trinity.Sessions.State.turn()) :: :ok | {:cap, reason()}
+  def check(%{turns: turns, tokens: tokens, started_at: started_at}) do
+    cond do
+      turns >= @max_turns_per_message -> {:cap, :max_turns}
+      tokens >= @max_tokens_per_message -> {:cap, :max_tokens}
+      System.monotonic_time(:millisecond) - started_at >= @max_wall_ms -> {:cap, :max_wall_ms}
+      true -> :ok
+    end
+  end
+end

lib/trinity/sessions/events.ex

@@ -0,0 +1,49 @@
+# SPDX-FileCopyrightText: Sudo Apt Holdings LLC
+# SPDX-License-Identifier: Apache-2.0
+defmodule Trinity.Sessions.Events do
+  @moduledoc """
+  The seven event shapes a Session broadcasts on `session:<id>`, and nothing else. Slice 012.
+  Slice 013's UI and slice 070's gateways subscribe here. `broadcast/2` refuses a shape that is
+  not one of the seven, so a new event is a change to this file first.
+  """
+
+  alias Trinity.Sessions.Message
+
+  @type t ::
+          {:user_message, Message.t()}
+          | {:assistant_delta, String.t()}
+          | {:assistant_message, Message.t()}
+          | {:tool_call, map()}
+          | {:state, atom()}
+          | {:turn_interrupted, Message.t()}
+          | {:error, term()}
+
+  @doc "The PubSub topic for a session."
+  @spec topic(String.t()) :: String.t()
+  def topic(session_id), do: "session:" <> session_id
+
+  @doc "True for exactly the seven shapes."
+  @spec valid?(term()) :: boolean()
+  def valid?({:user_message, %Message{}}), do: true
+  def valid?({:assistant_delta, s}) when is_binary(s), do: true
+  def valid?({:assistant_message, %Message{}}), do: true
+  def valid?({:tool_call, %{id: _, name: _}}), do: true
+  def valid?({:state, s}) when is_atom(s), do: true
+  def valid?({:turn_interrupted, %Message{}}), do: true
+  def valid?({:error, _}), do: true
+  def valid?(_), do: false
+
+  @doc "Broadcasts one event; raises on a shape that is not one of the seven."
+  @spec broadcast(String.t(), t()) :: :ok
+  def broadcast(session_id, event) do
+    if valid?(event) do
+      Phoenix.PubSub.broadcast(Trinity.PubSub, topic(session_id), {:session, session_id, event})
+    else
+      raise ArgumentError, "not a session event: #{inspect(event)}"
+    end
+  end
+
+  @doc "Subscribes the calling process to a session's events."
+  @spec subscribe(String.t()) :: :ok | {:error, term()}
+  def subscribe(session_id), do: Phoenix.PubSub.subscribe(Trinity.PubSub, topic(session_id))
+end