Class: Insika::DistillEngine

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

Overview

the trigger — finds due sessions and distills them. It spawns a worker fiber (supervisor child, Tick#start shape) whose loop claims a window and runs the distillation for due sessions ON the worker fiber — off the tick's critical path and off every customer turn's path. The scan uses the engine default idle_hours (6) as the LOWER BOUND; the per-agent value is re-checked inside RunDistillation (a pack that wants 12 h is never distilled at 6).

D2 — no queue: the per-session marker is the claim. A crash mid-pass leaves markers unwritten and the next boot re-scans; the dedup ledger filters duplicates. Two workers racing the same session both run, and the ledger still filters.

Constant Summary collapse

SCOPE = 
"distill"
KEY = 
"claim"
DEFAULT_WINDOW =

seconds — the O(n) scan never rides the 60 s loop

300
DEFAULT_IDLE_HOURS = 
6
DEFAULT_MIN_MESSAGES = 
3

Instance Method Summary collapse

Constructor Details

#initialize(store:, proposal_store:, session_store:, runner:, profiles: nil, logger: nil, window: DEFAULT_WINDOW, idle_hours: DEFAULT_IDLE_HOURS, sleeper: nil) ⇒ DistillEngine

Returns a new instance of DistillEngine.



25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/insika/distill_engine.rb', line 25

def initialize(store:, proposal_store:, session_store:, runner:, profiles: nil,
               logger: nil, window: DEFAULT_WINDOW,
               idle_hours: DEFAULT_IDLE_HOURS, sleeper: nil)
  @store = store
  @proposal_store = proposal_store
  @session_store = session_store
  @runner = runner
  @profiles = profiles # profile source; nil/empty = nothing distills (parity)
  @logger = logger
  @window = window.to_i
  @idle_hours = idle_hours.to_i
  @sleeper = sleeper || method(:default_sleep)
end

Instance Method Details

#run_once ⇒ Object

One pass. -> { claimed: false } | { claimed: true, distilled: N, skipped: N, errors: N }



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/insika/distill_engine.rb', line 63

def run_once
  return { claimed: false } unless distillable?
  return { claimed: false } unless claim_window

  distilled = 0
  skipped = 0
  errors = 0
  due_sessions.each do |session|
    begin
      outcome = @runner.call(Insika::Command.build(:run_distillation,
                                                   { session_id: session.id }))
      if outcome[:distilled]
        distilled += 1
      else
        skipped += 1
      end
    rescue StandardError
      # a broken session must not hold the pass; the marker discipline
      # keeps it re-runnable.
      errors += 1
    end
  end
  { claimed: true, distilled: distilled, skipped: skipped, errors: errors }
end

#start(parent:) ⇒ Object

The supervisor child (tick.rb:69-83's shape): loop { sleeper(window); run_once rescue log }. Returns false when idle_hours <= 0 (the engine default OFF switch — parity) OR when no profile declares a distillation (a scan with nothing to distill would re-read every session record every window forever — each pass skips at the command, writes no marker, and repeats).



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

def start(parent:)
  return false unless distillable?
  return true if @task&.running?

  @task = parent.async do |t|
    t.annotate("insika-distill")
    loop do
      @sleeper.call(@window)
      run_once
    rescue StandardError => e
      log(:warn, "distill pass failed: #{e.class}: #{e.message}")
    end
  end
  true
end