Class: Terret::Sessions

Inherits:
Hames::Service
  • Object
show all
Defined in:
lib/terret/sessions.rb

Overview

ctx.sessions — the append-only session log. The single source of the context the model sees: derive_messages projects model history from it, and a digest invariant asserts that what goes out to an adapter is exactly what replaying the log yields ("model-visible means logged").

Defined Under Namespace

Classes: Session

Constant Summary collapse

STRUCTURAL_KEYS =

Payload keys whose values the harness minted to point at something else: tool call ids and their approval foreign keys, the part tag decode_part dispatches on, lineage, verdicts, the live allow list. A pattern written for a credential — long hex, a UUID shape — matches these too, and rewriting one protects nothing (none of it is model-carried) while it can wreck the log for good: two tool calls that collapse to one id are a request providers reject, a mangled tag makes the session undecodable, and a rewritten pattern list silently changes what an agent may run. Add a key here only when the harness itself generates its value. A tool NAME is deliberately absent: the model chooses it, so it is content — and redacting one fails safe, because a name that stops resolving comes back as a not-found error rather than collapsing two calls onto one identifier.

%i[id call_id type verdict status agent parent_id
from boundary upto_seq n patterns].freeze
STRUCTURAL_CONTAINERS =

Keys whose value is a structural CONTAINER rather than a leaf: the exemption reaches through them to the identifiers inside, because an assistant message's encoded parts each carry their own tag and id.

%i[parts].freeze

Instance Method Summary collapse

Instance Method Details

#append(session_id, type, payload = {}) ⇒ Object

Raises:

  • (Hames::ContractError) 


115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/terret/sessions.rb', line 115

def append(session_id, type, payload = {})
  decl = Hames.event(type)
  raise Hames::ContractError, "#{type} is not a durable event" unless decl.durable

  s = fetch(session_id)
  normalized = normalize_payload(payload)
  ev = lock_for(session_id).synchronize do
    e = SessionEvent.new(
      id: SecureRandom.hex(8), session_id:, seq: s.events.length,
      at: Time.now.utc, type: type.to_s, payload: normalized
    )
    # durable first: if the store raises, nothing believes the event happened
    @store.append(e)
    s.events << e
    e
  end
  fan_out(ev)
  ev
end

#assert_log_invariant!(session_id, outbound_messages) ⇒ Object

The enforcement point for "model-visible means logged": the loop calls this with the message list it is about to send; a mismatch against the log projection raises in dev/test.

Raises:



184
185
186
187
188
189
190
 
# File 'lib/terret/sessions.rb', line 184

def assert_log_invariant!(session_id, outbound_messages)
  derived = derive_messages(session_id)
  return if digest(derived) == digest(outbound_messages)

  raise LogInvariantViolation,
        "outbound request diverges from session log projection for #{session_id}"
end

#create(id: SecureRandom.hex(6), parent_id: nil) ⇒ Object 



106
107
108
109
110
111
 
# File 'lib/terret/sessions.rb', line 106

def create(id: SecureRandom.hex(6), parent_id: nil)
  s = Session.new(id:, events: [], parent_id:)
  @cache[id] = s
  append(id, "session/created", { parent_id: })
  s
end

#derive_messages(session_id, upto: nil) ⇒ Object

Project provider-neutral model history from the durable log.



136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/terret/sessions.rb', line 136

def derive_messages(session_id, upto: nil)
  events = fetch(session_id).events
  events = events.take(upto) if upto
  apply_compaction(events).filter_map do |ev|
    case ev.type
    when "user/message", "context/injected"
      LLM::Message.new(role: :user, parts: [LLM::Text.new(text: ev.payload[:text])])
    when "session/compacted"
      LLM::Message.new(role: :user, parts: [LLM::Text.new(text: ev.payload[:summary])])
    when "assistant/message"
      LLM::Message.new(role: :assistant,
                       parts: ev.payload[:parts].map { |p| LLM.decode_part(p) })
    when "tool/result"
      LLM::Message.new(role: :tool, parts: [
        LLM::ToolResult.new(id: ev.payload[:id], content: ev.payload[:content],
                            error: ev.payload[:error])
      ])
    end
  end
end

#fetch(id) ⇒ Object 



113
 
# File 'lib/terret/sessions.rb', line 113

def fetch(id) = @cache.fetch(id)

#fork(source_id, boundary: nil, child_id: SecureRandom.hex(6)) ⇒ Object

Forks the in-memory working set; resume a store-only session first.



193
194
195
196
197
198
199
200
201
202
203
204
205
 
# File 'lib/terret/sessions.rb', line 193

def fork(source_id, boundary: nil, child_id: SecureRandom.hex(6))
  src = fetch(source_id)
  child = Session.new(id: child_id, events: [], parent_id: source_id)
  @cache[child_id] = child
  events = boundary ? src.events.take(boundary) : src.events.dup
  events.each do |ev|
    copy = ev.with(session_id: child_id)
    @store.append(copy)
    child.events << copy
  end
  append(child_id, "session/forked", { from: source_id, boundary: boundary })
  child
