osr-enterprise-builder

RouterBuilder(registry, settings=None) is a fluent builder; every with_* returns the builder and build() returns an EnterpriseRouter (thread-safe facade: route, learn, execute, run, health_snapshot). It raises ConfigurationError for an empty registry, no strategies, or duplicate strategy names.

Reference chain

from opensmartroute.enterprise import (RouterBuilder, CacheMiddleware, TenantMiddleware, TimeoutMiddleware,
                                       LoggingTelemetry, MetricsTelemetry, FileAuditSink)
from opensmartroute.observability import MemorySink, MetricsSink
from opensmartroute.security import GuardMiddleware

metrics = MetricsTelemetry()
app = (
    RouterBuilder(registry)
    .with_rules(load_rules("rules.yaml"))            # declarative preferences (RulesStrategy)
    .with_defaults()                                 # capability + similarity + Thompson bandit
    .with_auto_learning(state_dir=".osr-state")      # IRT + Bradley-Terry + LinUCB + Markov, persisted
    .with_health(latency_slo_ms=3000)                # breakers, rate limits, budgets; HealthPolicy + HealthStrategy
    .with_middleware(GuardMiddleware(redact=True),   # order = execution order (outermost first)
                     TenantMiddleware({"acme": {"max_cost_per_1k": 0.005, "data_boundary": "private",
                                                "deny_targets": ["llm-frontier"]}}),
                     CacheMiddleware(max_size=4096, ttl_s=30),
                     TimeoutMiddleware(budget_ms=250))
    .with_telemetry(LoggingTelemetry(), metrics)     # on_decision / on_outcome / on_error
    .with_tracing(MemorySink(), MetricsSink())       # spans + events of every stage (docs/OBSERVABILITY.md)
    .with_audit(FileAuditSink("audit.jsonl"))        # hash-chained, tamper-evident
    .with_calibration(conformal_alpha=0.1)           # temperature scaling + conformal candidate sets
    .with_retrieval(narrow_above=500, narrow_to=50)  # BM25 + dense RRF narrowing for big catalogues
    .with_llm_judge(LLMJudgeStrategy(judge), escalate_below=0.6)
    .with_objective(Objective(cost=0.3))
    .build()
)

Other options: with_strategy(s, weight), with_policy(policy) (wrapped by HealthPolicy when health is on), with_feedback(FeedbackStore), with_state_store(store, save_every), with_queue_awareness(slo_ms) (in-flight tracking + Erlang-C latency), with_fair_share(weights), with_shadow(candidate_router, mode="shadow"|"ab") (SPRT-judged), with_router_options(**kw), with_components(registry) (decorator SDK).

Optional learners and controls added with with_strategy(...) (each is a Strategy; those with a model are persisted by with_auto_learning / with_state_store like the built-in ones): learning.HistoryTargetStrategy() (multi-turn: history-target joint embeddings), learning.UserAdaptiveStrategy() (per-user shrinkage on context["user_id"]), strategies.EdgeCloudStrategy() (metadata.tier edge/cloud vs latency SLO and data boundary), strategies.TokenBudgetStrategy() (elastic <id>@<budget> siblings), strategies.AuctionStrategy() (bias-corrected second-price bids), strategies.HiddenStateStrategy(state_fn, targets, dim), discovery.SchemaAwareStrategy() (tool input_schema coverage). After the answer: signals.UncertaintyGate as the Cascade quality gate, strategies.ProtocolPolicy to pick single / cascade / aggregate / debate / handoff per request, strategies.MixtureOfAgents for top-k aggregation, strategies.SelfEscalation + wrap_stream for mid-stream escalation and learning.HandoffPolicy for permanent hand-off of failing agent tasks. Energy and carbon: math.EnergyModel / hardware_profile() fill cost.wh_per_1k_tokens / gco2_per_1k_tokens that Objective(energy=, carbon=) weighs.

Middleware contract

class Middleware(ABC):
    def __call__(self, request: RouteRequest, next_: RouteFn) -> RouteDecision: ...

Raise SecurityError/ValidationError to reject; mutate request.constraints to enforce tenant defaults (that is what TenantMiddleware does); wrap next_(request) for caching/timing.

Telemetry contract

Telemetry.on_decision(request, decision), on_outcome(outcome), on_error(request, error). Never log raw text - LoggingTelemetry logs a SHA-256 prefix + length. MetricsTelemetry.snapshot() gives decisions_by_target, route_latency_ms{p50,p95,p99}, policy_rejections_total, mean_confidence. opensmartroute.adapters.OpenTelemetryTelemetry needs the otel extra.

Tracing contract

Telemetry sees decisions; the tracer sees how they were made. with_tracing(*sinks_or_tracer, sample_rate=) (no arguments: Tracer.from_settings() / OSR_OBSERVABILITY_*) records a request span per call with nested route / plan / execute spans and events route.signals, route.policy, route.rank (per-strategy ms), cache.hit|miss, tenant.rejected, route.slow, guard.*, health.breaker, shadow.*, fairshare.*, learn.outcome. Sinks: MemorySink (trace(request_id)), MetricsSink (prometheus()), LoggingSink, FileSink, adapters.optional.OpenTelemetrySink. Inside your own middleware emit with current_tracer().event("name", **attrs) - never pass text, only ids and numbers. See docs/OBSERVABILITY.md.

Persisting learner state

with_auto_learning(state_dir) writes JSON files; for shared infrastructure use enterprise.stores: RedisStateStore, SQLStateStore, plus wrappers EncryptedStateStore (needs crypto extra), VersionedStateStore(Migration...), BatchedStateStore, NamespacedStateStore. Pass with with_state_store(store, save_every=10). Multi-replica: one writer, readers reload; state is quarantined on load failure rather than trusted blindly.

Closing the loop

d = app.route(RouteRequest("...", constraints=RequestConstraints(tenant="acme")))
app.learn(Outcome(d.request_id, d.target.id, success=True, quality=0.9, cost_usd=0.002, latency_ms=1200,
                  domains=d.trace.signals.domains, complexity=d.trace.signals.complexity))
res = app.run("...")          # route -> execute plan -> learn, with the full middleware chain

app.health_snapshot() exposes breaker state per target.health_key (family-aware); app.shadow returns the ShadowMiddleware for A/B statistics.

Checklist before shipping

• Middleware order: guard -> tenant -> cache -> timeout.
• TenantMiddleware(require=True) unless anonymous traffic is expected.
• Set latency_slo_ms, Objective.quality_floor, and per-tenant max_cost_per_1k.
• Persist state (state_dir or a StateStore) and mount it in the container (OSR_STATE).
• Export metrics.snapshot() and keep the audit sink on durable storage.