Class: PiAgent::Session

Inherits:
Object
  • Object
show all
Defined in:
lib/pi_agent/session.rb

Overview

High-level agent session. Wraps a Client and exposes prompt-flow ergonomics: submit a prompt and iterate the resulting event stream.

PiAgent.session do |session|
session.prompt("Write a haiku").each do |event|
  print event.delta if event.type == :message_update
end
end

A pi RPC process hosts exactly one session, so there is no create/select step — the Session is the running pi process.

prompt streams until the session-level run settles, including retries, automatic compaction retries, and queued continuations. A message queued with blockless follow_up runs after an active agent; pass a block to submit it as a streaming-aware prompt and drain it race-free. events is a prompt-less drain for when you have already subscribed before processing starts.

If the transport dies mid-stream (pi killed, sandbox torn down), event streams raise TransportClosedError promptly instead of waiting out the event timeout — provided the transport reports death (see Transport).

Constant Summary collapse

DEFAULT_EVENT_TIMEOUT =

Max time to wait for the next event before assuming the agent stalled.

300
DEFAULT_ACK_TIMEOUT =

Max time to wait for a command to be acknowledged.

30

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(client) ⇒ Session

Returns a new instance of Session.



34
35
36
37
38
 
# File 'lib/pi_agent/session.rb', line 34

def initialize(client)
  @client = client
  @stream_state_mutex = Mutex.new
  @event_stream_active = false
end

Instance Attribute Details

#client ⇒ Object (readonly)

Returns the value of attribute client.



32
33
34
 
# File 'lib/pi_agent/session.rb', line 32

def client
  @client
end

Instance Method Details

#abort ⇒ Object

Abort the current agent run. Fire-and-forget.



120
121
122
123
 
# File 'lib/pi_agent/session.rb', line 120

def abort
  @client.notify("abort")
  self
end

#available_models ⇒ Object

All configured models, as an array of Model hashes.



142
143
144
 
# File 'lib/pi_agent/session.rb', line 142

def available_models
  request_data("get_available_models").fetch("models", [])
end

#available_thinking_levels ⇒ Object

Thinking levels supported by the current model, as an array of strings (e.g. ["off", "low", "medium", "high"]). Returns ["off"] for a model without reasoning support.



156
157
158
 
# File 'lib/pi_agent/session.rb', line 156

def available_thinking_levels
  request_data("get_available_thinking_levels").fetch("levels", [])
end

#clone_session ⇒ Object

