Class: Insika::SessionStore

Inherits:
Object
  • Object
show all
Includes:
Coercion
Defined in:
lib/insika/session_store.rb

Overview

Domain store for sessions. Persists transcript + vars over an injected Insika::Store, with a fixed schema session:<id> in the "sessions" scope.

The persisted transcript is the SOURCE OF TRUTH for reconstruction; live events are just delivery state. The message shape ({"role"=>, "content"=>}, role ∈ user|assistant|system|tool) is the same one Runner#seed_history already consumes — the Executor converts nothing.

Normalizes symbol→string on WRITE (the backend only guarantees round-trip of JSON types); READ returns the data as it comes from the backend (string keys), never symmetrizing back to symbols.

Defined Under Namespace

Classes: Session

Constant Summary collapse

SCOPE = 
"sessions"
KEY_PREFIX = 
"session:"

Constants included from Coercion

Coercion::TRUTHY

Instance Method Summary collapse

Methods included from Coercion

blank?, deep_stringify, presence, present?, truthy?, utf8

Constructor Details

#initialize(store:) ⇒ SessionStore

store: any Insika::Store (Memory, SQLite, ...) — injected by the composition root (config/wiring.rb). SessionStore does not know the concrete backend.



38
39
40
 
# File 'lib/insika/session_store.rb', line 38

def initialize(store:)
  @store = store
end

Instance Method Details

#append_evidence(id, ids:, ungrounded:) ⇒ Object

appends this turn's evidence (ids + ungrounded delta) to the session record. RMW like append_messages — the SessionActor serializes same-session turns; the copy is in the method comment.



104
105
106
107
108
109
110
111
112
113
 
# File 'lib/insika/session_store.rb', line 104

def append_evidence(id, ids:, ungrounded:)
  record = fetch!(id)
  ev = record["evidence"] ||= { "ids" => [], "ungrounded" => 0 }
  fresh = (ev["ids"] + Array(ids).map(&:to_s).reject(&:empty?)).uniq.last(EvidenceLedger::MAX_IDS)
  ev["ids"] = fresh
  ev["ungrounded"] = ev["ungrounded"].to_i + ungrounded.to_i
  record["updated_at"] = timestamp
  @store.set(SCOPE, key_for(id), record)
  to_session(record)
end

#append_messages(id, messages) ⇒ Object

-> Session (transcript += messages). Read-modify-write on the task's own fiber, without a lock. Each message gets an "at" (ISO8601 UTC) if not provided. NotFoundError if the session does not exist.

CONCURRENCY LIMITATION (R2c): the RMW (read record -> += -> set) is atomic ONLY because the SessionActor serializes turns of the same session (one owner at a time). That serialization exists solely in SUPERVISED mode (the actor loop lives on the supervisor). Two concurrent send_message on the same session_id OUTSIDE that path (e.g. calling append_messages directly, or a non-supervised deployment) would interleave read/set and LOSE messages — there is no compare-and-swap here. Route same-session writes through the SessionActor; see session_actor.rb.



81
82
83
84
85
86
87
88
89
 
# File 'lib/insika/session_store.rb', line 81

def append_messages(id, messages)
  record = fetch!(id)
  incoming = (messages.is_a?(Hash) ? [messages] : Array(messages))
             .map { |msg| stamp(deep_stringify(msg)) }
  record["messages"] += incoming
  record["updated_at"] = timestamp
  @store.set(SCOPE, key_for(id), record)
  to_session(record)
end

#create(id: SecureRandom.uuid, vars: {}) ⇒ Object

-> Session; ArgumentError if id already exists (a duplicate session is a domain violation — it never overwrites silently).

Raises:

  • (ArgumentError) 


44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/insika/session_store.rb', line 44

