Class: Insika::HarvestStore

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

Overview

the durable half of the harvest. One record per mining RUN (window, candidates, cost), the per-candidate lifecycle (the unit a human acts on — D8), the APPEND-ONLY promotion log, the pre-promotion snapshots, and the per-session mined markers (D10's re-scan discipline). A dumb domain store: no policy (which candidate is worth gating is the gates'), no model, no skills — the SkillStore stays the skill's home.

Key shapes (string keys, Store-contract JSON):

"harvest"   "run:<agent>:<started_at>:<id>"     -> the mining run
"harvest"   "cand:<id>"                         -> one candidate lifecycle
"harvest"   "promo:<agent>:<at>:<id>"         -> APPEND ONLY (D8)
"harvest"   "snap:<id>"                       -> the pre-promotion state
"harvest"   "session:<session_ref>"           -> the D10 marker

The agent id must not contain ":" (the run-key split — the RefinementStore rule). The body is the only content the store holds: behavior instructions, the same trust level as SkillStore content (D11).

Defined Under Namespace

Classes: Candidate, Promotion, Run, Snapshot

Constant Summary collapse

SCOPE = 
"harvest"
RUN_STATUSES =

Run statuses: mining -> completed | no_candidates | failed. The run does NOT carry awaiting_approval: several candidates of one run can be at different gates (D8) — the human answers CANDIDATES, the run only frames.

%w[mining completed no_candidates failed].freeze
CANDIDATE_STATUSES =

Candidate statuses (the lifecycle a human acts on):

pending -> gated -> awaiting_approval -> promoted | rejected
     (gated FAIL -> rejected with the report; the latch)
%w[pending gated awaiting_approval promoted rejected].freeze
OPEN_STATUSES =

A candidate is OPEN (dedup-suppressing) until terminal.

%w[pending gated awaiting_approval].freeze

Instance Method Summary collapse

Constructor Details

#initialize(store:, session_store: nil) ⇒ HarvestStore

session_store: optional — the session source for unmined_sessions (D10). nil = the scan is inert (the engine may enumerate sessions itself; the store never invents a session space).



53
54
55
56
 
# File 'lib/insika/harvest_store.rb', line 53

def initialize(store:, session_store: nil)
  @store = store
  @sessions = session_store
end

Instance Method Details

#append_promotion(id:, agent:, skill:, origin: [], eval_ref: nil, conversion_ref: nil, approver:, snapshot_ref: nil, criterion_sha: nil, at: nil) ⇒ Object

Appends the promotion row — NO update, NO delete, no re-key. A second row with the same id is refused loudly.

Raises:



248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
 
# File 'lib/insika/harvest_store.rb', line 248

def append_promotion(id:, agent:, skill:, origin: [], eval_ref: nil,
                     conversion_ref: nil, approver:, snapshot_ref: nil,
                     criterion_sha: nil, at: nil)
  id = id.to_s
  agent = agent.to_s
  raise Insika::ValidationError, "promotion id is required" if id.empty?
  raise Insika::ValidationError, "promotion already recorded: #{id}" if find_promotion_key(id)

  now = at || timestamp
  record = {
    "id" => id, "agent" => agent, "skill" => skill.to_s,
    "origin" => Array(origin).map(&:to_s), "eval_ref" => eval_ref.to_s,
    "conversion_ref" => conversion_ref.to_s, "approver" => (Coercion.presence(approver) || "operator").to_s,
    "snapshot_ref" => snapshot_ref.to_s, "criterion_sha" => criterion_sha.to_s,
    "rolled_back_at" => nil, "at" => now
  }
  @store.set(SCOPE, "promo:#{agent}:#{now}:#{id}", record)
  to_promotion(record)
end

#append_rollback(promotion_id:, operator: nil, reason: nil) ⇒ Object

Stamps rolled_back_at on the row — the log stays the single ledger (D9). -> Promotion.

Raises:



270
271
272
273
274
275
276
277
278
 
