defmodule Planck.Agent.Session do
@moduledoc """
Persistent session store backed by SQLite.
One GenServer per session, registered globally so any node in the cluster
can append messages or query history via transparent GenServer calls.
Each session writes to `<dir>/<id>_<name>.db`.
Both `id` and `name` appear in the filename so either can be resolved with
a single directory glob — see `find_by_id/2` and `find_by_name/2`.
## Usage
{:ok, _pid} = Planck.Agent.Session.start("a1b2c3d4", name: "crazy-mango", dir: "/path/to/sessions")
:ok = Planck.Agent.Session.append("my-session", "agent-1", message)
{:ok, rows} = Planck.Agent.Session.messages("my-session")
{:ok, rows} = Planck.Agent.Session.messages("my-session", agent_id: "agent-1")
Each row is `%{db_id: pos_integer(), agent_id: String.t(), message: Message.t(), inserted_at: integer()}`.
`db_id` is the SQLite autoincrement row id — use it with `truncate_after/2` to
anchor a truncation to a specific message.
Messages are serialized with `:erlang.term_to_binary/1` and read back with
`:erlang.binary_to_term/2` (`:safe` — no new atoms created from DB content).
`start/2` requires an explicit `:dir` option — the sessions directory is
resolved by the caller (typically `Planck.Headless` from its config).
## Distribution
Sessions are registered via `:global` as `{:session, session_id}`. Any node
in the Erlang cluster can call `append/3` or `messages/2` — the call is routed
transparently to the node that owns the session's SQLite file.
## Pagination
Messages with role `{:custom, :summary}` are stored as checkpoints
(`checkpoint = 1` in the DB). Two functions support cursor-based pagination
anchored on these checkpoints:
- `messages_from_latest_checkpoint/2` — initial load: latest checkpoint +
everything after. Returns `{:ok, rows, checkpoint_id | nil}`.
- `messages_before_checkpoint/3` — load more: the previous chapter.
Returns `{:ok, rows, prev_checkpoint_id | nil}`. `nil` means no more history.
Pass the returned `checkpoint_id` integer back as the cursor for the next page.
"""
use GenServer
require Logger
alias Planck.Agent.Message
@type session_id :: String.t()
@typedoc "A row returned by `messages/2` and related query functions."
@type row :: %{
db_id: pos_integer(),
agent_id: String.t(),
message: Message.t(),
inserted_at: integer()
}
defstruct [:id, :name, :conn]
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
@doc "Start a session under the SessionSupervisor."
@spec start(session_id(), keyword()) :: {:ok, pid()} | {:error, term()}
def start(session_id, opts \\ []) do
DynamicSupervisor.start_child(
Planck.Agent.SessionSupervisor,
{__MODULE__, Keyword.put(opts, :id, session_id)}
)
end
@doc "Stop a running session."
@spec stop(session_id()) :: :ok | {:error, :not_found | term()}
def stop(session_id) do
case whereis(session_id) do
{:ok, pid} -> DynamicSupervisor.terminate_child(Planck.Agent.SessionSupervisor, pid)
error -> error
end
end
@doc """
Append a message and return its DB row id. Returns `nil` if the session is
not found (agent has no persistent session).
"""
@spec append(session_id(), String.t(), Message.t()) :: pos_integer() | nil
def append(session_id, agent_id, message) do
case whereis(session_id) do
{:ok, pid} -> GenServer.call(pid, {:append, agent_id, message})
_ -> nil
end
end
@doc """
Retrieve messages for a session in insertion order.
Options:
- `agent_id:` — filter to messages from a specific agent
"""
@spec messages(session_id(), keyword()) :: {:ok, [row()]} | {:error, :not_found}
def messages(session_id, opts \\ []) do
case whereis(session_id) do
{:ok, pid} -> GenServer.call(pid, {:messages, opts})
error -> error
end
end
@doc """
Return the latest summary checkpoint and all messages after it.
If no checkpoint exists, returns all messages from the beginning.
The `checkpoint_id` in the return tuple is the DB row id of the checkpoint —
pass it to `messages_before_checkpoint/3` to load the previous page.
Options:
- `agent_id:` — filter to a specific agent
"""
@spec messages_from_latest_checkpoint(session_id(), keyword()) ::
{:ok, [row()], non_neg_integer() | nil} | {:error, :not_found}
def messages_from_latest_checkpoint(session_id, opts \\ []) do
case whereis(session_id) do
{:ok, pid} -> GenServer.call(pid, {:messages_from_latest_checkpoint, opts})
error -> error
end
end
@doc """
Return the chapter before a given checkpoint: the previous summary checkpoint
and all messages between it and `checkpoint_id`.
Returns `{:ok, rows, prev_checkpoint_id | nil}`. When `prev_checkpoint_id` is
`nil` there is no further history to load.
Options:
- `agent_id:` — filter to a specific agent
"""
@spec messages_before_checkpoint(session_id(), non_neg_integer(), keyword()) ::
{:ok, [row()], non_neg_integer() | nil} | {:error, :not_found}
def messages_before_checkpoint(session_id, checkpoint_id, opts \\ []) do
case whereis(session_id) do
{:ok, pid} -> GenServer.call(pid, {:messages_before_checkpoint, checkpoint_id, opts})
error -> error
end
end
@doc """
Delete all messages with a DB row id >= `db_id`, across all agents in the session.
Used when editing a previous message: truncates the session to strictly before
the given row, then the caller re-prompts with new text.
"""
@spec truncate_after(session_id(), pos_integer()) :: :ok | {:error, :not_found}
def truncate_after(session_id, db_id) do
case whereis(session_id) do
{:ok, pid} -> GenServer.call(pid, {:truncate_after, db_id})
error -> error
end
end
@doc """
Write key-value metadata for a session. Merges with any existing entries;
existing keys are overwritten. Values are stored as strings.
"""
@spec save_metadata(session_id(), map()) :: :ok | {:error, :not_found}
def save_metadata(session_id, metadata) do
case whereis(session_id) do
{:ok, pid} -> GenServer.call(pid, {:save_metadata, metadata})
error -> error
end
end
@doc "Return all metadata for a session as a `%{String.t() => String.t() | nil}` map."
@spec get_metadata(session_id()) ::
{:ok, %{optional(String.t()) => String.t() | nil}} | {:error, :not_found}
def get_metadata(session_id) do
case whereis(session_id) do
{:ok, pid} -> GenServer.call(pid, :get_metadata)
error -> error
end
end
@doc """
Resolve a session file by id. Globs `<sessions_dir>/<id>_*.db`.
Returns `{:ok, path, name}` or `{:error, :not_found}`.
"""
@spec find_by_id(Path.t(), String.t()) ::
{:ok, Path.t(), String.t()} | {:error, :not_found}
def find_by_id(sessions_dir, session_id) do
sessions_dir
|> Path.expand()
|> Path.join("#{session_id}_*.db")
|> Path.wildcard()
|> case do
[path | _] -> {:ok, path, parse_name(path)}
[] -> {:error, :not_found}
end
end
@doc """
Resolve a session file by name. Globs `<sessions_dir>/*_<name>.db`.
Returns `{:ok, path, session_id}` or `{:error, :not_found}`.
"""
@spec find_by_name(Path.t(), String.t()) ::
{:ok, Path.t(), String.t()} | {:error, :not_found}
def find_by_name(sessions_dir, name) do
sessions_dir
|> Path.expand()
|> Path.join("*_#{name}.db")
|> Path.wildcard()
|> case do
[path | _] -> {:ok, path, parse_id(path)}
[] -> {:error, :not_found}
end
end
@doc "Resolve a session id to its pid via `:global`."
@spec whereis(session_id()) :: {:ok, pid()} | {:error, :not_found}
def whereis(session_id) do
case :global.whereis_name({:session, session_id}) do
:undefined -> {:error, :not_found}
pid -> {:ok, pid}
end
end
@doc false
@spec start_link(keyword()) :: GenServer.on_start()
def start_link(opts) do
id = Keyword.fetch!(opts, :id)
GenServer.start_link(__MODULE__, opts, name: {:global, {:session, id}})
end
@doc false
@spec child_spec(keyword()) :: Supervisor.child_spec()
def child_spec(opts) do
%{
id: Keyword.fetch!(opts, :id),
start: {__MODULE__, :start_link, [opts]},
restart: :temporary
}
end
# ---------------------------------------------------------------------------
# GenServer callbacks
# ---------------------------------------------------------------------------
@impl true
def init(opts) do
id = Keyword.fetch!(opts, :id)
name = Keyword.fetch!(opts, :name)
dir = Keyword.fetch!(opts, :dir)
File.mkdir_p!(dir)
path = Path.join(dir, "#{id}_#{name}.db")
{:ok, conn} = Exqlite.Sqlite3.open(path)
:ok = create_tables(conn)
{:ok, %__MODULE__{id: id, name: name, conn: conn}}
end
@impl true
def handle_call(message, from, state)
def handle_call({:append, agent_id, message}, _from, state) do
db_id = insert_message(state.conn, agent_id, message)
{:reply, db_id, state}
end
def handle_call({:save_metadata, metadata}, _from, state) do
:ok = do_save_metadata(state.conn, metadata)
{:reply, :ok, state}
end
def handle_call(:get_metadata, _from, state) do
{:reply, {:ok, do_get_metadata(state.conn)}, state}
end
def handle_call({:messages, opts}, _from, state) do
{:reply, query_messages(state.conn, opts), state}
end
def handle_call({:messages_from_latest_checkpoint, opts}, _from, state) do
agent_id = Keyword.get(opts, :agent_id)
checkpoint_id = find_latest_checkpoint(state.conn, agent_id)
rows = query_rows_from(state.conn, checkpoint_id, agent_id)
{:reply, {:ok, rows, checkpoint_id}, state}
end
def handle_call({:messages_before_checkpoint, checkpoint_id, opts}, _from, state) do
agent_id = Keyword.get(opts, :agent_id)
prev_id = find_prev_checkpoint(state.conn, checkpoint_id, agent_id)
rows = query_rows_between(state.conn, prev_id, checkpoint_id, agent_id)
{:reply, {:ok, rows, prev_id}, state}
end
def handle_call({:truncate_after, db_id}, _from, state) do
{:ok, stmt} =
Exqlite.Sqlite3.prepare(state.conn, "DELETE FROM messages WHERE id >= ?1")
:ok = Exqlite.Sqlite3.bind(stmt, [db_id])
:done = Exqlite.Sqlite3.step(state.conn, stmt)
:ok = Exqlite.Sqlite3.release(state.conn, stmt)
{:reply, :ok, state}
end
@impl true
def terminate(reason, state)
def terminate(_reason, %{conn: conn}) when not is_nil(conn) do
Exqlite.Sqlite3.close(conn)
:ok
end
def terminate(_reason, _state), do: :ok
# ---------------------------------------------------------------------------
# Private helpers
# ---------------------------------------------------------------------------
@spec create_tables(Exqlite.Sqlite3.db()) :: :ok
defp create_tables(conn) do
:ok =
Exqlite.Sqlite3.execute(conn, """
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
agent_id TEXT NOT NULL,
data BLOB NOT NULL,
inserted_at INTEGER NOT NULL,
checkpoint INTEGER NOT NULL DEFAULT 0
)
""")
:ok =
Exqlite.Sqlite3.execute(conn, """
CREATE TABLE IF NOT EXISTS metadata (
key TEXT NOT NULL UNIQUE,
value TEXT
)
""")
:ok
end
@spec do_save_metadata(Exqlite.Sqlite3.db(), map()) :: :ok
defp do_save_metadata(conn, metadata) do
Enum.each(metadata, fn {key, value} ->
{:ok, stmt} =
Exqlite.Sqlite3.prepare(conn, """
INSERT INTO metadata (key, value) VALUES (?1, ?2)
ON CONFLICT(key) DO UPDATE SET value = excluded.value
""")
str_value = if is_nil(value), do: nil, else: to_string(value)
:ok = Exqlite.Sqlite3.bind(stmt, [to_string(key), str_value])
:done = Exqlite.Sqlite3.step(conn, stmt)
:ok = Exqlite.Sqlite3.release(conn, stmt)
end)
:ok
end
@spec do_get_metadata(Exqlite.Sqlite3.db()) :: %{optional(String.t()) => String.t() | nil}
defp do_get_metadata(conn) do
{:ok, stmt} = Exqlite.Sqlite3.prepare(conn, "SELECT key, value FROM metadata")
:ok = Exqlite.Sqlite3.bind(stmt, [])
rows = collect_metadata_rows(conn, stmt)
:ok = Exqlite.Sqlite3.release(conn, stmt)
Map.new(rows)
end
@spec collect_metadata_rows(Exqlite.Sqlite3.db(), Exqlite.Sqlite3.statement(), [
{String.t(), String.t() | nil}
]) :: [{String.t(), String.t() | nil}]
defp collect_metadata_rows(conn, stmt, acc \\ []) do
case Exqlite.Sqlite3.step(conn, stmt) do
{:row, [key, value]} -> collect_metadata_rows(conn, stmt, [{key, value} | acc])
:done -> acc
end
end
@spec parse_name(Path.t()) :: String.t()
defp parse_name(path) do
[_id, name] = path |> Path.basename(".db") |> String.split("_", parts: 2)
name
end
@spec parse_id(Path.t()) :: String.t()
defp parse_id(path) do
[id, _name] = path |> Path.basename(".db") |> String.split("_", parts: 2)
id
end
@spec insert_message(Exqlite.Sqlite3.db(), String.t(), Message.t()) :: pos_integer()
defp insert_message(conn, agent_id, message) do
# Strip the id before serialising — it is redundant since the DB row id
# is authoritative and set on every read in collect_rows.
data =
message
|> Map.drop([:id])
|> :erlang.term_to_binary()
now = System.system_time(:second)
checkpoint = if match?({:custom, :summary}, message.role), do: 1, else: 0
{:ok, stmt} =
Exqlite.Sqlite3.prepare(conn, """
INSERT INTO messages (agent_id, data, inserted_at, checkpoint) VALUES (?1, ?2, ?3, ?4)
""")
:ok = Exqlite.Sqlite3.bind(stmt, [agent_id, data, now, checkpoint])
:done = Exqlite.Sqlite3.step(conn, stmt)
:ok = Exqlite.Sqlite3.release(conn, stmt)
{:ok, row_id_stmt} = Exqlite.Sqlite3.prepare(conn, "SELECT last_insert_rowid()")
{:row, [db_id]} = Exqlite.Sqlite3.step(conn, row_id_stmt)
:ok = Exqlite.Sqlite3.release(conn, row_id_stmt)
db_id
end
@spec query_messages(Exqlite.Sqlite3.db(), keyword()) :: {:ok, [row()]}
defp query_messages(conn, opts) do
agent_id = Keyword.get(opts, :agent_id)
{:ok, query_rows_from(conn, nil, agent_id)}
end
@spec find_latest_checkpoint(Exqlite.Sqlite3.db(), String.t() | nil) ::
non_neg_integer() | nil
defp find_latest_checkpoint(conn, agent_id) do
{sql, params} =
if agent_id do
{"SELECT id FROM messages WHERE checkpoint = 1 AND agent_id = ?1 ORDER BY id DESC LIMIT 1",
[agent_id]}
else
{"SELECT id FROM messages WHERE checkpoint = 1 ORDER BY id DESC LIMIT 1", []}
end
fetch_one_id(conn, sql, params)
end
@spec find_prev_checkpoint(Exqlite.Sqlite3.db(), non_neg_integer(), String.t() | nil) ::
non_neg_integer() | nil
defp find_prev_checkpoint(conn, checkpoint_id, agent_id) do
{sql, params} =
if agent_id do
{"SELECT id FROM messages WHERE checkpoint = 1 AND id < ?1 AND agent_id = ?2 ORDER BY id DESC LIMIT 1",
[checkpoint_id, agent_id]}
else
{"SELECT id FROM messages WHERE checkpoint = 1 AND id < ?1 ORDER BY id DESC LIMIT 1",
[checkpoint_id]}
end
fetch_one_id(conn, sql, params)
end
@spec fetch_one_id(Exqlite.Sqlite3.db(), String.t(), list()) :: non_neg_integer() | nil
defp fetch_one_id(conn, sql, params) do
{:ok, stmt} = Exqlite.Sqlite3.prepare(conn, sql)
:ok = Exqlite.Sqlite3.bind(stmt, params)
result =
case Exqlite.Sqlite3.step(conn, stmt) do
{:row, [id]} -> id
:done -> nil
end
:ok = Exqlite.Sqlite3.release(conn, stmt)
result
end
@spec query_rows_from(Exqlite.Sqlite3.db(), non_neg_integer() | nil, String.t() | nil) ::
[row()]
defp query_rows_from(conn, from_id, agent_id) do
{sql, params} =
cond do
from_id && agent_id ->
{"SELECT id, agent_id, data, inserted_at FROM messages WHERE id >= ?1 AND agent_id = ?2 ORDER BY id",
[from_id, agent_id]}
from_id ->
{"SELECT id, agent_id, data, inserted_at FROM messages WHERE id >= ?1 ORDER BY id",
[from_id]}
agent_id ->
{"SELECT id, agent_id, data, inserted_at FROM messages WHERE agent_id = ?1 ORDER BY id",
[agent_id]}
true ->
{"SELECT id, agent_id, data, inserted_at FROM messages ORDER BY id", []}
end
run_query(conn, sql, params)
end
@spec query_rows_between(
Exqlite.Sqlite3.db(),
non_neg_integer() | nil,
non_neg_integer(),
String.t() | nil
) :: [row()]
defp query_rows_between(conn, from_id, before_id, agent_id) do
{sql, params} =
cond do
from_id && agent_id ->
{"SELECT id, agent_id, data, inserted_at FROM messages WHERE id >= ?1 AND id < ?2 AND agent_id = ?3 ORDER BY id",
[from_id, before_id, agent_id]}
from_id ->
{"SELECT id, agent_id, data, inserted_at FROM messages WHERE id >= ?1 AND id < ?2 ORDER BY id",
[from_id, before_id]}
agent_id ->
{"SELECT id, agent_id, data, inserted_at FROM messages WHERE id < ?1 AND agent_id = ?2 ORDER BY id",
[before_id, agent_id]}
true ->
{"SELECT id, agent_id, data, inserted_at FROM messages WHERE id < ?1 ORDER BY id",
[before_id]}
end
run_query(conn, sql, params)
end
@spec run_query(Exqlite.Sqlite3.db(), String.t(), list()) :: [row()]
defp run_query(conn, sql, params) do
{:ok, stmt} = Exqlite.Sqlite3.prepare(conn, sql)
:ok = Exqlite.Sqlite3.bind(stmt, params)
rows = collect_rows(conn, stmt)
:ok = Exqlite.Sqlite3.release(conn, stmt)
rows
end
@spec collect_rows(Exqlite.Sqlite3.db(), Exqlite.Sqlite3.statement(), [row()]) :: [row()]
defp collect_rows(conn, stmt, acc \\ []) do
case Exqlite.Sqlite3.step(conn, stmt) do
{:row, [db_id, agent_id, data, inserted_at]} ->
# binary_to_term restores the %Message{} struct (Map.drop preserved __struct__);
# Map.put adds back the :id that was stripped before serialization.
%Message{} = message = data |> :erlang.binary_to_term([:safe]) |> Map.put(:id, db_id)
collect_rows(conn, stmt, [
%{db_id: db_id, agent_id: agent_id, message: message, inserted_at: inserted_at} | acc
])
:done ->
Enum.reverse(acc)
end
end
end