end

#read(session_id, from_seq: 0) ⇒ Object 



207
 
# File 'lib/terret/sessions.rb', line 207

def read(session_id, from_seq: 0) = @store.read(session_id, from_seq: from_seq)

#register_scrubber(callable, ctx: @ctx) ⇒ Object

ctx: decides the registration's LIFETIME, not its reach: the scrubber list is the service's, so a scrubber always sees every append, but a caller passing its forked agent context ties ownership to that fork — the same bleed Registry#register closed, where a registration made by an agent outlived the agent that made it.



93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/terret/sessions.rb', line 93

def register_scrubber(callable, ctx: @ctx)
  ctx.effect do
    @scrubbers << callable
    # Identity, not ==: removing "the entry equal to this callable" would
    # take a twin down with it if the same object were registered twice,
    # and a scrubber that defines its own == could unregister a stranger.
    lambda do
      i = @scrubbers.rindex { |s| s.equal?(callable) }
      @scrubbers.delete_at(i) if i
    end
  end
end

#resume(session_id) ⇒ Object

Rebuild a session's working set from the durable store. Idempotent: a session already in memory is returned as-is (write-through keeps the store equal). New appends continue after the last recorded seq.

Raises:

  • (KeyError) 


214
215
216
217
218
219
220
221
222
 
# File 'lib/terret/sessions.rb', line 214

def resume(session_id)
  return @cache[session_id] if @cache.key?(session_id)

  events = @store.read(session_id)
  raise KeyError, "unknown session #{session_id}" if events.empty?

  @cache[session_id] = Session.new(id: session_id, events: events,
                                   parent_id: parent_id_from(events))
end

#scrubbing? ⇒ Boolean

Whether anything is registered. Callers that must reshape what they append to give a scrubber a fair look at it — the loop's chunk carry — ask this so they can stay exactly as they were when nobody is scrubbing.

Returns:

  • (Boolean) 


86
 
# File 'lib/terret/sessions.rb', line 86

def scrubbing? = !@scrubbers.empty?

#session_ids ⇒ Object 



209
 
# File 'lib/terret/sessions.rb', line 209

def session_ids = @store.session_ids

#start(ctx) ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/terret/sessions.rb', line 50

def start(ctx)
  @ctx = ctx
  @store = ctx[:session_store]
  @cache = {}
  @locks = {}
  @locks_mutex = Mutex.new
  @emit_mutex = Mutex.new
  @emit_queue = []
  @emitting = false
  @scrubbers = []
end

#stored_form(value) ⇒ Object

Rewrite every String of every durable payload on its way into the log (§13's log boundary; docs/exec.md §6). Registration is an effect, so the returned disposer unregisters and unloading the owning plugin reaps it.

This is the boundary the scrubbing has to happen at rather than anywhere downstream: the stored event and every projection derived from it — derive_messages, and so both sides of the digest assert_log_invariant! compares — read the same already-scrubbed bytes, so "model-visible means logged" holds by construction. A read-time filter over the projection would leave the secret in the log itself and split the digest in two.

Scrubbers fold in registration order: each is handed the previous one's output, so a later one can rewrite what an earlier one produced. What a live in-memory value would look like once stored, for a caller that has to compare one against a payload already in the log — the approvals gate matches a recorded verdict on the call's args, and scrubbing rewrites one side of that comparison. CONTENT scope, because such a value always sits nested under a content key (args), so its own keys get no structural exemption.



81
 
# File 'lib/terret/sessions.rb', line 81

def stored_form(value) = normalize_payload(value, structural: false)

#title(session_id) ⇒ Object

The latest session/titled's title, or nil. Metadata, not model history.



175
176
177
178
179
 
# File 'lib/terret/sessions.rb', line 175

def title(session_id)
  fetch(session_id).events.reverse_each
                   .find { |e| e.type == "session/titled" }
                   &.payload&.[](:title)
end

#usage(session_id) ⇒ Object

Lifetime spend, projected from the log: sums every step/end's usage. A step whose provider sent no usage still counts as a step; its costs count as zero rather than poisoning the sum.



160
161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'lib/terret/sessions.rb', line 160

def usage(session_id)
  out = { prompt_tokens: 0, completion_tokens: 0, cost: 0.0, steps: 0 }
  fetch(session_id).events.each do |ev|
    next unless ev.type == "step/end"

    out[:steps] += 1
    u = ev.payload[:usage] or next
    out[:prompt_tokens]     += u[:prompt_tokens]     || 0
    out[:completion_tokens] += u[:completion_tokens] || 0
    out[:cost]              += u[:cost]              || 0.0
  end
  out
end