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

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.



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

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

#deny(call_id, note: nil) ⇒ Object 



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

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.



95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/mistri/session.rb', line 95

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_steers ⇒ Object

Steers not yet folded into the transcript, oldest first. The folding message entry carries the steer id, so consumption is derived from the log alone and reads the same from every process.



80
81
82
83
 
# File 'lib/mistri/session.rb', line 80

def pending_steers
  folded = entries.filter_map { |entry| entry["steer_id"] }
  entries.select { |entry| entry["type"] == "steer" && !folded.include?(entry["id"]) }
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.



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

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