# File 'lib/insika/harvest_store.rb', line 270

def append_rollback(promotion_id:, operator: nil, reason: nil)
  key = find_promotion_key(promotion_id.to_s)
  raise Insika::NotFoundError, "promotion not found: #{promotion_id}" if key.nil?

  record = @store.get(SCOPE, key)
  record["rolled_back_at"] = timestamp
  @store.set(SCOPE, key, record)
  to_promotion(record)
end

#attach_gate(candidate_id, eval_gate:, conversion_gate:, criterion_sha:) ⇒ Object

Records both gate reports + the boot-loaded criterion sha; pending -> gated.



186
187
188
189
190
191
192
193
194
195
 
# File 'lib/insika/harvest_store.rb', line 186

def attach_gate(candidate_id, eval_gate:, conversion_gate:, criterion_sha:)
  update_candidate(candidate_id) do |record|
    guard_candidate_state!(record, "pending")
    record["eval_gate"] = Coercion.deep_stringify(eval_gate)
    record["conversion_gate"] = Coercion.deep_stringify(conversion_gate)
    record["criterion_sha"] = criterion_sha.to_s
    record["status"] = "gated"
    record["updated_at"] = timestamp
  end
end

#attach_rejected(candidate_id, rules:, reason:) ⇒ Object

Appends a { rule, reason } rejection entry (the negative-list / gate evidence lines). Accumulates; only on a non-terminal candidate.



176
177
178
179
180
181
182
 
# File 'lib/insika/harvest_store.rb', line 176

def attach_rejected(candidate_id, rules:, reason:)
  update_candidate(candidate_id) do |record|
    guard_candidate_non_terminal!(record)
    list = record["rejected"] || []
    record["rejected"] = list + Array(rules).map { |r| { "rule" => r.to_s, "reason" => reason.to_s } }
  end
end

#awaiting_approval(limit: 50) ⇒ Object

-> [Candidate] EVERY candidate parked on a human, most recent first.



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

def awaiting_approval(limit: 50)
  list = candidates(status: "awaiting_approval").sort_by { |c| c.updated_at.to_s }.reverse
  list.first(limit)
end

#candidates(agent_id: nil, status: nil) ⇒ Object

-> [Candidate] filtered by agent/status, most recent first.



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

def candidates(agent_id: nil, status: nil)
  scan_candidates.select do |c|
    (agent_id.nil? || c.agent == agent_id.to_s) &&
      (status.nil? || c.status == status.to_s)
  end
end

#complete_run(id, candidates:, cost: nil, rejected: nil) ⇒ Object

Closes a run with its candidate count. Zero candidates -> :no_candidates (a distinct outcome from :completed — "we looked and it was clean"). rejected is the { reason/rule => count } map the filters counted (D4: "every rejected-by-list candidate is logged with the matching rule"). -> Run. ArgumentError when the run is already terminal.



86
87
88
89
90
91
92
93
94
95
 
# File 'lib/insika/harvest_store.rb', line 86

def complete_run(id, candidates:, cost: nil, rejected: nil)
  update_run(id) do |record|
    guard_run_state!(record, "mining")
    record["candidates"] = Integer(candidates)
    record["rejected"] = rejected ? Coercion.deep_stringify(rejected) : {}
    record["cost"] = Coercion.deep_stringify(cost) if cost
    record["status"] = Integer(candidates).positive? ? "completed" : "no_candidates"
    record["finished_at"] = timestamp
  end
end

#create_candidate(run_id:, agent:, name:, description:, body:, triggers: [], rationale:, origin:, evidence_turns: [], proposer:, id: SecureRandom.uuid) ⇒ Object

The engine stamps agent/origin/proposer — never the model (D3).



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/insika/harvest_store.rb', line 122

