feat(llm-router): standalone LLM router worker in Rust - #241
Merged
Conversation
Contributor
skill-check — worker0 verified, 16 skipped (no docs/).
Four for four. Nicely done. |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds a new Rust llm-router worker: crate and manifest, typed router contracts, durable registry/catalog stores, provider registration/resolution, chat streaming orchestration with retries/abort, engine function/trigger wiring, integration tests, and documentation/permissions. ChangesLLM Router Implementation
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
Suggested reviewers
Poem
✨ Finishing Touches🧪 Generate unit tests (beta)
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 16
🧹 Nitpick comments (4)
llm-router/src/catalog/handlers.rs (2)
56-62: ⚡ Quick winEmpty string defaults mask missing required parameters.
Lines 58-60 use
unwrap_or("")to default missingprovider,id, andcapabilityparameters to empty strings. Since all three are required to check capability support, callers who omit these fields will receivefalse(failing closed) rather than an explicit error. While fail-closed is safe, it makes debugging harder when callers provide malformed requests.♻️ Proposed fix to validate required parameters
Box::pin(async move { + let provider = raw.get("provider").and_then(Value::as_str) + .filter(|s| !s.is_empty()) + .ok_or_else(|| IIIError::Other("provider parameter is required".into()))?; + let id = raw.get("id").and_then(Value::as_str) + .filter(|s| !s.is_empty()) + .ok_or_else(|| IIIError::Other("id parameter is required".into()))?; + let capability = raw.get("capability").and_then(Value::as_str) + .filter(|s| !s.is_empty()) + .ok_or_else(|| IIIError::Other("capability parameter is required".into()))?; let supported = models_supports( &catalog, - raw.get("provider").and_then(Value::as_str).unwrap_or(""), - raw.get("id").and_then(Value::as_str).unwrap_or(""), - raw.get("capability").and_then(Value::as_str).unwrap_or(""), + provider, + id, + capability, ) .await;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@llm-router/src/catalog/handlers.rs` around lines 56 - 62, The code currently passes empty strings into models_supports by using raw.get(...).and_then(Value::as_str).unwrap_or(""), which masks missing required parameters; instead validate that "provider", "id", and "capability" are present and are strings before calling models_supports (e.g., extract with raw.get(...).and_then(Value::as_str) and if any is None return an explicit error/BadRequest response), then call models_supports with the validated values; update the handler around the models_supports call to return a clear error when parameters are missing rather than silently defaulting to "".
36-41: ⚡ Quick winEmpty string defaults mask missing required parameters.
Lines 38-39 use
unwrap_or("")to default missingproviderandidparameters to empty strings. Sincemodels_getrequires both parameters to identify a model, callers who omit these fields will receiveNone(line 42-45 returnsValue::Null) rather than an explicit error indicating the malformed request. This makes debugging harder when callers forget to provide required parameters.♻️ Proposed fix to validate required parameters
Box::pin(async move { + let provider = raw.get("provider").and_then(Value::as_str) + .filter(|s| !s.is_empty()) + .ok_or_else(|| IIIError::Other("provider parameter is required".into()))?; + let id = raw.get("id").and_then(Value::as_str) + .filter(|s| !s.is_empty()) + .ok_or_else(|| IIIError::Other("id parameter is required".into()))?; let model = models_get( &catalog, - raw.get("provider").and_then(Value::as_str).unwrap_or(""), - raw.get("id").and_then(Value::as_str).unwrap_or(""), + provider, + id, ) .await;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@llm-router/src/catalog/handlers.rs` around lines 36 - 41, The code currently defaults missing provider and id to empty strings before calling models_get, which hides missing required parameters; update the handler to validate raw.get("provider") and raw.get("id") (the values accessed by raw.get("provider").and_then(Value::as_str) and raw.get("id").and_then(Value::as_str)) and return an explicit error (e.g., BadRequest / JSON error response) when either is absent or not a string instead of calling models_get with "". Only call models_get(&catalog, provider_str, id_str).await when both provider and id are present.llm-router/src/types/model.rs (1)
68-68: ⚡ Quick winConstrain
execution_modeto an enum instead of free-formString.On Line 68,
Option<String>accepts invalid values and weakens the public contract. Use a serde enum (parallel|sequential) so bad inputs fail fast at deserialization.Suggested change
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum ExecutionMode { + Parallel, + Sequential, +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct AgentFunction { pub name: String, pub description: String, pub parameters: serde_json::Value, // JSON Schema of the arguments #[serde(skip_serializing_if = "Option::is_none")] pub label: Option<String>, #[serde(skip_serializing_if = "Option::is_none")] - pub execution_mode: Option<String>, // "parallel" | "sequential" + pub execution_mode: Option<ExecutionMode>, }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@llm-router/src/types/model.rs` at line 68, Change the free-form field execution_mode: Option<String> to a serde-backed enum so invalid values fail deserialization: add an enum named ExecutionMode (variants like Parallel and Sequential) with #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] and use #[serde(rename_all = "lowercase")] or per-variant #[serde(rename = "...")] so inputs "parallel"|"sequential" map correctly, then change the struct field signature to execution_mode: Option<ExecutionMode> and update any code that constructs or compares the field to use ExecutionMode::Parallel / ExecutionMode::Sequential.llm-router/src/types/router.rs (1)
205-208: ⚡ Quick winMake
ProviderChangedPayload.opa typed enum instead of free-formString.Line 207 currently permits arbitrary values, so typos can silently break consumers of
router::provider::changed.Suggested fix
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProviderChangedOp { + Register, + Available, + Unavailable, +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct ProviderChangedPayload { pub provider: String, - pub op: String, // "register" | "available" | "unavailable" + pub op: ProviderChangedOp, }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@llm-router/src/types/router.rs` around lines 205 - 208, The op field on ProviderChangedPayload is currently a free-form String which allows typos; replace it with a strongly-typed enum (e.g., ProviderOp with variants Register, Available, Unavailable) and change ProviderChangedPayload::op to that enum. Derive/implement the necessary traits used in the codebase (Clone/Debug/PartialEq/Serialize/Deserialize/Display/FromStr as appropriate) and add serde renames or attributes so the serialized form matches existing messages. Update any usages/constructions/parsing of ProviderChangedPayload and tests to construct the enum variants instead of raw strings.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@llm-router/src/catalog/store.rs`:
- Around line 68-73: The set_slice function currently calls
serde_json::to_value(&*slices).unwrap_or_default(), which masks serialization
failures and can write Null/empty state; change it to propagate the
serialization error instead: replace unwrap_or_default with handling the Result
from serde_json::to_value (using ? or map_err to convert the serde_json::Error
into the function's IIIError) and only call state_set(&self.iii, CATALOG_KEY,
value).await on successful serialization, ensuring set_slice returns Err on
serialization failure; reference symbols: set_slice, serde_json::to_value,
slices, state_set, CATALOG_KEY, IIIError.
- Around line 28-32: The load method silently replaces corrupted catalog JSON by
using serde_json::from_value(...).unwrap_or_default(); change it to fail on
deserialization errors and surface the error instead of returning an empty map:
replace the unwrap_or_default usage in load (where state_get(&self.iii,
CATALOG_KEY) is deserialized into *self.slices.lock()) with proper error
propagation (e.g., map_err or ? converting serde_json::Error into IIIError) so
deserialization failures return Err(IIIError) with context rather than silently
resetting the catalog (see function load, state_get, CATALOG_KEY, and slices).
In `@llm-router/src/chat/chat.rs`:
- Around line 104-105: The code assumes candidates[0] always exists and will
panic for an empty routing set; update the selection to use candidates.first()
and return a typed router error if None is found, then use that provider value
(cloned) to call self.registry.get(&provider).await.ok_or_else(...) so indexing
is avoided and an explicit error variant is returned when there are no routing
candidates.
- Around line 248-251: The current code awaits call_task (call_task.await)
before processing the relay outcome which causes CallerGone/Aborted paths to
block on provider completion; change the control flow to handle relay outcomes
immediately and not wait for the provider to finish: use a tokio::select! (or an
Abortable/JoinHandle.abort) between the provider task (call_task) and the relay
outcome so that if you observe RelayResult::CallerGone or RelayResult::Aborted
you cancel/abort the provider task (call_task) and return early, otherwise
continue to await the provider result; adjust both the block around
call_task.await at the shown location and the analogous section covering lines
305–320 to implement this non-blocking cancellation behavior.
In `@llm-router/src/chat/complete.rs`:
- Around line 57-60: The error here constructs RouterError::new with
RouterCode::InvalidRequest for the case "stream produced no terminal frame" —
change that RouterCode to a non-client-fault/server-transient code (e.g.,
RouterCode::Internal or RouterCode::ServerError/Transient) so callers aren't
blamed; locate the expression RouterError::new(RouterCode::InvalidRequest,
"stream produced no terminal frame") in complete.rs and replace the RouterCode
variant with the appropriate server-side/transient RouterCode.
In `@llm-router/src/chat/output_tokens.rs`:
- Around line 13-14: Return value currently uses provider_default when
model_ceiling is None, which bypasses the safety soft_cap; change the expression
to clamp provider_default by soft_cap when model_ceiling is None. Replace
model_ceiling.map_or(provider_default, |c| c.min(soft_cap)) with something that
uses provider_default.min(soft_cap) (e.g.
model_ceiling.map_or(provider_default.min(soft_cap), |c| c.min(soft_cap))) so
both branches respect soft_cap; keep references to model_ceiling,
provider_default, and soft_cap.
In `@llm-router/src/config/entry.rs`:
- Around line 37-49: The current read_entry_value function masks all trigger
errors as Value::Null; change it to surface real failures instead of coercing
every Err to Null: update read_entry_value signature to return Result<Value, E>
(e.g., Result<Value, anyhow::Error> or the III trigger error type), use ? to
propagate errors from iii.trigger(TriggerRequest { function_id:
"configuration::get", ... }).await, and only map a genuine “not found” response
to Value::Null (or map a specific NotFound variant to Ok(Value::Null) after
inspecting the trigger result); reference read_entry_value, III, TriggerRequest
and "configuration::get" when making these changes.
In `@llm-router/src/config/on_changed.rs`:
- Around line 68-89: The current spawn (the block using flush_task.lock and
tokio::spawn) both sleeps and executes triggers, so if a newer event aborts that
handle after it has already drained pending, the drained ids are lost; fix by
making the debounce task only perform the sleep and drain, and then immediately
spawn a separate fire-and-forget task to execute the triggers so aborting the
debounce cannot drop work: inside the existing tokio::spawn (the closure
referencing debounce_ms, pending, and iii) call tokio::time::sleep(...).await,
then take the ids with std::mem::take(&mut *pending.lock().unwrap()), and then
spawn a new tokio::spawn(async move { for id in ids { let _ =
iii.trigger(TriggerRequest { function_id:
format!("provider::{id}::refresh_models"), payload: json!({}), action: None,
timeout_ms: None, }).await; } }); leaving flush_task to only hold the debounce
handle.
In `@llm-router/src/config/schema.rs`:
- Around line 28-40: validate_custom_schema currently only checks top-level
properties and misses nested object/array fields; update it to recursively
traverse the JSON Schema structure (walk into "properties" objects and "items"
arrays/objects) and apply the same secretish regex check for every property name
encountered, using the existing secretish regex and the same writeOnly check
(def.get("writeOnly") == Some(&Value::Bool(true))). Ensure the traversal handles
nested "properties" maps and "items" that may be a schema or array of schemas,
and return the same RouterError when any matching field lacks writeOnly: true.
In `@llm-router/src/registry/availability.rs`:
- Around line 32-33: The code determines availability by checking
raw.get("event").and_then(Value::as_str) and using event.contains("disconnect")
to set available, which is fragile; change the logic in the availability
calculation to first prefer a structured boolean (e.g.,
raw.get("available").and_then(Value::as_bool)) and if absent perform exact
event-name matching or an explicit allowlist instead of substring matching
(replace event.contains("disconnect") with a check like event == "disconnect" or
look up event in a HashSet of known disconnect events), updating the assignment
to the available variable accordingly (references: raw.get("event"),
Value::as_str, available).
In `@llm-router/src/registry/register.rs`:
- Around line 96-101: The code silently falls back to Value::Null when
serde_json::to_value(rec.declaration.defaults) fails, which can produce
incorrect schemas; change the logic in the block that builds schema (where
rec.declaration.config_schema.clone().unwrap_or_else(||
default_provider_schema(...))) to instead attempt to serialize
rec.declaration.defaults and propagate the serialization error (using ? or
map_err to convert into the function's error type) so that failures from
serde_json::to_value are returned to the caller rather than substituted with
Value::Null; update the call sites around default_provider_schema,
serde_json::to_value, and rec.declaration.config_schema handling to accept the
propagated Result and return an Err on serialization failure.
In `@llm-router/src/registry/store.rs`:
- Around line 62-65: The persist function currently swallows
serde_json::to_value errors by calling unwrap_or_default(), which can cause
Null/empty state writes; update async fn persist(&self, records:
&HashMap<String, ProviderRecord>) -> Result<(), IIIError> to propagate
serialization errors instead of defaulting: call serde_json::to_value(records)
and return an Err converted to IIIError if serialization fails, then only call
state_set(&self.iii, REGISTRY_KEY, value).await on success (preserve
REGISTRY_KEY and state_set usage and ensure the error conversion maps the
serde_json::Error into IIIError).
- Around line 55-60: The load method currently masks deserialization failures by
using serde_json::from_value(...).unwrap_or_default(), which silently drops
corrupted registry state; update load (the function named load which calls
state_get and assigns to self.records) to handle serde_json::from_value errors
explicitly: call from_value and match the Result, returning an Err(IIIError) (or
mapping the serde error into IIIError) if deserialization fails, or at minimum
log the error with context including REGISTRY_KEY before falling back; remove
unwrap_or_default and ensure records (the Mutex-protected self.records) is only
replaced on successful deserialization so corrupted/mismatched state is surfaced
to operators.
In `@llm-router/src/routing.rs`:
- Around line 36-41: The owners collection currently pulls provider names from
input.catalog without checking whether those providers are still registered;
update the filter to exclude unregistered providers by adding a check against
the runtime/registry of known providers (e.g.,
self.is_provider_registered(provider) or
input.registered_providers.contains(provider.as_str())) before mapping to
p.as_str(); ensure the chain that builds owners (using input.catalog and
input.model) only collects providers that pass that registration check so
stale/unregistered providers are ignored.
In `@llm-router/src/settings.rs`:
- Line 62: The assignment to out.retry_max uses a blind cast from u64 to u32
which can wrap large values; update the parsing to detect values > u32::MAX and
either clamp to u32::MAX or return a validation error instead of using `as` so
truncation cannot occur—modify the code around pos_u64(s.get("retry_max"), ...)
and the out.retry_max assignment to perform a safe conversion (e.g., check the
parsed u64 against u32::MAX and then set out.retry_max = value as u32 or
propagate an error), referencing the out.retry_max field and the pos_u64 call to
find the location.
In `@llm-router/src/types/credential.rs`:
- Around line 3-20: The derived Debug on enum Credential exposes secrets
(ApiKey.key, Oauth.access_token/refresh_token) — replace the auto-derived Debug
with a custom impl for Credential that redacts sensitive fields: implement
std::fmt::Debug for Credential and format ApiKey and Oauth variants so that key,
access_token, and refresh_token are replaced with a fixed redact string (e.g.,
"<redacted>") while still displaying non-sensitive fields (expires_at, scopes,
provider_extra) for diagnostics; keep Clone/PartialEq/Serialize/Deserialize
derives unchanged and ensure the impl covers both Credential::ApiKey { key } and
Credential::Oauth { access_token, refresh_token, expires_at, scopes,
provider_extra }.
---
Nitpick comments:
In `@llm-router/src/catalog/handlers.rs`:
- Around line 56-62: The code currently passes empty strings into
models_supports by using raw.get(...).and_then(Value::as_str).unwrap_or(""),
which masks missing required parameters; instead validate that "provider", "id",
and "capability" are present and are strings before calling models_supports
(e.g., extract with raw.get(...).and_then(Value::as_str) and if any is None
return an explicit error/BadRequest response), then call models_supports with
the validated values; update the handler around the models_supports call to
return a clear error when parameters are missing rather than silently defaulting
to "".
- Around line 36-41: The code currently defaults missing provider and id to
empty strings before calling models_get, which hides missing required
parameters; update the handler to validate raw.get("provider") and raw.get("id")
(the values accessed by raw.get("provider").and_then(Value::as_str) and
raw.get("id").and_then(Value::as_str)) and return an explicit error (e.g.,
BadRequest / JSON error response) when either is absent or not a string instead
of calling models_get with "". Only call models_get(&catalog, provider_str,
id_str).await when both provider and id are present.
In `@llm-router/src/types/model.rs`:
- Line 68: Change the free-form field execution_mode: Option<String> to a
serde-backed enum so invalid values fail deserialization: add an enum named
ExecutionMode (variants like Parallel and Sequential) with #[derive(Serialize,
Deserialize, Clone, Debug, PartialEq)] and use #[serde(rename_all =
"lowercase")] or per-variant #[serde(rename = "...")] so inputs
"parallel"|"sequential" map correctly, then change the struct field signature to
execution_mode: Option<ExecutionMode> and update any code that constructs or
compares the field to use ExecutionMode::Parallel / ExecutionMode::Sequential.
In `@llm-router/src/types/router.rs`:
- Around line 205-208: The op field on ProviderChangedPayload is currently a
free-form String which allows typos; replace it with a strongly-typed enum
(e.g., ProviderOp with variants Register, Available, Unavailable) and change
ProviderChangedPayload::op to that enum. Derive/implement the necessary traits
used in the codebase
(Clone/Debug/PartialEq/Serialize/Deserialize/Display/FromStr as appropriate) and
add serde renames or attributes so the serialized form matches existing
messages. Update any usages/constructions/parsing of ProviderChangedPayload and
tests to construct the enum variants instead of raw strings.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 49c55183-751a-4b8a-a040-b1845de164c8
⛔ Files ignored due to path filters (1)
llm-router/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (52)
llm-router/.gitignorellm-router/Cargo.tomlllm-router/README.mdllm-router/build.rsllm-router/iii-permissions.yamlllm-router/iii.worker.yamlllm-router/src/catalog/handlers.rsllm-router/src/catalog/mod.rsllm-router/src/catalog/queries.rsllm-router/src/catalog/reconcile.rsllm-router/src/catalog/store.rsllm-router/src/channels.rsllm-router/src/chat/abort.rsllm-router/src/chat/chat.rsllm-router/src/chat/complete.rsllm-router/src/chat/inflight.rsllm-router/src/chat/mod.rsllm-router/src/chat/output_tokens.rsllm-router/src/chat/pricing.rsllm-router/src/chat/relay.rsllm-router/src/chat/retry.rsllm-router/src/chat/synthesize.rsllm-router/src/config/entry.rsllm-router/src/config/fingerprint.rsllm-router/src/config/mod.rsllm-router/src/config/on_changed.rsllm-router/src/config/schema.rsllm-router/src/lib.rsllm-router/src/main.rsllm-router/src/manifest.rsllm-router/src/register.rsllm-router/src/registry/availability.rsllm-router/src/registry/mod.rsllm-router/src/registry/register.rsllm-router/src/registry/resolve.rsllm-router/src/registry/store.rsllm-router/src/routing.rsllm-router/src/settings.rsllm-router/src/state.rsllm-router/src/testkit/fake_channels.rsllm-router/src/testkit/mod.rsllm-router/src/triggers.rsllm-router/src/types/content.rsllm-router/src/types/credential.rsllm-router/src/types/errors.rsllm-router/src/types/events.rsllm-router/src/types/messages.rsllm-router/src/types/mod.rsllm-router/src/types/model.rsllm-router/src/types/router.rsllm-router/tests/integration.rstech-specs/2026-06-agentic/llm-router.md
Comment on lines
+28
to
+32
| pub async fn load(&self) -> Result<(), IIIError> { | ||
| let stored = state_get(&self.iii, CATALOG_KEY).await?; | ||
| *self.slices.lock().await = serde_json::from_value(stored).unwrap_or_default(); | ||
| Ok(()) | ||
| } |
There was a problem hiding this comment.
Corrupted catalog state silently replaced with empty map.
Line 30 uses unwrap_or_default() to recover from deserialization failures, which silently replaces corrupted or schema-mismatched catalog state with an empty HashMap. If the persisted catalog is corrupted, all registered models vanish on restart without error. This mirrors the same issue in registry/store.rs.
🛡️ Proposed fix to surface deserialization errors
pub async fn load(&self) -> Result<(), IIIError> {
let stored = state_get(&self.iii, CATALOG_KEY).await?;
- *self.slices.lock().await = serde_json::from_value(stored).unwrap_or_default();
+ *self.slices.lock().await = serde_json::from_value(stored).map_err(|e| {
+ eprintln!("WARNING: Failed to deserialize catalog state, starting empty: {}", e);
+ e
+ }).unwrap_or_default();
Ok(())
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| pub async fn load(&self) -> Result<(), IIIError> { | |
| let stored = state_get(&self.iii, CATALOG_KEY).await?; | |
| *self.slices.lock().await = serde_json::from_value(stored).unwrap_or_default(); | |
| Ok(()) | |
| } | |
| pub async fn load(&self) -> Result<(), IIIError> { | |
| let stored = state_get(&self.iii, CATALOG_KEY).await?; | |
| *self.slices.lock().await = serde_json::from_value(stored).map_err(|e| { | |
| eprintln!("WARNING: Failed to deserialize catalog state, starting empty: {}", e); | |
| e | |
| }).unwrap_or_default(); | |
| Ok(()) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@llm-router/src/catalog/store.rs` around lines 28 - 32, The load method
silently replaces corrupted catalog JSON by using
serde_json::from_value(...).unwrap_or_default(); change it to fail on
deserialization errors and surface the error instead of returning an empty map:
replace the unwrap_or_default usage in load (where state_get(&self.iii,
CATALOG_KEY) is deserialized into *self.slices.lock()) with proper error
propagation (e.g., map_err or ? converting serde_json::Error into IIIError) so
deserialization failures return Err(IIIError) with context rather than silently
resetting the catalog (see function load, state_get, CATALOG_KEY, and slices).
Comment on lines
+68
to
+73
| pub async fn set_slice(&self, provider: &str, models: Vec<Model>) -> Result<(), IIIError> { | ||
| let mut slices = self.slices.lock().await; // serialized writer | ||
| slices.insert(provider.to_string(), models); | ||
| let value = serde_json::to_value(&*slices).unwrap_or_default(); | ||
| state_set(&self.iii, CATALOG_KEY, value).await | ||
| } |
There was a problem hiding this comment.
Serialization failure silently writes empty/null catalog.
Line 71 uses unwrap_or_default() when serializing the catalog map, which could write Null or empty state on unexpected serialization failures instead of propagating the error. This mirrors the same issue in registry/store.rs.
🛡️ Proposed fix to propagate serialization errors
pub async fn set_slice(&self, provider: &str, models: Vec<Model>) -> Result<(), IIIError> {
let mut slices = self.slices.lock().await; // serialized writer
slices.insert(provider.to_string(), models);
- let value = serde_json::to_value(&*slices).unwrap_or_default();
- state_set(&self.iii, CATALOG_KEY, value).await
+ let value = serde_json::to_value(&*slices)
+ .map_err(|e| IIIError::Other(format!("failed to serialize catalog: {}", e)))?;
+ state_set(&self.iii, CATALOG_KEY, value).await
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@llm-router/src/catalog/store.rs` around lines 68 - 73, The set_slice function
currently calls serde_json::to_value(&*slices).unwrap_or_default(), which masks
serialization failures and can write Null/empty state; change it to propagate
the serialization error instead: replace unwrap_or_default with handling the
Result from serde_json::to_value (using ? or map_err to convert the
serde_json::Error into the function's IIIError) and only call
state_set(&self.iii, CATALOG_KEY, value).await on successful serialization,
ensuring set_slice returns Err on serialization failure; reference symbols:
set_slice, serde_json::to_value, slices, state_set, CATALOG_KEY, IIIError.
Comment on lines
+104
to
+105
| let provider = candidates[0].clone(); // MVP consumes candidates[0] | ||
| let record = self.registry.get(&provider).await.ok_or_else(|| { |
There was a problem hiding this comment.
Guard against empty routing candidates before indexing.
Line 104 assumes at least one candidate and can panic on an empty route set. Convert this to a typed router error using first() instead of direct indexing.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@llm-router/src/chat/chat.rs` around lines 104 - 105, The code assumes
candidates[0] always exists and will panic for an empty routing set; update the
selection to use candidates.first() and return a typed router error if None is
found, then use that provider value (cloned) to call
self.registry.get(&provider).await.ok_or_else(...) so indexing is avoided and an
explicit error variant is returned when there are no routing candidates.
Comment on lines
+248
to
+251
| let call_outcome = call_task | ||
| .await | ||
| .unwrap_or(Err(IIIError::Handler("provider task panicked".into()))); | ||
|
|
There was a problem hiding this comment.
Don’t block abort/caller-gone paths on provider task completion.
call_task.await happens before relay outcome handling, so RelayResult::CallerGone/Aborted still wait for provider completion/timeout. That defeats prompt cancellation and holds inflight resources longer than necessary.
Also applies to: 305-320
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@llm-router/src/chat/chat.rs` around lines 248 - 251, The current code awaits
call_task (call_task.await) before processing the relay outcome which causes
CallerGone/Aborted paths to block on provider completion; change the control
flow to handle relay outcomes immediately and not wait for the provider to
finish: use a tokio::select! (or an Abortable/JoinHandle.abort) between the
provider task (call_task) and the relay outcome so that if you observe
RelayResult::CallerGone or RelayResult::Aborted you cancel/abort the provider
task (call_task) and return early, otherwise continue to await the provider
result; adjust both the block around call_task.await at the shown location and
the analogous section covering lines 305–320 to implement this non-blocking
cancellation behavior.
Comment on lines
+57
to
+60
| return Err(RouterError::new( | ||
| RouterCode::InvalidRequest, | ||
| "stream produced no terminal frame", | ||
| ) |
There was a problem hiding this comment.
Use a server-side/router transient code for “no terminal frame”.
Line 58 labels an internal stream invariant failure as InvalidRequest, which blames the caller and can skew client retry behavior. Return a non-client-fault router code here.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@llm-router/src/chat/complete.rs` around lines 57 - 60, The error here
constructs RouterError::new with RouterCode::InvalidRequest for the case "stream
produced no terminal frame" — change that RouterCode to a
non-client-fault/server-transient code (e.g., RouterCode::Internal or
RouterCode::ServerError/Transient) so callers aren't blamed; locate the
expression RouterError::new(RouterCode::InvalidRequest, "stream produced no
terminal frame") in complete.rs and replace the RouterCode variant with the
appropriate server-side/transient RouterCode.
Comment on lines
+55
to
+60
| pub async fn load(&self) -> Result<(), IIIError> { | ||
| let stored = state_get(&self.iii, REGISTRY_KEY).await?; | ||
| let mut records = self.records.lock().await; | ||
| *records = serde_json::from_value(stored).unwrap_or_default(); | ||
| Ok(()) | ||
| } |
There was a problem hiding this comment.
Corrupted state silently replaced with empty registry.
Line 58 uses unwrap_or_default() to recover from deserialization failures, which silently replaces corrupted or schema-mismatched state with an empty HashMap. If the persisted registry state is corrupted, malformed, or uses an incompatible schema after an upgrade, all registered providers vanish on restart without error or warning.
Consider propagating deserialization errors or logging them before falling back, so operators can detect and recover from state corruption:
🛡️ Proposed fix to surface deserialization errors
pub async fn load(&self) -> Result<(), IIIError> {
let stored = state_get(&self.iii, REGISTRY_KEY).await?;
let mut records = self.records.lock().await;
- *records = serde_json::from_value(stored).unwrap_or_default();
+ *records = serde_json::from_value(stored).map_err(|e| {
+ eprintln!("WARNING: Failed to deserialize registry state, starting empty: {}", e);
+ e
+ }).unwrap_or_default();
Ok(())
}Alternatively, return the error and let the caller decide whether to proceed:
pub async fn load(&self) -> Result<(), IIIError> {
let stored = state_get(&self.iii, REGISTRY_KEY).await?;
let mut records = self.records.lock().await;
- *records = serde_json::from_value(stored).unwrap_or_default();
+ *records = serde_json::from_value(stored)
+ .map_err(|e| IIIError::Other(format!("registry state corrupted: {}", e)))?;
Ok(())
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| pub async fn load(&self) -> Result<(), IIIError> { | |
| let stored = state_get(&self.iii, REGISTRY_KEY).await?; | |
| let mut records = self.records.lock().await; | |
| *records = serde_json::from_value(stored).unwrap_or_default(); | |
| Ok(()) | |
| } | |
| pub async fn load(&self) -> Result<(), IIIError> { | |
| let stored = state_get(&self.iii, REGISTRY_KEY).await?; | |
| let mut records = self.records.lock().await; | |
| *records = serde_json::from_value(stored).map_err(|e| { | |
| eprintln!("WARNING: Failed to deserialize registry state, starting empty: {}", e); | |
| e | |
| }).unwrap_or_default(); | |
| Ok(()) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@llm-router/src/registry/store.rs` around lines 55 - 60, The load method
currently masks deserialization failures by using
serde_json::from_value(...).unwrap_or_default(), which silently drops corrupted
registry state; update load (the function named load which calls state_get and
assigns to self.records) to handle serde_json::from_value errors explicitly:
call from_value and match the Result, returning an Err(IIIError) (or mapping the
serde error into IIIError) if deserialization fails, or at minimum log the error
with context including REGISTRY_KEY before falling back; remove
unwrap_or_default and ensure records (the Mutex-protected self.records) is only
replaced on successful deserialization so corrupted/mismatched state is surfaced
to operators.
Comment on lines
+62
to
+65
| async fn persist(&self, records: &HashMap<String, ProviderRecord>) -> Result<(), IIIError> { | ||
| let value = serde_json::to_value(records).unwrap_or_default(); | ||
| state_set(&self.iii, REGISTRY_KEY, value).await | ||
| } |
There was a problem hiding this comment.
Serialization failure silently writes empty/null state.
Line 63 uses unwrap_or_default() when serializing the registry map to JSON. While HashMap<String, ProviderRecord> serialization should not normally fail if the struct is well-formed, unexpected failures (e.g., recursive structures, custom serializers with bugs, or resource exhaustion) would cause persist to write Null or an empty value instead of propagating the error. This could corrupt the persisted state without the caller's knowledge.
🛡️ Proposed fix to propagate serialization errors
async fn persist(&self, records: &HashMap<String, ProviderRecord>) -> Result<(), IIIError> {
- let value = serde_json::to_value(records).unwrap_or_default();
- state_set(&self.iii, REGISTRY_KEY, value).await
+ let value = serde_json::to_value(records)
+ .map_err(|e| IIIError::Other(format!("failed to serialize registry: {}", e)))?;
+ state_set(&self.iii, REGISTRY_KEY, value).await
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@llm-router/src/registry/store.rs` around lines 62 - 65, The persist function
currently swallows serde_json::to_value errors by calling unwrap_or_default(),
which can cause Null/empty state writes; update async fn persist(&self, records:
&HashMap<String, ProviderRecord>) -> Result<(), IIIError> to propagate
serialization errors instead of defaulting: call serde_json::to_value(records)
and return an Err converted to IIIError if serialization fails, then only call
state_set(&self.iii, REGISTRY_KEY, value).await on success (preserve
REGISTRY_KEY and state_set usage and ensure the error conversion maps the
serde_json::Error into IIIError).
Comment on lines
+36
to
+41
| let mut owners: Vec<&str> = input | ||
| .catalog | ||
| .iter() | ||
| .filter(|(_, ids)| ids.iter().any(|m| m == &input.model)) | ||
| .map(|(p, _)| p.as_str()) | ||
| .collect(); |
There was a problem hiding this comment.
Catalog routing should ignore unregistered providers.
Lines 36–41 select owners from catalog without checking registration. A stale catalog row can return a dead/unregistered provider and fail later in the chat pipeline.
Suggested change
let mut owners: Vec<&str> = input
.catalog
.iter()
+ .filter(|(provider, _)| registered(provider))
.filter(|(_, ids)| ids.iter().any(|m| m == &input.model))
.map(|(p, _)| p.as_str())
.collect();🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@llm-router/src/routing.rs` around lines 36 - 41, The owners collection
currently pulls provider names from input.catalog without checking whether those
providers are still registered; update the filter to exclude unregistered
providers by adding a check against the runtime/registry of known providers
(e.g., self.is_provider_registered(provider) or
input.registered_providers.contains(provider.as_str())) before mapping to
p.as_str(); ensure the chain that builds owners (using input.catalog and
input.model) only collects providers that pass that registration check so
stale/unregistered providers are ignored.
| if let Some(Value::Object(s)) = root.get("settings") { | ||
| out.stream_timeout_ms = pos_u64(s.get("stream_timeout_ms"), out.stream_timeout_ms); | ||
| out.idle_timeout_ms = pos_u64(s.get("idle_timeout_ms"), out.idle_timeout_ms); | ||
| out.retry_max = pos_u64(s.get("retry_max"), out.retry_max as u64) as u32; |
There was a problem hiding this comment.
Prevent silent truncation when parsing retry_max.
On Line 62, casting u64 -> u32 with as can wrap large values and produce incorrect retry behavior.
Suggested change
- out.retry_max = pos_u64(s.get("retry_max"), out.retry_max as u64) as u32;
+ let retry_raw = pos_u64(s.get("retry_max"), out.retry_max as u64);
+ out.retry_max = u32::try_from(retry_raw).unwrap_or(u32::MAX);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| out.retry_max = pos_u64(s.get("retry_max"), out.retry_max as u64) as u32; | |
| let retry_raw = pos_u64(s.get("retry_max"), out.retry_max as u64); | |
| out.retry_max = u32::try_from(retry_raw).unwrap_or(u32::MAX); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@llm-router/src/settings.rs` at line 62, The assignment to out.retry_max uses
a blind cast from u64 to u32 which can wrap large values; update the parsing to
detect values > u32::MAX and either clamp to u32::MAX or return a validation
error instead of using `as` so truncation cannot occur—modify the code around
pos_u64(s.get("retry_max"), ...) and the out.retry_max assignment to perform a
safe conversion (e.g., check the parsed u64 against u32::MAX and then set
out.retry_max = value as u32 or propagate an error), referencing the
out.retry_max field and the pos_u64 call to find the location.
Comment on lines
+3
to
+20
| #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] | ||
| #[serde(tag = "type", rename_all = "snake_case")] | ||
| pub enum Credential { | ||
| ApiKey { | ||
| key: String, | ||
| }, | ||
| Oauth { | ||
| access_token: String, | ||
| #[serde(skip_serializing_if = "Option::is_none")] | ||
| refresh_token: Option<String>, | ||
| #[serde(skip_serializing_if = "Option::is_none")] | ||
| expires_at: Option<i64>, | ||
| #[serde(skip_serializing_if = "Option::is_none")] | ||
| scopes: Option<Vec<String>>, | ||
| #[serde(skip_serializing_if = "Option::is_none")] | ||
| provider_extra: Option<serde_json::Value>, | ||
| }, | ||
| } |
There was a problem hiding this comment.
Redact secrets from Debug output on Credential.
Line 3 derives Debug for fields that carry raw API/OAuth secrets (key, access_token, refresh_token). That can leak credentials in logs and panic traces.
Suggested fix
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum Credential {
@@
}
+
+impl std::fmt::Debug for Credential {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Credential::ApiKey { .. } => f.debug_struct("ApiKey").field("key", &"<redacted>").finish(),
+ Credential::Oauth { .. } => f.debug_struct("Oauth").field("access_token", &"<redacted>").finish(),
+ }
+ }
+}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@llm-router/src/types/credential.rs` around lines 3 - 20, The derived Debug on
enum Credential exposes secrets (ApiKey.key, Oauth.access_token/refresh_token) —
replace the auto-derived Debug with a custom impl for Credential that redacts
sensitive fields: implement std::fmt::Debug for Credential and format ApiKey and
Oauth variants so that key, access_token, and refresh_token are replaced with a
fixed redact string (e.g., "<redacted>") while still displaying non-sensitive
fields (expires_at, scopes, provider_extra) for diagnostics; keep
Clone/PartialEq/Serialize/Deserialize derives unchanged and ensure the impl
covers both Credential::ApiKey { key } and Credential::Oauth { access_token,
refresh_token, expires_at, scopes, provider_extra }.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
llm-router/src/registry/store.rs (1)
161-175:⚠️ Potential issue | 🟠 Major | ⚡ Quick winRace condition: lock dropped before persist violates serialized-writer invariant.
The module doc (lines 1-2) states the Mutex must be held across mutate +
state::set, butset_availabilitydrops the lock at line 172 before callingpersistat line 173. Ifupsertruns concurrently between these lines, its registration changes will be overwritten by the stale snapshot fromset_availability.Scenario:
set_availability("p1", true)locks, modifies, clones snapshot, drops lockupsert(new_provider, ...)locks, inserts, persists, releasesset_availabilitypersists its stale snapshot → new provider registration lostHold the lock across persist, consistent with
upsert:🔒 Proposed fix to maintain lock during persist
pub async fn set_availability(&self, id: &str, available: bool) -> bool { let mut records = self.records.lock().await; let Some(rec) = records.get_mut(id) else { return false; }; if rec.available == available { return false; } rec.available = available; - let snapshot = records.clone(); - drop(records); - let _ = self.persist(&snapshot).await; // best-effort persist of a flag flip + let _ = self.persist(&records).await; // best-effort persist of a flag flip true }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@llm-router/src/registry/store.rs` around lines 161 - 175, The set_availability function currently drops the records Mutex before calling self.persist, which allows concurrent upsert to interleave and be lost; change set_availability so it holds the records lock across the call to self.persist (i.e., do not drop the mutex before calling persist), creating the snapshot while holding the lock and invoking self.persist(&snapshot).await while still holding the lock to preserve the serialized-writer invariant (consistent with how upsert behaves) and prevent overwriting concurrent registrations.
🧹 Nitpick comments (1)
llm-router/tests/integration.rs (1)
126-127: 💤 Low valueConsider capturing engine output for test debugging.
Suppressing stdout/stderr with
Stdio::null()makes it harder to diagnose test failures when the engine misbehaves. Consider capturing to temp files or usingStdio::inherit()(gated by an env var likeDEBUG_ENGINE=1) for local debugging sessions.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@llm-router/tests/integration.rs` around lines 126 - 127, The test currently silences the engine by using std::process::Stdio::null() for .stdout() and .stderr(), which hampers debugging; change this to conditionally use std::process::Stdio::inherit() when an environment variable like DEBUG_ENGINE=1 is set (falling back to Stdio::null()), or alternatively open temp files (e.g. tempfile::NamedTempFile) and pass their handles to .stdout()/.stderr() so logs are captured for failing tests; update the invocation in llm-router/tests/integration.rs where .stdout(...) and .stderr(...) are set to implement this conditional behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@llm-router/src/registry/store.rs`:
- Around line 161-175: The set_availability function currently drops the records
Mutex before calling self.persist, which allows concurrent upsert to interleave
and be lost; change set_availability so it holds the records lock across the
call to self.persist (i.e., do not drop the mutex before calling persist),
creating the snapshot while holding the lock and invoking
self.persist(&snapshot).await while still holding the lock to preserve the
serialized-writer invariant (consistent with how upsert behaves) and prevent
overwriting concurrent registrations.
---
Nitpick comments:
In `@llm-router/tests/integration.rs`:
- Around line 126-127: The test currently silences the engine by using
std::process::Stdio::null() for .stdout() and .stderr(), which hampers
debugging; change this to conditionally use std::process::Stdio::inherit() when
an environment variable like DEBUG_ENGINE=1 is set (falling back to
Stdio::null()), or alternatively open temp files (e.g. tempfile::NamedTempFile)
and pass their handles to .stdout()/.stderr() so logs are captured for failing
tests; update the invocation in llm-router/tests/integration.rs where
.stdout(...) and .stderr(...) are set to implement this conditional behavior.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 3702d228-89ca-45e5-bb10-931a8f80671d
📒 Files selected for processing (6)
llm-router/README.mdllm-router/src/catalog/store.rsllm-router/src/chat/chat.rsllm-router/src/registry/store.rsllm-router/src/types/errors.rsllm-router/tests/integration.rs
🚧 Files skipped from review as they are similar to previous changes (3)
- llm-router/src/types/errors.rs
- llm-router/src/catalog/store.rs
- llm-router/src/chat/chat.rs
3 tasks
Contributor
Author
|
Heads-up: #247 ( Merge order: this PR lands first; GitHub will retarget #247 to |
This was referenced Jun 12, 2026
ytallo
force-pushed
the
feat/llm-router-rs
branch
from
June 12, 2026 09:51
7a28a81 to
def74d2
Compare
The engine validates the existing entry value against the schema on re-register; a fresh entry holds null until the operator first writes, so a strict object type rejected every provider re-registration.
The invocation path reports a missing function as 'function_not_found' (engine/src/engine/mod.rs); bare NOT_FOUND is the configuration worker's missing-entry code and must stay Coded, or missing config entries would flip provider availability.
Removes trait Bus / bus.rs, the SdkBus adapter / bus_sdk.rs, FakeBus, the scripted provider, and the ChannelFactory dyn seam. Handlers and stores now take iii_sdk::III directly (thin state.rs wrappers per binary-worker.md §7); TriggerEmitter implements the SDK TriggerHandler itself; channel plumbing is plain functions in channels.rs. Bus-shaped test coverage moves to the engine-backed tests/integration.rs (8 scenarios, self-skipping when no engine is available, storage-worker pattern); pure-logic unit tests are unchanged.
The router owned three custom trigger types, tracking subscribers and fanning out via per-subscriber iii.trigger — a reimplementation of the engine's built-in iii-pubsub worker (publish function + subscribe trigger type). Publish to the same three names as pubsub topics instead; subscribers bind trigger_type "subscribe" with config.topic. Payloads are delivered verbatim (no envelope) and failures stay isolated per subscriber, so behavior is unchanged. Delivery is now concurrent per subscriber (engine-side spawn) instead of sequential, and publishing no longer waits for subscriber completion; all emit sites are fire-and-forget so neither is observable. Subscriptions now live engine-side, removing the registration-replay dance on router restart. New env-gated integration test pins raw-payload delivery to a subscribe-bound probe function.
The local StreamChannelRef/ChannelDirection mirror in types/channel.rs existed because the old Bus seam barred types/ from importing iii_sdk; the seam is gone, leaving a duplicate type and a serde round-trip on every channel mint and open_sink. Use the SDK type everywhere and drop the conversions. ChatRequest/ProviderStreamInput lose their unused PartialEq derive (the SDK type does not implement it); wire shape is identical, as the deleted round-trip itself proved.
…s absent The registry-publish flow (and CI's interface-boot smoke) boots the worker against a bare 'workers: []' engine that has no iii-state worker, so the boot-time state::get died with function_not_found before any function registered and interface collection timed out. Tolerate exactly that error class in the two store loads: warn and start empty. It is safe to special-case — with no state worker a later persist can't overwrite the stored snapshot either. Any other state::get failure still fails the boot. The function_not_found matcher moves from chat.rs to types/errors.rs for reuse, and a new env-gated integration test boots the router against a bare engine and asserts the read surface answers.
…able, provider guide
… failures Refactor error handling in the ChatPipeline to guarantee that a terminal error frame is sent to the sink during pre-stream failures. This change addresses issues where consumers may not receive a terminal frame, particularly when routing to an unknown provider or when invalid input is provided. Additionally, introduce a new test to validate that exactly one terminal error frame is emitted in such scenarios.
synthesize_error stamped every terminal it built as transient — a retryable kind. Correct for the mid-stream no-terminal/idle path, but the pre-stream failures (invalid request, unrouted model, unknown provider, structured-output gate) are permanent: a streaming consumer inspecting error_kind on the frame would retry requests that can never succeed. error_kind is now a caller choice; pre-stream sites pass Permanent, the mid-stream synthesis keeps Transient.
…wned provider Registration flips available back to true, but the register handler only published op:register — subscribers tracking the op:available/unavailable transitions stayed stuck on the prior unavailable. upsert now reports whether the registration recovered a downed provider (decided under the records lock) and the handler emits an explicit op:available event on that transition. Fresh registers and already-up re-registers emit nothing extra.
…rs cannot hang the caller router::complete drained its internal channel to EOF before consuming the pipeline result. A pipeline error that never wrote a frame leaves the channel without an EOF (a zero-write close does not propagate), so the drain blocked for its full 600s budget — reachable by simply killing a provider worker: dispatch fails function_not_found, run returns ProviderUnavailable, the caller times out instead of seeing the typed error. Drive the drain and the pipeline concurrently: a pipeline Err propagates to the caller immediately, for every current and future error path; on Ok the remaining in-flight frames are drained on a short budget instead of the streaming one. Engine-backed regression test registers a provider declaration with no worker behind it and asserts router::complete answers with router/provider_unavailable fast.
ytallo
force-pushed
the
feat/llm-router-rs
branch
from
June 12, 2026 13:28
def74d2 to
2731bbc
Compare
sergiofilhowz
approved these changes
Jun 12, 2026
This was referenced Jun 12, 2026
Merged
4 tasks
This was referenced Jul 3, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds
llm-router/, a standalone Rust worker that is the single front door for LLM traffic: consumers call one chat surface, provider workers plug in at runtime through a self-registration protocol, and the router owns routing, retries, streaming relay, cost fill, and credential resolution. The router never compiles against a provider; the protocol is proven against a scripted live provider in the integration suite.Consumer surface
router::chat— streaming chat over an iii channel (caller supplies awriter_ref);router::complete— non-streaming convenience that drains an internal channel;router::abort— cancels an in-flight request byrequest_id.router::models::{list,get,supports}androuter::provider::list— catalog/registry read surface. Agent exposure is restricted to the read surface viaiii-permissions.yaml.Provider protocol (token-gated)
router::provider::register— self-declaration at attach time; identity binding is a bearer registration token (the engine exposes no caller identity) and only its sha256 hash is persisted. Re-declare with the original token is idempotent; takeover without it is rejected.router::provider::resolve— per-request credential resolution with config > env > none precedence;router::provider::update_credential— OAuth token write-back;router::models::reconcile— a provider replaces its catalog slice in one write.Architecture
router::readyto reconcile with reality.llm-routerentry whose schema is composed from every registered provider's declared config schema; custom schemas must mark secret-bearing fieldswriteOnly. Config changes are fingerprint-diffed per provider slice and — debounced ~2s — kickprovider::<id>::refresh_modelsdiscovery (paste-a-key: add an API key in the config UI and models appear without a restart).router::ready,router::models::changed, androuter::provider::changedare published over the engine'siii-pubsubworker; subscribers bind thesubscribetrigger type with the topic name and receive payloads verbatim. Provider availability flips off theengine::workers-availabletopic.iii-sdkcalls (no abstraction seam), reusing SDK types (StreamChannelRef) where they exist.Also updates
tech-specs/2026-06-agentic/llm-router.md(+8 lines) to document the events as pubsub topics.Test plan
iiibinary is available: end-to-end chat relay, consumer cancellation propagating to the provider, abort terminating the stream, registry surviving a router restart with the token staying bound, token gating (takeover/resolve/reconcile), resolve precedence, paste-a-key debounced discovery, credential update round-trip, and pubsub delivery ofrouter::models::changedcargo fmt --checkandcargo clippy --workspace --all-targets -- -D warningscleanprovider-anthropic) attaching against this protocol — follow-up branchSummary by CodeRabbit
New Features
Documentation
Tests
Chores