Class: Mistri::Session

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

Overview

The durable record of a run: an append-only log of typed entries over a pluggable store. Messages replay into provider-ready history; other entry types (reminders, custom host data) ride alongside without the loop knowing their shapes.

A store implements two methods: append(id, entry_hash) and load(id) -> array of entry hashes in append order. Everything else lives here.

Constant Summary collapse

INTERRUPTED_RESULT =

What a run killed mid-tool answers in place of the result it never got.

"[interrupted: the run stopped before this result was persisted; " \
"the tool may have executed, so verify its effects before retrying]"
INBOX =

The inbox: entry types queued for the loop's next turn boundary, each mapped to the marker key its fold leaves on the consuming message entry.

{ "steer" => "steer_id", "subagent_report" => "report_id" }.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(store:, id: nil) ⇒ Session

Returns a new instance of Session.



18
19
20
21
 
# File 'lib/mistri/session.rb', line 18

def initialize(store:, id: nil)
  @store = store
  @id = id || SecureRandom.uuid
end

Instance Attribute Details

#id ⇒ Object (readonly)

rubocop:disable Metrics/ClassLength -- append-only ordering is the class contract



16
17
18
 
# File 'lib/mistri/session.rb', line 16

def id
  @id
end

#store ⇒ Object (readonly)

rubocop:disable Metrics/ClassLength -- append-only ordering is the class contract



16
17
18
 
# File 'lib/mistri/session.rb', line 16

def store
  @store
end

Instance Method Details

#append(type, data = {}) ⇒ Object

Entries are normalized through JSON at write time, so every store holds the same canonical shape (string keys, JSON values) and reads behave identically whether the entry round-tripped a database or stayed in memory.



32
33
34
35
36
 
# File 'lib/mistri/session.rb', line 32

def append(type, data = {})
  entry = { "type" => type, "at" => Time.now.utc.iso8601 }.merge(data)
  @store.append(@id, JSON.parse(JSON.generate(entry)))
  nil
end

#append_message(message) ⇒ Object 



23
24
25
26
 
# File 'lib/mistri/session.rb', line 23

def append_message(message)
  append("message", "message" => message.to_h)
  message
end

#approve(call_id, note: nil) ⇒ Object

Record a human's decision on a parked tool call. Decisions are session entries, so they can be written from any process, days later, with no agent constructed; the next resume settles them.



173
 
# File 'lib/mistri/session.rb', line 173

def approve(call_id, note: nil) = decide(call_id, approved: true, note: note)

#children ⇒ Object

Sub-agents this session has spawned, in spawn order: one Child window per link entry, each reading the child's own session. Derived from the log alone, like everything else here.



150
151
152
153
154
155
156
157
 
# File 'lib/mistri/session.rb', line 150

def children
  entries.filter_map do |entry|
    next unless entry["type"] == "subagent"

    Child.new(name: entry["name"], session_id: entry["session_id"], store: @store,
              parent_session_id: @id)
  end
end

#context_tokens ⇒ Object

Context accounting ignores provider usage reported before the latest compaction: those prompt counts describe the larger, pre-summary replay. Until a post-compaction assistant turn reports fresh usage, the compacted replay is estimated directly.



79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/mistri/session.rb', line 79

def context_tokens
  log = entries
  replay = replay_from(log)
  compacted_at = log.rindex { |entry| entry["type"] == "compaction" }
  usage_from = if compacted_at
                 replay.index { |(_, index)| index && index > compacted_at } || replay.length
               else
                 0
               end
  Compaction.context_tokens(replay.map(&:first), usage_from:)
end

#deliver_report(name:, session_id:, status:, text: nil) ⇒ Object

A sub-agent's report, queued for this session the way a steer is: it folds into the transcript at the next turn boundary as a labeled block the model can react to ("[Magpie finished] "), while the typed entry keeps name, status, and the raw text for hosts to render as a report card rather than a fake user message. Sequential redelivery of one child's report is dropped, and the return says which happened. Concurrent callers need their own serialization; with a lock adapter, the background runner and inactive cancellation use the child's lease for that. Reports normally arrive via SubAgent.run_dispatched; call this directly only from a custom dispatch path.



118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/mistri/session.rb', line 118