def create_candidate(run_id:, agent:, name:, description:, body:,
                     triggers: [], rationale:, origin:, evidence_turns: [],
                     proposer:, id: SecureRandom.uuid)
  now = timestamp
  record = {
    "id" => id.to_s, "run_id" => run_id.to_s, "agent" => agent.to_s,
    "name" => name.to_s, "description" => description.to_s, "body" => body.to_s,
    "triggers" => Array(triggers).map(&:to_s), "rationale" => rationale.to_s,
    "origin" => Array(origin).map(&:to_s),
    "evidence_turns" => Array(evidence_turns).map(&:to_i),
    "proposer" => proposer.to_s, "status" => "pending",
    "rejected" => [], "eval_gate" => nil, "conversion_gate" => nil,
    "criterion_sha" => nil, "decision" => nil, "promotion_ref" => nil,
    "created_at" => now, "updated_at" => now
  }
  @store.set(SCOPE, "cand:#{id}", record)
  to_candidate(record)
end

#create_run(agent_id:, window: {}, budget: nil, id: SecureRandom.uuid) ⇒ Object

Opens a run (:mining). window is the miner's window as data ({ "last_sessions" => N } | { "since" => iso8601 } | { "session_ids" => [...] }). budget is the pack's harvest.miner.budget (the cap the gates read — the refinement budget discipline). -> Run.

Raises:



64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/insika/harvest_store.rb', line 64

def create_run(agent_id:, window: {}, budget: nil, id: SecureRandom.uuid)
  agent = agent_id.to_s
  raise Insika::ValidationError, "agent_id is required" if agent.empty?
  raise Insika::ValidationError, "agent_id must not contain ':'" if agent.include?(":")

  started = timestamp
  record = {
    "id" => id.to_s, "agent_id" => agent, "status" => "mining",
    "window" => Coercion.deep_stringify(window || {}),
    "candidates" => 0, "rejected" => {}, "budget" => Coercion.deep_stringify(budget),
    "cost" => nil,
    "started_at" => started, "finished_at" => nil, "error" => nil
  }
  @store.set(SCOPE, "run:#{agent}:#{started}:#{id}", record)
  to_run(record)
end

#create_snapshot(agent:, skill:, content:, existed:, enabled_for:) ⇒ Object

The pre-promotion state (D8 — snapshot FIRST, then the writes).



288
289
290
291
292
293
294
295
296
 
# File 'lib/insika/harvest_store.rb', line 288

def create_snapshot(agent:, skill:, content:, existed:, enabled_for:)
  id = SecureRandom.uuid
  record = { "id" => id, "agent" => agent.to_s, "skill" => skill.to_s,
             "content" => content, "existed" => !!existed,
             "enabled_for" => Array(enabled_for).map(&:to_s),
             "at" => timestamp }
  @store.set(SCOPE, "snap:#{id}", record)
  to_snapshot(record)
end

#delete_older_than(time) ⇒ Object

Candidates (pending AND terminal), log rows, snapshots and runs past the cutoff. They are re-derivable (D2) — pruning is never data loss. The session MARKERS are never pruned (the marker is the claim). -> count removed.



391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
 
# File 'lib/insika/harvest_store.rb', line 391

def delete_older_than(time)
  cutoff = Time.parse(time.to_s).utc
  removed = 0

  [["cand:", "created_at"], ["snap:", "at"], ["run:", "started_at"]].each do |prefix, field|
    @store.list(SCOPE, prefix).each do |k|
      record = @store.get(SCOPE, k)
      next unless record && record[field]

      begin
        next unless Time.parse(record[field].to_s).utc < cutoff
      rescue ArgumentError
        next
      end

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

  @store.list(SCOPE, "promo:").each do |k|
    record = @store.get(SCOPE, k)
    next unless record && record["at"]

    begin
      next unless Time.parse(record["at"].to_s).utc < cutoff
    rescue ArgumentError
      next
    end

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

  removed
end

#fail_run(id, error:) ⇒ Object

Closes a run as :failed, recording the error. -> Run.



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

def fail_run(id, error:)
  update_run(id) do |record|
    guard_run_state!(record, "mining")
    record["status"] = "failed"
    record["error"] = error.to_s
    record["finished_at"] = timestamp
  end