def create(id: SecureRandom.uuid, vars: {})
  key = key_for(id)
  raise ArgumentError, "session already exists: #{id}" unless @store.get(SCOPE, key).nil?

  now = timestamp
  record = {
    "id" => id.to_s,
    "messages" => [],
    "vars" => deep_stringify(vars),
    "memory_refs" => [],
    "briefing" => { "fields" => {}, "next_step" => nil },
    "evidence" => { "ids" => [], "ungrounded" => 0 },
    "created_at" => now,
    "updated_at" => now
  }
  @store.set(SCOPE, key, record)
  to_session(record)
end

#delete(id) ⇒ Object

-> bool (delegates to the backend: false for a nonexistent id)



154
155
156
 
# File 'lib/insika/session_store.rb', line 154

def delete(id)
  @store.delete(SCOPE, key_for(id))
end

#each_id ⇒ Object

-> enumerates ids without the "session:" prefix. Without a block, returns an Enumerator.



160
161
162
163
164
165
166
 
# File 'lib/insika/session_store.rb', line 160

def each_id
  return enum_for(:each_id) unless block_given?

  @store.list(SCOPE, KEY_PREFIX).each do |key|
    yield key.delete_prefix(KEY_PREFIX)
  end
end

#find(id) ⇒ Object

-> Session | nil



64
65
66
67
 
# File 'lib/insika/session_store.rb', line 64

def find(id)
  record = @store.get(SCOPE, key_for(id))
  record && to_session(record)
end

#set_next_step(id, text:) ⇒ Object

-> Session. Upsert the agreed next step; a blank text clears to nil NotFoundError if absent.



144
145
146
147
148
149
150
151
 
# File 'lib/insika/session_store.rb', line 144

def set_next_step(id, text:)
  record = fetch!(id)
  briefing = record["briefing"] ||= { "fields" => {}, "next_step" => nil }
  briefing["next_step"] = Coercion.presence(Coercion.utf8(text.to_s))
  record["updated_at"] = timestamp
  @store.set(SCOPE, key_for(id), record)
  to_session(record)
end

#update_briefing(id, field:, value:) ⇒ Object

-> Session. Upsert ONE briefing field. The pack owns the schema; the engine validates nothing about field NAMES here (the tools do, at the write edge). value is a String (anything else -> to_s); a BLANK value (after strip) REMOVES the key — absence means "not yet asked" ( D4). NotFoundError if the session does not exist.

CONCURRENCY NOTE: an unlocked RMW (read -> mutate -> set), like append_messages — but the SessionActor argument does NOT apply here. A briefing writer is a system tool, never enveloped, and with tool_concurrency > 1 the gem runs each call in its OWN fiber — outside the actor's per-session turn serialization. The RMW is still safe, for a different reason: nothing in the read/mutate/set path suspends. Store get/set are synchronous (the SQLite write semaphore is a non-yielding fast path when free), and a fiber only switches at a scheduler suspension point — so no other writer can interleave mid-RMW (measured: N concurrent writers lose nothing). It holds ONLY while that path never suspends; an async store (a real yield in get/set) would need a lock or CAS.



132
133
134
135
136
137
138
139
140
 
# File 'lib/insika/session_store.rb', line 132

def update_briefing(id, field:, value:)
  record = fetch!(id)
  briefing = record["briefing"] ||= { "fields" => {}, "next_step" => nil }
  value = Coercion.presence(Coercion.utf8(value.to_s))
  value ? briefing["fields"][field.to_s] = value : briefing["fields"].delete(field.to_s)
  record["updated_at"] = timestamp
  @store.set(SCOPE, key_for(id), record)
  to_session(record)
end

#update_vars(id, vars) ⇒ Object

-> Session (SHALLOW merge: an existing nested key is replaced wholesale, not merged). NotFoundError if absent.



93
94
95
96
97
98
99
 
# File 'lib/insika/session_store.rb', line 93

def update_vars(id, vars)
  record = fetch!(id)
  record["vars"] = record["vars"].merge(deep_stringify(vars))
  record["updated_at"] = timestamp
  @store.set(SCOPE, key_for(id), record)
  to_session(record)
end