Class: Insika::Reliability

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

Overview

The reliability policy for the turn's provider interaction (WS3): retries with exponential backoff, mid-turn ROTATION across the fallback chain, and the circuit breaker — all as DATA on AgentProfile#reliability, never a parallel code path. The failure classification is B9's (ProviderErrorClassifier): ONLY :retryable / :rate_limited_* ever retry or rotate; a :fatal is re-raised immediately (a poisoned credential must not hammer N models).

Each ATTEMPT is a fresh chat built by the caller (the coordinator yields the selection): a failed ask leaves its user message in the chat, so re-asking the same chat would double the input. The customer-visible answer comes only from the attempt that returns; the fragments of failed attempts ride :intermediate (operator-side) and die with the turn.

The breaker is per (tenant, provider/model): a node whose circuit is open is skipped (fail-fast with CircuitOpenError when the PRIMARY is open — the turn dies in ms, no provider call); a node that tripped mid-turn moves on.

Constant Summary collapse

DEFAULT_TIMEOUT =

seconds per attempt (reliability)

30

Instance Method Summary collapse

Constructor Details

#initialize(circuit_store:, event_stream:, sleeper: nil) ⇒ Reliability

circuit_store: CircuitState (the breaker's durable cells). event_stream: where the reliability events go (:provider_fallback, ...). sleeper: ->(seconds) — the backoff wait; injectable for specs.



27
28
29
30
31
 
# File 'lib/insika/reliability.rb', line 27

def initialize(circuit_store:, event_stream:, sleeper: nil)
  @circuit_store = circuit_store
  @event_stream = event_stream
  @sleeper = sleeper || method(:backoff_wait)
end

Instance Method Details

#call(policy:, tenant:, agent: nil, selection:, chain:, &attempt) ⇒ Object

policy: the profile's reliability data (string keys) — the caller skips this coordinator entirely when nil (parity). tenant: the command tenant (breaker scoping; nil = platform). agent: the agent id (event attribution — WS6 alerts read it). selection: the resolved primary ModelSelection. chain: [{ model:, provider: }] fallback candidates (profile's first, then the platform's resolved fallbacks). attempt: ->(selection, attempt_index) { response } — build the chat for that selection and ask. May RAISE a provider-family error.

-> the successful response. Raises CircuitOpenError (primary open), or the last retryable error when every node exhausted its retries. NOTHING about a run is stored on self: one Reliability instance serves every concurrent turn, and ask is a suspension point — an ivar written here would be read back after another fiber's turn overwrote it, and the WS6 alert would name the wrong agent (and, through it, the wrong tenant). The run's identity rides the stack.

Raises:



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
110
111
112
113
114
115
116
117
 
# File 'lib/insika/reliability.rb', line 50

def call(policy:, tenant:, agent: nil, selection:, chain:, &attempt)
  nodes = ([selection] + Array(chain)).map { |node| { selection: node, tries: 0 } }
  retries = [policy["retries"].to_i, 0].max
  breaker = breaker_config(policy)
  # The per-attempt ceiling. A policy WITHOUT a timeout is DEFAULT_TIMEOUT —
  # nothing config overrides here (the old [.., 1].max silently made every
  # unset profile die in ~1s, WS3).
  configured_timeout = policy["timeout"].to_i
  timeout = configured_timeout.positive? ? configured_timeout : DEFAULT_TIMEOUT
  # declaraed HERE (not inside a block) so the post-loop `raise` sees the
  # method-local binding — a first assignment inside a block would not leak.
  last_error = nil

  # Fail-fast BEFORE any provider call: the PRIMARY's circuit open means
  # this provider family is known-dead — the turn dies in ms.
  if breaker && breaker_open?(tenant, selection, breaker)
    raise circuit_open(tenant, selection, breaker)
  end
  # the last node we actually ASKED: the `from` of a rotation. A node the
  # breaker skipped was never asked, so it is never the `from`.
  asked = nil
  nodes.each do |node|
    selection = node[:selection]
    next if breaker && breaker_open?(tenant, selection, breaker)

    # ROTATION is an EVENT, not an inference from the usage attribution
    # (WS3): the trace names the node we left, the node we moved to and why.
    # Emitted for every node past the first one asked — with or WITHOUT a
    # breaker (a fallback policy with no circuit_breaker used to rotate in
    # complete silence).
    emit_fallback(agent, asked, selection, last_error) if asked
    asked = selection

    attempts = retries + 1
    attempts.times do |index|
      node[:tries] += 1
      begin
        response = with_attempt_timeout(timeout) { yield selection, node[:tries] }
        @circuit_store.record_success(tenant: tenant, ref: ref_of(selection)) if breaker
        return response
      rescue StandardError => e
        last_error = e
        # A :fatal provider error — or ANYTHING that is not a provider
        # failure at all (a bug, a domain error, a guardrail raise) — is
        # never retried, never rotated (B9's structural rule). Only
        # retryable/rate-limited (and the per-attempt timeout we raised)
        # spend the retry budget. The B9 classifier is class-name based, so
        # OUR TimeoutError reads as :fatal — the retryable_failure? check
        # (which owns the reliability-stage timeout) must decide FIRST, or
        # the :fatal guard would swallow it (WS3: a timeout never retried,
        # never rotated).
        retryable = retryable_failure?(e)
        raise unless retryable
        raise if kind_of(e) == :fatal && !e.is_a?(Insika::TimeoutError)

        record_failure(tenant, selection, breaker, e, agent)
        # the last attempt of the last node re-raises; otherwise back off
        # and give the next attempt/node a turn.
        if index < attempts - 1 || node != nodes.last
          @sleeper.call(backoff_seconds(policy, index))
        end
      end
    end
  end
  raise last_error if last_error

  raise Insika::Error, "reliability loop exhausted without a result"
end