end

#find_candidate(id) ⇒ Object 



141
142
143
 
# File 'lib/insika/harvest_store.rb', line 141

def find_candidate(id)
  to_candidate(@store.get(SCOPE, "cand:#{id}"))
end

#find_run(id) ⇒ Object 



107
108
109
110
 
# File 'lib/insika/harvest_store.rb', line 107

def find_run(id)
  key = run_key(id)
  key && to_run(@store.get(SCOPE, key))
end

#find_snapshot(id) ⇒ Object 



298
299
300
 
# File 'lib/insika/harvest_store.rb', line 298

def find_snapshot(id)
  to_snapshot(@store.get(SCOPE, "snap:#{id}"))
end

#mark_awaiting(candidate_id) ⇒ Object

gated -> awaiting_approval. The refusal parks: a conversion REFUSAL keeps the candidate AT gated (the page shows the ruler's hole).



199
200
201
202
203
204
205
 
# File 'lib/insika/harvest_store.rb', line 199

def mark_awaiting(candidate_id)
  update_candidate(candidate_id) do |record|
    guard_candidate_state!(record, "gated")
    record["status"] = "awaiting_approval"
    record["updated_at"] = timestamp
  end
end

#mark_mined(session_ref, candidates: 0) ⇒ Object

---- the per-session marker (D10 - the re-scan discipline) ----------------



304
305
306
307
308
309
310
311
312
 
# File 'lib/insika/harvest_store.rb', line 304

def mark_mined(session_ref, candidates: 0)
  ref = session_ref.to_s
  return {} if ref.empty?

  record = { "session_ref" => ref, "mined_at" => timestamp,
             "candidates" => Integer(candidates) }
  @store.set(SCOPE, "session:#{ref}", record)
  record
end

#mark_promoted(candidate_id, promotion_ref:) ⇒ Object

awaiting_approval -> promoted. Recorded ONLY after the writes land (the record-after rule).



221
222
223
224
225
226
227
228
 
# File 'lib/insika/harvest_store.rb', line 221

def mark_promoted(candidate_id, promotion_ref:)
  update_candidate(candidate_id) do |record|
    guard_candidate_state!(record, "awaiting_approval")
    record["status"] = "promoted"
    record["promotion_ref"] = promotion_ref.to_s
    record["updated_at"] = timestamp
  end
end

#mark_rejected(candidate_id, operator:, note: nil) ⇒ Object

-> rejected (terminal) from pending | gated | awaiting_approval — a human may always outvote the miner. The decision records by/at/note.



209
210
211
212
213
214
215
216
217
 
# File 'lib/insika/harvest_store.rb', line 209

def mark_rejected(candidate_id, operator:, note: nil)
  update_candidate(candidate_id) do |record|
    guard_candidate_non_terminal!(record)
    record["status"] = "rejected"
    record["decision"] = { "by" => (Coercion.presence(operator) || "operator").to_s,
                           "at" => timestamp, "note" => Coercion.presence(note) }.compact
    record["updated_at"] = timestamp
  end
end

#mined?(session_ref) ⇒ Boolean

Returns:

  • (Boolean) 


314
315
316
 
# File 'lib/insika/harvest_store.rb', line 314

def mined?(session_ref)
  !@store.get(SCOPE, "session:#{session_ref.to_s}").nil?
end

#open_pending?(agent:, name:) ⇒ Boolean

Dedup: an OPEN (non-terminal) candidate with the same (agent, name) suppresses a re-proposal. -> bool.

Returns:

  • (Boolean) 


161
162
163
164
165
166
167
168
169
 
# File 'lib/insika/harvest_store.rb', line 161

def open_pending?(agent:, name:)
  @store.list(SCOPE, "cand:").any? do |k|
    record = @store.get(SCOPE, k)
    next false unless record
    next false unless record["agent"] == agent.to_s && record["name"] == name.to_s

    OPEN_STATUSES.include?(record["status"].to_s)
  end
end

#promotions(agent_id: nil, limit: 100) ⇒ Object

-> [Promotion] new-first, optionally per agent, capped.



281
282
283
284
285
 
# File 'lib/insika/harvest_store.rb', line 281

def promotions(agent_id: nil, limit: 100)
  keys = @store.list(SCOPE, agent_id ? "promo:#{agent_id}:" : "promo:")
  rows = keys.filter_map { |k| to_promotion(@store.get(SCOPE, k)) }
  rows.sort_by { |p| p.at.to_s }.reverse.first(limit)
end

#purge(tenant:) ⇒ Object

The tenant is the store's (candidates reference sessions; sessions carry the tenant prefix) — a prefix scan over the scope keys. Removes the tenant's candidates, promotion rows, their snapshots and the session markers. -> count removed.



346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
 
# File 'lib/insika/harvest_store.rb', line 346

def purge(tenant:)
  prefix = "#{tenant_id(tenant)}:"
  removed = 0

  promo_refs = []
  @store.list(SCOPE, "promo:").each do |k|
    record = @store.get(SCOPE, k)
    next unless record && Array(record["origin"]).any? { |o| o.to_s.start_with?(prefix) }

    @store.delete(SCOPE, k)
    promo_refs << record["snapshot_ref"].to_s
    removed += 1
  end

  @store.list(SCOPE, "cand:").each do |k|
    record = @store.get(SCOPE, k)
    next unless record && Array(record["origin"]).any? { |o| o.to_s.start_with?(prefix) }

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

  promo_refs.each do |ref|
    next if ref.empty?

    snap_key = "snap:#{ref}"
    existing = @store.get(SCOPE, snap_key)
    next unless existing

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

  @store.list(SCOPE, "session:#{prefix}").each do |k|
    @store.delete(SCOPE, k)
    removed += 1
  end

  removed
end

#recheck_conversion(candidate_id, conversion_gate:) ⇒ Object

D8-bis: the conversion ruler is re-read AT APPROVE TIME and may have moved below the threshold since gating. The promote path re-checks; a dip parks the candidate BACK at gated with the FRESH report — the operator re-decides, and a skill does not land into a ruler that has moved the wrong way. awaiting_approval -> gated.



235
236
237
238
239
240
241
242
 
# File 'lib/insika/harvest_store.rb', line 235

def recheck_conversion(candidate_id, conversion_gate:)
  update_candidate(candidate_id) do |record|
    guard_candidate_state!(record, "awaiting_approval")
    record["conversion_gate"] = Coercion.deep_stringify(conversion_gate)
    record["status"] = "gated"
    record["updated_at"] = timestamp
  end
end

#runs_for(agent_id, limit: 20) ⇒ Object

-> [Run] for one agent, MOST RECENT FIRST, capped.



113
114
115
116
117
 
# File 'lib/insika/harvest_store.rb', line 113

def runs_for(agent_id, limit: 20)
  keys = @store.list(SCOPE, "run:#{agent_id}:").reverse
  keys = keys.first(limit) if limit
  keys.filter_map { |k| to_run(@store.get(SCOPE, k)) }
end

#unmined_sessions(since: nil) ⇒ Object

The engine's scan space: every session the source knows MINUS the marked set, optionally bounded by since (sessions updated at/after it — the incremental boundary). Nil source = inert (parity — the caller that owns a SessionStore enumerates itself). -> [String]



323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
 
# File 'lib/insika/harvest_store.rb', line 323

def unmined_sessions(since: nil)
  return [] unless @sessions

  @sessions.each_id.filter_map do |id|
    next if mined?(id)

    if since
      session = @sessions.respond_to?(:find) ? @sessions.find(id) : nil
      next unless session
      next unless Time.parse(session.updated_at.to_s).utc >= Time.parse(since.to_s).utc
    end
    id.to_s
  rescue ArgumentError
    next
  end.to_a
end