Pixir.Session (pixir v0.1.13)

Copy Markdown View Source

The unit of agency (ADR 0001): a single GenServer that owns one conversation — its :role (the Agent configuration), its monotonic seq counter, and the append to its Log. There is exactly one Session process per session_id, registered in Pixir.Sessions.Registry.

Recording events

Canonical events go through record/2, which runs the load-bearing sequence inside the GenServer (so it is serialized): stamp seq → append to the Log → publish on the bus. The Log is the source of truth, so an event that fails to persist is not published. Ephemeral events go through emit/2 (publish only — no seq, no Log).

Turns as supervised Tasks

A Turn runs in a Task under Pixir.TurnSupervisor, monitored (not linked) by the Session. interrupt/1 kills that Task — the load-bearing invariant that makes a Turn cleanly cancellable without taking the Session down. The Turn body is a 1-arity function given a context map (%{session_id, workspace, role, fork_root_session_id}); the real tool-loop (build step 7) plugs in here.

Resume

On init/1 the Session folds its existing Log to seed seq (so new canonical events continue the sequence). History itself is always re-derived from the Log on demand (history/1), never held as authoritative state (ADR 0003). A Session also acquires a filesystem-backed writer lease so a second OS process cannot become a competing Log writer while this process is alive.

Summary

Functions

Returns a specification to start this module under a supervisor.

Declare the function calls the Provider committed for the current Turn, before the Executor starts running them (#462, layer 1).

Publish an ephemeral Event (live display only; never persisted).

Generate a sortable, filename-safe Session id.

Reconstruct History by folding the Log (the source of truth).

Snapshot of Session metadata (id, workspace, role, next seq, turn state).

Kill the currently running Turn's Task, if any.

Record a canonical Event: stamp seq, append to the Log, then publish. Returns the stamped Event, or a structured error if it was not canonical / failed to persist.

Hysteresis gate for context-pressure warnings (ADR 0020): returns {:ok, :warn} the first time a (latest checkpoint to_seq, tier) pair is seen and {:ok, :already_warned} afterwards — no warning spam on consecutive turns for the same pair. A new compaction checkpoint (different to_seq) or a different tier re-arms the gate. State is ephemeral process state by design: it is never logged, and re-warning after a Session restart is acceptable.

Start a Turn: run turn_fun.(ctx) in a supervised Task. Returns {:ok, ref} or {:error, :busy} if a Turn is already running. If the previous Turn left orphan tool calls in the Log, Pixir first records fallback tool_result events so Provider replay stays valid.

Whether a Turn is currently running.

{:via, Registry, …} name for a Session id.

Types

ctx()

@type ctx() :: %{
  session_id: String.t(),
  workspace: String.t(),
  role: role(),
  fork_root_session_id: String.t(),
  turn_generation: pos_integer()
}

role()

@type role() :: atom()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

declare_committed_calls(session_id, calls, generation \\ nil)

@spec declare_committed_calls(String.t(), [map()], pos_integer() | nil) ::
  :ok | {:error, map()}

Declare the function calls the Provider committed for the current Turn, before the Executor starts running them (#462, layer 1).

interrupt/1 kills the Turn Task outright, so no Turn-side cleanup can be trusted to run. A call the Provider already committed but Pixir has not yet persisted would then exist only provider-side, and the next Turn's request would omit its output — which the Responses API rejects with "No tool output found for function call <id>". Holding the committed set in the Session (which survives the kill) lets interrupt/1 drain it to the Log before the Turn closes.

Entries are %{call_id, name, args} maps. Recording the real tool_call clears its id, so only genuinely un-persisted calls are ever drained. The set is Turn-scoped: it is cleared when the Turn ends (and drained when the Turn is killed or crashes), so a declaration the Executor never reached cannot leak into a later interrupt. A declaration that lands with no Turn running is drained on the spot rather than accumulated: the streaming runner outlives the Turn it belonged to, so a late declaration is real evidence with no Turn left to own it.

generation is the caller's TURN IDENTITY, not merely a claim that some Turn is running (#462 round 3). Keying the drain on turn STATE alone (nil vs alive) is not enough: the surviving runner of a killed Turn can declare while a NEW Turn is already alive, and a state-only check accumulates that straggler into the successor, which then drops it at its own clean end — the same silent evidence loss one Turn over. Each Turn captures its generation at Turn start and stamps every declaration with it, so the Session can tell "my live Turn's call" from "a dead Turn's straggler" and drain the latter on the spot. Passing nil means "no identity claimed" and is always treated as a straggler; it is never accumulated into the live Turn.

This call carries its own generous timeout rather than the GenServer.call/2 default of 5 s (#462 round 3). The Session serializes Log appends and Log.fold/2 history folds, so a big Log can hold this call for longer than the default while nothing is actually wrong — and a mid-stream timeout is worse than a slow one: the caller lives inside the provider stream reducer, where an escaping exit is classified :network and IS provider-retryable, so a stalled fold would re-stream the request.

emit(session_id, event)

@spec emit(String.t(), Pixir.Event.t()) :: :ok | {:error, map()}

Publish an ephemeral Event (live display only; never persisted).

gen_id()

@spec gen_id() :: String.t()

Generate a sortable, filename-safe Session id.

history(session_id)

@spec history(String.t()) :: {:ok, Pixir.Log.history()} | {:error, map()}

Reconstruct History by folding the Log (the source of truth).

info(session_id)

@spec info(String.t()) :: map() | {:error, map()}

Snapshot of Session metadata (id, workspace, role, next seq, turn state).

interrupt(session_id)

@spec interrupt(String.t()) :: :ok | {:error, :no_turn} | {:error, map()}

Kill the currently running Turn's Task, if any.

record(session_id, event)

@spec record(String.t(), Pixir.Event.t()) :: {:ok, Pixir.Event.t()} | {:error, map()}

Record a canonical Event: stamp seq, append to the Log, then publish. Returns the stamped Event, or a structured error if it was not canonical / failed to persist.

register_pressure_warning(session_id, checkpoint_to_seq, tier)

@spec register_pressure_warning(String.t(), integer() | nil, String.t()) ::
  {:ok, :warn | :already_warned}

Hysteresis gate for context-pressure warnings (ADR 0020): returns {:ok, :warn} the first time a (latest checkpoint to_seq, tier) pair is seen and {:ok, :already_warned} afterwards — no warning spam on consecutive turns for the same pair. A new compaction checkpoint (different to_seq) or a different tier re-arms the gate. State is ephemeral process state by design: it is never logged, and re-warning after a Session restart is acceptable.

start_link(opts)

start_turn(session_id, turn_fun)

@spec start_turn(String.t(), (ctx() -> any())) ::
  {:ok, reference()} | {:error, :busy} | {:error, map()}

Start a Turn: run turn_fun.(ctx) in a supervised Task. Returns {:ok, ref} or {:error, :busy} if a Turn is already running. If the previous Turn left orphan tool calls in the Log, Pixir first records fallback tool_result events so Provider replay stays valid.

turn_running?(session_id)

@spec turn_running?(String.t()) :: boolean() | {:error, map()}

Whether a Turn is currently running.

via(session_id)

{:via, Registry, …} name for a Session id.