def deliver_report(name:, session_id:, status:, text: nil) # rubocop:disable Naming/PredicateMethod
  already = entries.any? do |entry|
    entry["type"] == "subagent_report" && entry["session_id"] == session_id
  end
  return false if already

  append("subagent_report", "id" => SecureRandom.uuid, "name" => name,
                            "session_id" => session_id, "status" => status, "report" => text,
                            "message" => Message.user(report_label(name, status, text)).to_h)
  true
end

#deny(call_id, note: nil) ⇒ Object 



175
 
# File 'lib/mistri/session.rb', line 175

def deny(call_id, note: nil) = decide(call_id, approved: false, note: note)

#entries ⇒ Object 



38
 
# File 'lib/mistri/session.rb', line 38

def entries = @store.load(@id)

#last_compaction ⇒ Object 



91
92
93
 
# File 'lib/mistri/session.rb', line 91

def last_compaction
  entries.reverse_each.find { |entry| entry["type"] == "compaction" }
end

#messages ⇒ Object

The conversation as the model replays it.



56
 
# File 'lib/mistri/session.rb', line 56

def messages = replay.map(&:first)

#open_approvals ⇒ Object

Parked tool calls not yet settled by a tool result, each with its decision when one has been recorded. Derived from the entry log alone, so it survives crashes and reads the same from every process.



180
 
# File 'lib/mistri/session.rb', line 180

def open_approvals = tool_control_state.fetch(:approvals)

#pending_inbox ⇒ Object

Everything queued for the loop's next turn boundary (steers and sub-agent reports), oldest first, in arrival order. The folding message entry carries the source entry's id under its marker key, so consumption is derived from the log alone and reads the same from every process. A host that wakes an idle session when a steer arrives should watch this instead: a report deserves the same pickup.



136
137
138
139
140
 
# File 'lib/mistri/session.rb', line 136

def pending_inbox
  log = entries
  folded = log.flat_map { |entry| entry.values_at(*INBOX.values) }.compact.to_set
  log.select { |entry| INBOX.key?(entry["type"]) && !folded.include?(entry["id"]) }
end

#pending_steers ⇒ Object

The steer-only slice of the inbox, oldest first.



143
144
145
 
# File 'lib/mistri/session.rb', line 143

def pending_steers
  pending_inbox.select { |entry| entry["type"] == "steer" }
end

#replay ⇒ Object

Replay messages paired with the entry index each came from, starting at the latest compaction boundary. The synthetic summary message carries a nil index. Compaction places its cuts by these indexes; the full entry log stays in the store for transcript views. One store read builds the whole context, healed: a crash that left tool calls unanswered would brick every later turn with a provider rejection, so unsettled calls get a synthesized interrupted result. Calls parked for human approval stay open; resume owns those.



73
 
# File 'lib/mistri/session.rb', line 73

def replay = replay_from(entries)

#steer(text) ⇒ Object

Queue a message for a running exchange from any process. The loop folds pending steers into the transcript at the next turn boundary, so the model sees them mid-run; one that arrives as the model finishes cleanly extends the run another turn so it is answered, not left dangling.



104
105
106
 
# File 'lib/mistri/session.rb', line 104

def steer(text)
  append("steer", "id" => SecureRandom.uuid, "message" => Message.user(text).to_h)
end

#tool_call_ids ⇒ Object 



53
 
# File 'lib/mistri/session.rb', line 53

def tool_call_ids = tool_control_state.fetch(:tool_call_ids)

#tool_control_state ⇒ Object

Every call ID is a session-wide correlation key. Read and validate the append-only history once, returning an owned set the loop can extend as it accepts later turns. Approval control entries are audited in the same ordered pass: they may only advance a prior assistant call.



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

def tool_control_state
  audited = audit_history(entries)
  approvals = audited.fetch(:approvals).map do |approval|
    decision = approval[:decision] && own_json(approval[:decision])
    { call: rebuild_call(approval[:call]), decision: }
  end
  { tool_call_ids: audited.fetch(:tool_call_ids), approvals: }
end

#transcript(include_children: false) ⇒ Object

The session as a reader renders it: entries in order with inline image bytes stripped, and, with include_children, every sub-agent's own transcript spliced in after its link entry. Spliced entries carry an "origin" key shaped exactly like the live stream's event origins ("Magpie#ab12cd34", nesting joined with ">"), so a UI that rebuilds from this sees what it saw live, lanes included, running children's progress-so-far included. The raw log stays available as #entries.



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

def transcript(include_children: false)
  splice(entries, include_children: include_children, prefix: nil, seen: Set[@id])
end