Class: Insika::ScheduleEngine

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

Overview

the tick-driven FIRER of recurring schedules: the engine's fourth duty (after the outbox drain, retention/funnel and the follow-up firer). One pass per claim window (the followup/funnel idiom); each due schedule is claimed transactionally — re-read inside its own transaction, so two workers racing a window serialize on the backend lock and exactly one fires (the multi-worker at-most-once claim, per row).

Gating order:

declared? (reconciled from the profiles) -> enabled + due -> no-catch-up
(a window older than one claim window is MISSED, recorded, never
replayed) -> overlap (the last task still live) -> budget (a hard
window at/over cap) -> FIRE.

Skips are DATA, never silent: last_skip { at, reason } on the row, for the Studio. The engine never queues — a skipped window advances the schedule's lattice. The turn it creates is delivered by the existing pipeline (it holds no channel code, like the FollowupEngine).

Constant Summary collapse

SCOPE = 
"schedule_fire"
KEY = 
"claim"
DEFAULT_WINDOW =

seconds; one firing worker per window

300
TENANT =

the single-tenant default (ledger rules) — see tenant_for

"platform"
ACTIVE_STATUSES = 
%i[queued running waiting paused].freeze

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(store:, schedule_store:, task_store:, session_store:, profiles:, executor:, budget_ledger: nil, event_stream: nil, window: DEFAULT_WINDOW, now: nil) ⇒ ScheduleEngine

Returns a new instance of ScheduleEngine.



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

def initialize(store:, schedule_store:, task_store:, session_store:,
               profiles:, executor:, budget_ledger: nil, event_stream: nil,
               window: DEFAULT_WINDOW, now: nil)
  @store = store
  @schedule_store = schedule_store
  @task_store = task_store
  @session_store = session_store
  @profiles = profiles
  @executor = executor
  @budget_ledger = budget_ledger
  @event_stream = event_stream
  @window = window
  @now = now
end

Class Method Details

.tenant_for(profile) ⇒ Object

The tenant a SCHEDULED turn declares. Every other turn gets its tenant from its CALLER (an authenticated tenant token, a Command built with tenant:) — a scheduled turn has no caller, so it is the agent's OWN declaration instead: metadata["tenant"], the same "stable per agent, from the pack" home store_id already lives in (never the model, never a policy — just a fact the pack states). Absent -> the single-tenant default, unchanged for every profile that does not declare one.

PUBLIC and STATELESS on purpose: the Studio's own schedule list (studio/app.rb) must resolve to the exact SAME tenant the fire path uses, or a declared schedule becomes invisible there — one formula, two callers, never a second copy to drift.



44
45
46
47
 
# File 'lib/insika/schedule_engine.rb', line 44

def self.tenant_for(profile)
  meta = profile.respond_to?(:metadata) ? profile.metadata : nil
  Insika::Coercion.presence(meta && meta["tenant"]) || TENANT
end

Instance Method Details

#run ⇒ Object

-> { claimed: false } | { claimed: true, fired: N, skipped: N, errors: N, skip_reasons: { "reason" => N } } A StoreError on ONE schedule aborts THAT schedule's transaction (rescued, counted, the loop continues) — a broken row must not hold the other schedules' runs hostage.



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/insika/schedule_engine.rb', line 70

def run
  now_time = @now || Time.now.utc
  return { claimed: false } unless claim_window(now_time)

  sync_from_profiles(now_time)

  fired = 0
  skipped = 0
  errors = 0
  reasons = Hash.new(0)

  @schedule_store.due(now: now_time).each do |record|
    begin
      outcome = fire_record(record, now_time)
      case outcome
      when :fired then fired += 1
      when Array
        # a skip is recorded on the row (never silent) and counted here.
        skipped += 1
        reasons[outcome[1].to_s] += 1
      end
    rescue StandardError
      # a broken schedule must not hold the other schedules' runs
      # hostage — its own transaction already rolled back.
      errors += 1
    end
  end

  { claimed: true, fired: fired, skipped: skipped, errors: errors,
    skip_reasons: reasons }
end