Duplicate the current active branch into a new session at the current position. Returns { "cancelled" => bool }. Maps to the clone RPC command (named clone_session to avoid shadowing Object#clone).



239
240
241
 
# File 'lib/pi_agent/session.rb', line 239

def clone_session
  request_data("clone")
end

#close ⇒ Object 



248
249
250
 
# File 'lib/pi_agent/session.rb', line 248

def close
  @client.close
end

#compact(custom_instructions: nil) ⇒ Object

Manually compact the conversation context to reduce token usage. Returns the result hash ({ "summary" =>, "firstKeptEntryId" =>, "tokensBefore" => }).



177
178
179
180
181
 
# File 'lib/pi_agent/session.rb', line 177

def compact(custom_instructions: nil)
  params = {}
  params[:customInstructions] = custom_instructions if custom_instructions
  request_data("compact", params)
end

#cycle_model ⇒ Object

Switch to the next configured model. Returns the new { "model" =>, "thinkingLevel" =>, "isScoped" => } hash, or {} when only one model is available.



137
138
139
 
# File 'lib/pi_agent/session.rb', line 137

def cycle_model
  request_data("cycle_model")
end

#entries(since: nil) ⇒ Object

Session entries in append order (excluding the session header). Unlike messages, this includes pre-compaction history and abandoned branches. Entry ids are stable, so pass since: (the last entry id you have seen) to get only entries strictly after it — a durable cursor across restarts. Returns { "entries" => [...], "leafId" => }.



215
216
217
218
219
 
# File 'lib/pi_agent/session.rb', line 215

def entries(since: nil)
  params = {}
  params[:since] = since if since
  request_data("get_entries", params)
end

#events(event_timeout: DEFAULT_EVENT_TIMEOUT, &block) ⇒ Object

Drain the event stream without submitting a new prompt. With a block, yields each Event until the run fully settles (agent_settled) and returns self; without a block, returns an Enumerator of Events.

The subscription is established lazily, when iteration begins — so any cycle triggered before you call events may have already emitted events that are then missed. To drain a follow_up cycle race-free, pass a block to follow_up instead. Use events only when you subscribe before the cycle starts (e.g. from a thread that begins iterating, then trigger the cycle). With nothing queued, it blocks for event_timeout (300s default).



67
68
69
70
71
72
73
74
 
# File 'lib/pi_agent/session.rb', line 67

def events(event_timeout: DEFAULT_EVENT_TIMEOUT, &block)
  stream = subscribed_stream(event_timeout: event_timeout)

  return stream unless block

  stream.each(&block)
  self
end

#follow_up(message, images: nil, event_timeout: DEFAULT_EVENT_TIMEOUT, &block) ⇒ Object

Queue a follow-up message, delivered only after the agent stops.

Without a block this maps to pi's raw follow_up command: it only queues the message and returns self, so call it while an agent run is active. With a block it uses pi's streaming-aware prompt command, which starts a run when idle or queues a follow-up when busy. The subscription is established before sending, then yields through agent_settled and returns self. Prefer this form for a sequential, consumable follow-up.



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/pi_agent/session.rb', line 92

def follow_up(message, images: nil, event_timeout: DEFAULT_EVENT_TIMEOUT, &block)
  params = message_params(message, images)
  unless block
    @client.request("follow_up", params).value!(timeout: DEFAULT_ACK_TIMEOUT)
    return self
  end

  if message.start_with?("/")
    raise SessionError, "Block-form follow_up does not support slash commands; use prompt instead"
  end

  params[:streamingBehavior] = "followUp"
  event_stream("prompt", params, event_timeout: event_timeout).each(&block)
  self
end

#fork(entry_id) ⇒ Object

Fork a new branch from a previous user message (an entryId from fork_messages). Returns { "text" => , "cancelled" => bool }; cancelled is true if an extension vetoed it.



231
232
233
 
# File 'lib/pi_agent/session.rb', line 231

def fork(entry_id)
  request_data("fork", entryId: entry_id)
end

#fork_messages ⇒ Object

List user messages available for forking. Returns an array of { "entryId" => ..., "text" => ... } hashes.



206
207
208
 
# File 'lib/pi_agent/session.rb', line 206

def fork_messages
  request_data("get_fork_messages").fetch("messages", [])
end

#get_state ⇒ Object 



160
161
162
 
# File 'lib/pi_agent/session.rb', line 160

def get_state
  @client.request("get_state").value!(timeout: DEFAULT_ACK_TIMEOUT)
end

#last_assistant_text ⇒ Object

Text of the last assistant message, or nil if there is none.



170
171
172
 
# File 'lib/pi_agent/session.rb', line 170

def last_assistant_text
  request_data("get_last_assistant_text")["text"]
end

#messages ⇒ Object

Full conversation history, as an array of AgentMessage hashes.



165
166
167
 
# File 'lib/pi_agent/session.rb', line 165

def messages
  request_data("get_messages").fetch("messages", [])
end

#new_session(parent_session: nil) ⇒ Object

Start a fresh session in the same pi process. Pass parent_session: (a session file path) to record provenance. Returns { "cancelled" => bool }; cancelled is true if an extension vetoed it.



186
187
188
189
190
 
# File 'lib/pi_agent/session.rb', line 186

def new_session(parent_session: nil)
  params = {}
  params[:parentSession] = parent_session if parent_session
  request_data("new_session", params)
end

#prompt(message, images: nil, event_timeout: DEFAULT_EVENT_TIMEOUT, &block) ⇒ Object

Submit a user prompt. With a block, yields each Event until the agent fully settles (agent_settled), then returns self. This includes any automatic retry, compaction retry, or queued continuation. Without a block, returns an Enumerator of Events.

images accepts PiAgent::Image objects, file path strings, or raw ImageContent hashes — in any mix.



47
48
49
50
51
52
53
54
 
# File 'lib/pi_agent/session.rb', line 47

def prompt(message, images: nil, event_timeout: DEFAULT_EVENT_TIMEOUT, &block)
  stream = event_stream("prompt", message_params(message, images), event_timeout: event_timeout)

  return stream unless block

  stream.each(&block)
  self
end

#run(message, images: nil, event_timeout: DEFAULT_EVENT_TIMEOUT) ⇒ Object

Single-shot helper mirroring pi's print mode: submit message, drain the whole event stream, and return the final assistant text (nil if the agent produced none). Yields each Event to an optional block while the stream drains.



112
113
114
115
116
117
 
# File 'lib/pi_agent/session.rb', line 112

def run(message, images: nil, event_timeout: DEFAULT_EVENT_TIMEOUT)
  prompt(message, images: images, event_timeout: event_timeout) do |event|
    yield event if block_given?
  end
  last_assistant_text
end

#session_stats ⇒ Object

Token usage, cost, and context-window stats for the current session. Returns the data hash, including "sessionId" and "sessionFile".



200
201
202
 
# File 'lib/pi_agent/session.rb', line 200

def session_stats
  request_data("get_session_stats")
end

#set_model(provider, model_id = nil) ⇒ Object

Switch to a specific model. Accepts either a single "provider/modelId" string or the two parts as separate arguments.



127
128
129
130
131
132
 
# File 'lib/pi_agent/session.rb', line 127

def set_model(provider, model_id = nil)
  provider, model_id = provider.split("/", 2) if model_id.nil?
  @client.request("set_model", provider: provider, modelId: model_id)
         .value!(timeout: DEFAULT_ACK_TIMEOUT)
  self
end

#set_session_name(name) ⇒ Object 



243
244
245
246
 
# File 'lib/pi_agent/session.rb', line 243

def set_session_name(name)
  @client.request("set_session_name", name: name).value!(timeout: DEFAULT_ACK_TIMEOUT)
  self
end

#set_thinking(level) ⇒ Object

Set the reasoning level: "off", "minimal", "low", "medium", "high", or "xhigh" (xhigh is OpenAI codex-max only).



148
149
150
151
 
# File 'lib/pi_agent/session.rb', line 148

def set_thinking(level)
  @client.request("set_thinking_level", level: level).value!(timeout: DEFAULT_ACK_TIMEOUT)
  self
end

#steer(message, images: nil) ⇒ Object

Queue a steering message while the agent is running. Delivered after the current assistant turn finishes its tool calls, before the next LLM call. Fire-and-forget; raises on rejection.



79
80
81
82
 
# File 'lib/pi_agent/session.rb', line 79

def steer(message, images: nil)
  @client.request("steer", message_params(message, images)).value!(timeout: DEFAULT_ACK_TIMEOUT)
  self
end

#switch_session(path) ⇒ Object

Load a different session file into this process. Returns { "cancelled" => bool }; cancelled is true if an extension vetoed it.



194
195
196
 
# File 'lib/pi_agent/session.rb', line 194

def switch_session(path)
  request_data("switch_session", sessionPath: path)
end

#tree ⇒ Object

The session as a tree of entries. Each node is { "entry" =>, "children" => [...], "label"? =>, "labelTimestamp"? => }. Returns { "tree" => [...], "leafId" => }.



224
225
226
 
# File 'lib/pi_agent/session.rb', line 224

def tree
  request_data("get_tree")
end