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.

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.



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

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.



28
29
30
 
# File 'lib/pi_agent/session.rb', line 28

def client
  @client
end

Instance Method Details

#abort ⇒ Object

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



116
117
118
119
 
# File 'lib/pi_agent/session.rb', line 116

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

#available_models ⇒ Object

All configured models, as an array of Model hashes.



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

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.



152
153
154
 
# File 'lib/pi_agent/session.rb', line 152

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).



235
236
237
 
# File 'lib/pi_agent/session.rb', line 235

def clone_session
  request_data("clone")
end

#close ⇒ Object 



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

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" => }).



173
174
175
176
177
 
# File 'lib/pi_agent/session.rb', line 173

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.



133
134
135
 
# File 'lib/pi_agent/session.rb', line 133

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" => }.



211
212
213
214
215
 
# File 'lib/pi_agent/session.rb', line 211

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).



63
64
65
66
67
68
69
70
 
# File 'lib/pi_agent/session.rb', line 63

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.



88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/pi_agent/session.rb', line 88

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.



227
228
229
 
# File 'lib/pi_agent/session.rb', line 227

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.



202
203
204
 
# File 'lib/pi_agent/session.rb', line 202

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

#get_state ⇒ Object 



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

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.



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

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

#messages ⇒ Object

Full conversation history, as an array of AgentMessage hashes.



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

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.



182
183
184
185
186
 
# File 'lib/pi_agent/session.rb', line 182

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.



43
44
45
46
47
48
49
50
 
# File 'lib/pi_agent/session.rb', line 43

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.



108
109
110
111
112
113
 
# File 'lib/pi_agent/session.rb', line 108

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".



196
197
198
 
# File 'lib/pi_agent/session.rb', line 196

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.



123
124
125
126
127
128
 
# File 'lib/pi_agent/session.rb', line 123

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 



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

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).



144
145
146
147
 
# File 'lib/pi_agent/session.rb', line 144

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.



75
76
77
78
 
# File 'lib/pi_agent/session.rb', line 75

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.



190
191
192
 
# File 'lib/pi_agent/session.rb', line 190

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" => }.



220
221
222
 
# File 'lib/pi_agent/session.rb', line 220

def tree
  request_data("get_tree")
end