Class: Insika::ProposalStore

Inherits:
Object
  • Object
show all
Defined in:
lib/insika/proposal_store.rb

Overview

the proposals and the two persisted mechanisms the gates on — the latched dedup ledger (D3: the rows themselves ARE the ledger — a dismissed/rejected tuple is never proposed again, and an unanswered proposal is not piled on) and the per-session distilled marker (D2: written only after a pass completes, so a crash mid-pass leaves the marker unwritten and the next pass re-scans). A dumb domain store — it holds no policy (which tuple is a fact is the distiller's job), no memory facts and no model. The scope string (the memory cell) is built by the callers from the MemoryStore::parse_cell shape; the store keys by (tenant, customer) explicitly.

Statuses: pending -> approved | rejected | dismissed | stale. stale is the CAS-lost re-present (E3): the proposal carries the fact's CURRENT value (current_value) next to the proposed one, never a silent overwrite.

Defined Under Namespace

Classes: Proposal

Constant Summary collapse

SCOPE = 
"proposals"
STATUSES = 
%w[pending approved rejected dismissed stale].freeze
TERMINAL = 
%w[approved rejected dismissed].freeze
PROPOSAL_PREFIX = 
"p:"
MARKER_PREFIX = 
"s:"

Instance Method Summary collapse

Constructor Details

#initialize(store:) ⇒ ProposalStore

Returns a new instance of ProposalStore.



34
35
36
 
# File 'lib/insika/proposal_store.rb', line 34

def initialize(store:)
  @store = store
end

Instance Method Details

#approve(id:, operator: nil, note: nil, now: Time.now.utc) ⇒ Object

---- transitions, each read-check-write on @store.transaction ---- pending -> terminal. ArgumentError for a wrong source state (the task_store.rb state-machine idiom).



118
119
120
 
# File 'lib/insika/proposal_store.rb', line 118

def approve(id:, operator: nil, note: nil, now: Time.now.utc)
  transition(id, "approved", operator: operator, note: note, now: now)
end

#create(tenant:, customer:, session_ref:, key:, value:, confidence: nil, evidence: [], expected_revision: nil, expected_existed: false, id: SecureRandom.uuid, now: Time.now.utc) ⇒ Object

-> Proposal (status :pending). The caller (RunDistillation) already ran the dedup checks; the store writes. evidence = message indexes; the revision baseline (D5) travels with the record.

The tenant is stored VERBATIM — nil in a single-tenant deployment, so the scope is the bare customer cell and the approval reads/writes the SAME cell the Memory provider injects (memory_store.rb's blank-tenant + customer -> "memory:" rule). Coercing a blank tenant to a sentinel here would orphan every approved fact in a phantom "memory:platform:" cell.



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/insika/proposal_store.rb', line 48

def create(tenant:, customer:, session_ref:, key:, value:, confidence: nil,
           evidence: [], expected_revision: nil, expected_existed: false,
           id: SecureRandom.uuid, now: Time.now.utc)
  tenant = tenant_key(tenant)
  stamp = now.iso8601(6)
  record = { "id" => id.to_s, "status" => "pending",
             "tenant" => tenant, "customer" => customer.to_s,
             "scope" => [tenant, customer.to_s].compact.join(":"),
             "session_ref" => session_ref.to_s, "key" => key.to_s,
             "value" => value.to_s, "confidence" => confidence,
             "evidence" => Array(evidence).map(&:to_i),
             "expected_revision" => expected_revision,
             "expected_existed" => !!expected_existed,
             "current_value" => nil, "operator" => nil, "note" => nil,
             "created_at" => stamp, "updated_at" => stamp }
  @store.set(SCOPE, PROPOSAL_PREFIX + id.to_s, record)
  to_proposal(record)
end

#decided?(tenant:, customer:, key:, value:) ⇒ Boolean

---- the latched dedup (D3) ---- true when a dismissed/rejected row exists for the exact tuple — the latch. Persisted rows ARE the ledger. A different value for the same name is a different tuple.

Returns:

  • (Boolean) 


99
100
101
102
103
104
105
 
# File 'lib/insika/proposal_store.rb', line 99

def decided?(tenant:, customer:, key:, value:)
  scan.any? do |p|
    tenant_key(p.tenant) == tenant_key(tenant) && p.customer == customer.to_s &&
      p.key == key.to_s && p.value == value.to_s &&
      %w[dismissed rejected].include?(p.status)
  end
end

#delete_older_than(time) ⇒ Object

Age-based prune (the retention sweep). TERMINAL rows age by their updated_at; a PENDING row is a zombie past the cutoff (its transcript is dead). Session MARKERS die WITH their proposals — a marker past the cutoff is evidence about a dead transcript (the session aged out under the same retention window), and keeping it would lock an unreviewed proposal out of re-distillation forever. -> count removed.



195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
 
# File 'lib/insika/proposal_store.rb', line 195

def delete_older_than(time)
  cutoff = time.utc.iso8601
  removed = 0
  @store.transaction do
    proposal_keys.each do |k|
      record = @store.get(SCOPE, k)
      next unless record

      terminal = TERMINAL.include?(record["status"])
      stamp = terminal ? record["updated_at"] : record["created_at"]
      next unless stamp && stamp.to_s < cutoff

      @store.delete(SCOPE, k)
      removed += 1
    end
    marker_keys.each do |k|
      marker = @store.get(SCOPE, k)
      next unless marker && marker["distilled_at"].to_s < cutoff

      @store.delete(SCOPE, k)
      removed += 1
    end
  end
  removed
end

#dismiss(id:, operator: nil, note: nil, now: Time.now.utc) ⇒ Object 



126
127
128
 
# File 'lib/insika/proposal_store.rb', line 126

def dismiss(id:, operator: nil, note: nil, now: Time.now.utc)
  transition(id, "dismissed", operator: operator, note: note, now: now)
end

#distilled?(session_ref) ⇒ Boolean

Returns:

  • (Boolean) 


147
148
149
 
# File 'lib/insika/proposal_store.rb', line 147

def distilled?(session_ref)
  !@store.get(SCOPE, MARKER_PREFIX + session_ref.to_s).nil?
end

#distilled_sessions(agent_id = nil) ⇒ Object 



151
152
153
154
 
# File 'lib/insika/proposal_store.rb', line 151

def distilled_sessions(agent_id = nil)
  keys = agent_id ? marker_keys.select { |k| marker_agent(k) == agent_id.to_s } : marker_keys
  keys.map { |k| k.delete_prefix(MARKER_PREFIX) }
end

#find(id) ⇒ Object 



67
68
69
70
 
# File 'lib/insika/proposal_store.rb', line 67

def find(id)
  record = @store.get(SCOPE, PROPOSAL_PREFIX + id.to_s)
  record && to_proposal(record)
end

#mark_distilled(session_ref, agent:, proposals:, dropped:, deduped: 0, cost: nil, now: Time.now.utc) ⇒ Object

---- the per-session marker (D2) ---- Written ONLY after a pass completes (RunDistillation). -> the marker hash.



138
139
140
141
142
143
144
145
 
# File 'lib/insika/proposal_store.rb', line 138

def mark_distilled(session_ref, agent:, proposals:, dropped:, deduped: 0, cost: nil, now: Time.now.utc)
  marker = { "session_ref" => session_ref.to_s, "agent" => agent.to_s,
             "distilled_at" => now.iso8601, "proposals" => proposals.to_i,
             "dropped" => dropped, "deduped" => deduped.to_i,
             "cost" => cost }
  @store.set(SCOPE, MARKER_PREFIX + session_ref.to_s, marker)
  marker
end

#mark_stale(id:, current_value:, operator: nil, now: Time.now.utc) ⇒ Object

pending -> stale, CAS lost; current_value = the fact as it stands (the re-present's second value, E3).



132
133
134
 
# File 'lib/insika/proposal_store.rb', line 132

def mark_stale(id:, current_value:, operator: nil, now: Time.now.utc)
  transition(id, "stale", operator: operator, current_value: current_value, now: now)
end

#open_pending?(tenant:, customer:, key:) ⇒ Boolean

true when a pending row exists for (scope, key) — no piling.

Returns:

  • (Boolean) 


108
109
110
111
112
113
 
# File 'lib/insika/proposal_store.rb', line 108

def open_pending?(tenant:, customer:, key:)
  scan.any? do |p|
    tenant_key(p.tenant) == tenant_key(tenant) && p.customer == customer.to_s &&
      p.key == key.to_s && p.status == "pending"
  end
end

#pending(limit: 100) ⇒ Object

The wiki's lists. pending = pending, oldest first (the operator works the oldest proposal first — evidence ages).



74
75
76
77
78
 
# File 'lib/insika/proposal_store.rb', line 74

def pending(limit: 100)
  scan.select { |p| p.status == "pending" }
      .sort_by { |p| [p.created_at.to_s, p.id] }
      .first(limit)
end

#purge(tenant:) ⇒ Object

A tenant's proposals. -> count removed.



175
176
177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/insika/proposal_store.rb', line 175

def purge(tenant:)
  removed = 0
  @store.transaction do
    proposal_keys.each do |k|
      record = @store.get(SCOPE, k)
      next unless record && record["tenant"] == tenant_key(tenant)

      @store.delete(SCOPE, k)
      removed += 1
    end
  end
  removed
end

#purge_customer(tenant:, customer:) ⇒ Object

One customer's proposals, EVERY status. -> count removed.



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

def purge_customer(tenant:, customer:)
  removed = 0
  @store.transaction do
    proposal_keys.each do |k|
      record = @store.get(SCOPE, k)
      next unless record && record["tenant"] == tenant_key(tenant)
      next unless record["customer"] == customer.to_s

      @store.delete(SCOPE, k)
      removed += 1
    end
  end
  removed
end

#reject(id:, operator: nil, note: nil, now: Time.now.utc) ⇒ Object 



122
123
124
 
# File 'lib/insika/proposal_store.rb', line 122

def reject(id:, operator: nil, note: nil, now: Time.now.utc)
  transition(id, "rejected", operator: operator, note: note, now: now)
end

#resolved(limit: 20) ⇒ Object

The wiki's Recent list: every terminal status (approved/rejected/ dismissed), most recent first — the operator's audit trail.



88
89
90
91
92
93
 
# File 'lib/insika/proposal_store.rb', line 88

def resolved(limit: 20)
  scan.select { |p| TERMINAL.include?(p.status) }
      .sort_by { |p| p.updated_at.to_s }
      .reverse
      .first(limit)
end

#stale(limit: 50) ⇒ Object 



80
81
82
83
84
 
# File 'lib/insika/proposal_store.rb', line 80

def stale(limit: 50)
  scan.select { |p| p.status == "stale" }
      .sort_by { |p| p.updated_at.to_s }
      .first(limit)
end