Class: Insika::Commands::RunDistillation

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

Overview

the ONLY path that writes proposals. Distills ONE session end to end: read the transcript and the memory baseline, ask the model, schema-drop, dedup against the ledger, write proposals, mark the session distilled. Synchronous (it runs on the engine's worker fiber, C5) — it creates no task and no turn (D2: no TaskStore work unit — a crash mid-pass leaves the marker unwritten and the next pass re-scans; a duplicate proposal is filtered by the ledger, never applied).

Constant Summary collapse

DEFAULT_IDLE_HOURS = 
6
DEFAULT_MIN_MESSAGES = 
3
DEFAULT_MAX_PROPOSALS = 
10

Instance Method Summary collapse

Constructor Details

#initialize(profiles:, proposal_store:, session_store:, memory_store:, settings_store:, event_stream:, distiller_factory: nil) ⇒ RunDistillation

Returns a new instance of RunDistillation.



17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/insika/commands/run_distillation.rb', line 17

def initialize(profiles:, proposal_store:, session_store:, memory_store:,
               settings_store:, event_stream:, distiller_factory: nil)
  @profiles = profiles
  @proposal_store = proposal_store
  @session_store = session_store
  @memory_store = memory_store
  @settings_store = settings_store
  @event_stream = event_stream
  @distiller_factory = distiller_factory ||
                       ->(config) { Distill::DistillerFactory.build(config, utility_model: utility_model) }
end

Instance Method Details

#call(command) ⇒ Object

payload: { session_id: } (agent resolved from the session, D6-bis) -> { distilled: true, proposals: N, dropped: ..., deduped: N, cost: ... | nil } | { distilled: false, skipped: "already|untagged|no_agent|disabled| too_fresh|too_short|no_model" }

Raises:



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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
101
102
103
104
105
106
107
108
109
 
# File 'lib/insika/commands/run_distillation.rb', line 33

def call(command)
  session_id = Coercion.presence(command.payload[:session_id] || command.payload["session_id"])
  raise ValidationError, "session_id is required" if session_id.nil?

  session = @session_store.find(session_id)
  raise Insika::NotFoundError, "session not found: #{session_id}" if session.nil?

  return skip("already") if @proposal_store.distilled?(session_id)

  customer = Coercion.presence(session.vars["customer"])
  return skip("untagged") if customer.nil?

  agent_id = Coercion.presence(session.vars["agent"])
  return skip("no_agent") if agent_id.nil?

  profile = @profiles[agent_id]
  return skip("no_agent") if profile.nil?

  config = Coercion.deep_stringify(profile.distill)
  return skip("disabled") if config.nil? || !Coercion.truthy?(config["enabled"])

  # D4: the pack's own idle threshold wins over the scan's lower bound —
  # a pack that wants 12 h is never distilled at 6.
  idle_hours = config["idle_hours"].to_i
  idle_hours = DEFAULT_IDLE_HOURS unless idle_hours.positive?
  return skip("too_fresh") unless idle?(session.updated_at, idle_hours)

  min_messages = config["min_messages"].to_i
  min_messages = DEFAULT_MIN_MESSAGES unless min_messages.positive?
  return skip("too_short") if session.messages.size < min_messages

  tenant = tenant_of(session_id)
  distiller = @distiller_factory.call(config)
  return skip("no_model") if distiller.nil?

  baseline = memory_baseline(tenant, customer)
  prompt = build_prompt(config, session, baseline)
  result = distiller.distill(prompt: prompt, message_count: session.messages.size,
                             max_proposals: (config["max_proposals"] || DEFAULT_MAX_PROPOSALS).to_i)

  deduped = 0
  survivors = result[:proposals].reject do |proposal|
    dedup = deduped?(tenant, customer, proposal, baseline)
    deduped += 1 if dedup
    dedup
  end

  survivors.each do |proposal|
    expected = baseline[proposal["name"]]
    @proposal_store.create(
      tenant: tenant, customer: customer, session_ref: session_id,
      key: proposal["name"], value: proposal["value"],
      confidence: proposal["confidence"], evidence: proposal["turns"],
      expected_revision: expected && expected.updated_at,
      expected_existed: !expected.nil?
    )
  end

  @proposal_store.mark_distilled(session_id, agent: agent_id,
                                 proposals: survivors.size,
                                 dropped: result[:dropped],
                                 deduped: deduped,
                                 cost: result[:cost])
  @event_stream.emit(Insika::Event.new(
                       type: :distillation_completed,
                       # counts and ids only (D7) — a fact value never
                       # enters the stream.
                       data: { session_ref: session_id, agent: agent_id,
                               proposals: survivors.size,
                               dropped: result[:dropped],
                               deduped: deduped,
                               cost: result[:cost] },
                       meta: { at: Time.now.utc.iso8601 }
                     ))
  { distilled: true, proposals: survivors.size, dropped: result[:dropped],
    deduped: deduped, cost: result[:cost] }
end