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 tool returned]"
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)

Returns the value of attribute id.



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

def id
  @id
end

#store ⇒ Object (readonly)

Returns the value of attribute store.



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.



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

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.



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

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

    Child.new(name: entry["name"], session_id: entry["session_id"], store: @store)
  end
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. One report per child, ever: a duplicate delivery (a redelivered queue job, a lease race) is dropped, and the return says which happened. Reports normally arrive via SubAgent.run_dispatched; call this directly only from a custom dispatch path.



91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/mistri/session.rb', line 91

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 



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

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 



65
66
67
 
# File 'lib/mistri/session.rb', line 65

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

#messages ⇒ Object

The conversation as the model replays it.



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

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.



152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/mistri/session.rb', line 152

def open_approvals
  answered = []
  decisions = {}
  requests = []
  entries.each do |entry|
    case entry["type"]
    when "message"
      call_id = entry.dig("message", "tool_call_id")
      answered << call_id if call_id
    when "approval_request" then requests << entry["call"]
    when "approval_decision" then decisions[entry["call_id"]] = entry
    end
  end
  requests.reject { |call| answered.include?(call["id"]) }
          .map { |call| { call: rebuild_call(call), decision: decisions[call["id"]] } }
end

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



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

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.



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

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.



54
55
56
57
58
59
60
61
62
63
 
# File 'lib/mistri/session.rb', line 54

def replay
  log = entries
  compaction = log.reverse_each.find { |entry| entry["type"] == "compaction" }
  from = compaction ? compaction["kept_from"] : 0
  pairs = log.each_with_index.filter_map do |entry, index|
    [Message.from_h(entry["message"]), index] if index >= from && entry["type"] == "message"
  end
  pairs = heal(pairs, parked_call_ids(log))
  compaction ? [[summary_message(compaction["summary"]), nil], *pairs] : pairs
end

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



78
79
80
 
# File 'lib/mistri/session.rb', line 78

def steer(text)
  append("steer", "id" => SecureRandom.uuid, "message" => Message.user(text).to_h)
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.



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

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