(MOT-4348) feat(provider-deepseek): add the DeepSeek provider worker - #691
Conversation
Implements the provider protocol behind llm-router: provider::deepseek::stream, ::abort, ::refresh_models, and ::on_router_ready. Modeled on provider-zai, with the DeepSeek-specific behavior written against api-docs.deepseek.com. - Catalog: live discovery through GET /models, enriched from a local metadata table since the listing returns bare ids. Unknown ids land on conservative defaults rather than disappearing. - Reasoning: thinking + top-level reasoning_effort when a level is requested, both omitted otherwise so each model runs its documented default and an unconfigured chat streams its chain of thought into the console. - Reasoning replay: sent back as reasoning_content on tool-calling messages only, which the API requires there and ignores everywhere else. - Prompt caching: the wire layer serializes a growing session append-only so the upstream prefix cache keeps hitting; the hit slice maps to usage.cache_read and the miss slice to usage.input, so the router's additive cost fill bills each prompt token exactly once. - Blocks keep the order the model produced them, with thinking, text, and tool calls interleaved rather than merged per kind. - Errors: 402 is permanent, not a retryable rate limit; a truncated generation (insufficient_system_resource) is transient so the router retries. - Text only: images degrade to a marker with a warning instead of failing the turn on an API with no multimodal content-part array. Registers the worker in the README module table and wires it into create-tag and release.
📝 WalkthroughWalkthroughAdds a complete Rust ChangesDeepSeek provider worker
Estimated code review effort: 5 (Critical) | ~90 minutes Sequence Diagram(s)sequenceDiagram
participant Router
participant StreamWorker
participant UpstreamTransport
participant DeepSeekAPI
Router->>StreamWorker: Submit provider stream request
StreamWorker->>StreamWorker: Resolve credentials and build request
StreamWorker->>UpstreamTransport: Start upstream stream
UpstreamTransport->>DeepSeekAPI: POST chat completions
DeepSeekAPI-->>UpstreamTransport: Return SSE chunks
UpstreamTransport-->>StreamWorker: Emit assistant events
StreamWorker-->>Router: Relay stream events and completion
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
skill-check — worker0 verified, 54 skipped (no docs/).
Four for four. Nicely done. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
🧹 Nitpick comments (5)
provider-deepseek/tests/support/mod.rs (1)
55-90: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReport trailing-whitespace-only mismatches explicitly.
check_goldencompares whole strings, butdiff_hintsplits withlines(). If the only difference is a missing or extra trailing newline,first_difflands at the common length and both printed blocks are identical. The reader then sees a mismatch with no visible divergence. Add an explicit note for that case.♻️ Proposed fix
let mut out = format!( "golden mismatch: tests/golden/{rel}\n\ first divergence at line {} (expected {} lines, actual {} lines)\n", first_diff + 1, exp_lines.len(), act_lines.len() ); + if exp_lines == act_lines { + out.push_str( + "line contents are identical: the difference is trailing whitespace \ + or a trailing newline.\n", + ); + }🤖 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 `@provider-deepseek/tests/support/mod.rs` around lines 55 - 90, Update diff_hint to explicitly detect when expected and actual differ only in trailing newline or other end-of-string whitespace despite having identical lines, and append a clear note identifying the trailing-whitespace-only mismatch. Preserve the existing line-diff output for all substantive differences.provider-deepseek/tests/schemas.rs (1)
95-107: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDo not pass
no_orphan_schema_goldenswhen the golden directory is unreadable.
Err(_) => returnturns a missing or unreadabletests/golden/schemasinto a silent pass. A rename or an accidental directory removal then goes undetected, which is the case this test exists to catch. Also restrict the scan to.jsonfiles, so a nested directory or a stray file does not fail with an orphan-golden message.♻️ Proposed fix
- let entries = match std::fs::read_dir(&dir) { - Ok(e) => e, - Err(_) => return, - }; + let entries = std::fs::read_dir(&dir) + .unwrap_or_else(|e| panic!("read {}: {e}", dir.display())); for entry in entries.filter_map(Result::ok) { let name = entry.file_name().to_string_lossy().into_owned(); + if !name.ends_with(".json") { + continue; + } assert!( expected.iter().any(|e| e == &name), "orphan golden tests/golden/schemas/{name}: no catalog entry \ produces it. Delete it or fix the catalog." ); }🤖 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 `@provider-deepseek/tests/schemas.rs` around lines 95 - 107, Update no_orphan_schema_goldens so read_dir errors fail the test instead of returning silently, ensuring a missing or unreadable golden directory cannot pass. While scanning entries, inspect only regular files with a .json extension before comparing names against expected, leaving unrelated files or directories out of orphan validation.provider-deepseek/README.md (1)
32-42: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winBound the pricing claim to an effective date.
GET /modelssupplies model IDs, whilesrc/curated.rssupplies the prices. Unless that table is refreshed or versioned with every DeepSeek rate change, a released binary can report stale costs after regular or peak/off-peak pricing changes. Replace “reported cost is exact” with an effective-date or estimate statement, or add a runtime/versioned pricing source.Suggested wording
- The prices are DeepSeek's regular rates, which is what is billed today, so reported cost is exact. + The catalog prices are a checked-in snapshot. Reported cost is exact only for its documented effective date.🤖 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 `@provider-deepseek/README.md` around lines 32 - 42, Update the README pricing description near the Catalog section to avoid claiming reported costs are exact indefinitely. Describe the curated rates as estimates or state their effective date, and retain the existing explanation that catalog entries use conservative defaults and cannot represent time-of-day pricing.provider-deepseek/prompts/identity.txt (1)
289-304: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winScope the registration API rule to the selected language.
registerWorker()/registerFunction()applies only to Node/TypeScript. Python and Rust use snake_case APIs such asregister_worker(),register_function(), andregister_trigger(). Restrict this paragraph to Node/TypeScript, then have other language workers use their corresponding SDK reference.🤖 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 `@provider-deepseek/prompts/identity.txt` around lines 289 - 304, Update the worker registration guidance in the identity prompt to explicitly scope registerWorker, registerFunction, registerTrigger, and trigger to Node/TypeScript only. Direct Python and Rust workers to follow their selected SDK references and use the corresponding snake_case APIs, including register_worker, register_function, and register_trigger.provider-deepseek/src/upstream.rs (1)
44-52: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winAccept
data:without the optional space, and join multi-line data.The SSE specification makes the space after
data:optional and concatenates multipledata:lines in one event with\n.data_linerequires the space and keeps only the last line. DeepSeek itself is fine, but this provider deliberately supports OpenAI-compatible endpoints behind anapi_urloverride, and a gateway that writesdata:{…}yields zero events — the stream then ends with an emptyDone, which looks like an empty completion rather than a failure.♻️ Proposed refactor
-/// Last `data: ` payload in an SSE block, if any. Blocks that carry only +/// The `data` payload of an SSE block, if any. Multiple `data` lines join +/// with `\n` per the SSE specification, and the space after the colon is +/// optional. Blocks that carry only /// comments — DeepSeek emits `: keep-alive` while an overloaded scheduler /// makes the request wait — have none and decode to zero events. -fn data_line(block: &str) -> Option<&str> { - block - .lines() - .filter_map(|l| l.strip_prefix("data: ")) - .next_back() -} +fn data_line(block: &str) -> Option<String> { + let mut parts = block + .lines() + .filter_map(|l| l.strip_prefix("data:")) + .map(|v| v.strip_prefix(' ').unwrap_or(v)) + .peekable(); + parts.peek().is_some().then(|| parts.collect::<Vec<_>>().join("\n")) +}The call site at Line 113 then compares
data.as_str().🤖 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 `@provider-deepseek/src/upstream.rs` around lines 44 - 52, Update data_line to recognize both “data:” and “data: ” SSE fields, then collect every data field in the block and join their values with newline separators instead of returning only the last line. Preserve the existing None result for comment-only blocks, and ensure the returned value remains compatible with the call site’s data.as_str() comparison.
🤖 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 `@provider-deepseek/src/discovery.rs`:
- Around line 83-96: Normalize the credential in refresh_models using the shared
helper already used by config_from_resolve before calling credential_parts.
Treat a blank normalized result as missing, reconcile with an empty model list,
and return early; otherwise pass the normalized credential to fetch_live_models.
Add a refresh test covering a trailing-newline credential and verifying the
normalized bearer value.
- Around line 31-44: Update parse_live_models and the fetch_live_models success
path so a 2xx payload is considered valid only when it contains a data array;
return a non-success outcome for missing or malformed data instead of
reconciling an empty catalog. Preserve data: [] as the valid empty-success case,
and add a refresh test proving malformed 2xx responses leave the existing
provider catalog unchanged.
- Around line 20-26: Update models_url and its refresh_models caller so
unsupported api_url values are rejected or resolved using the configured API
origin rather than falling back to api.deepseek.com; preserve the existing
catalog when URL derivation is rejected, and update the fallback test covering
models_url to assert the new behavior.
In `@provider-deepseek/src/register.rs`:
- Around line 121-127: Update read_timeout so a parsed
PROVIDER_READ_TIMEOUT_SECS value of zero is rejected and falls back to the
existing 120-second default; preserve positive values as configured and keep the
current environment parsing flow.
In `@provider-deepseek/src/sse.rs`:
- Around line 292-306: Update the mid-stream error handling in the SSE chunk
processing path around the `if let Some(err) = chunk.get("error")` block to call
`close_open_block` before pushing the terminal `AssistantMessageEvent::Error`,
matching the `insufficient_system_resource` path. Preserve the existing error
classification and partial-content handling.
In `@provider-deepseek/src/upstream.rs`:
- Around line 77-90: Update the non-success response handling around classify
and synthetic_error_event to read the error body with a bounded size rather than
calling resp.text() unconditionally. Retain only the leading few kilobytes,
trimming to a valid UTF-8 boundary by walking backward with is_char_boundary,
and use the truncated text for classification and the emitted error message.
- Around line 133-161: After the stream ends in the while loop, before sending
AssistantMessageEvent::Done with build_final, process any remaining incomplete
SSE block left in buf. The drain_sse_blocks function only handles complete
\n\n-terminated blocks, so a connection close can leave a final data: {...} line
unprocessed. Flush this trailing block by decoding it through the same decode
flow used in drain_sse_blocks to update state with the final payload before
build_final constructs the Done message.
In `@provider-deepseek/tests/integration.rs`:
- Around line 91-126: Update spawn_engine to construct the Engine owner
immediately after spawning the child and before the readiness polling loop, so
Engine::drop handles failed readiness assertions and cleans up the process and
temporary directory. Preserve the existing URL and child ownership, cloning the
URL or reusing engine.url as needed before register_worker and the final return.
- Around line 320-326: Remove the process-global std::env::remove_var call from
provider_registers_with_persisted_token_and_credential_gated_catalog. Configure
the test’s engine or router setup through the existing explicit configuration
mechanism to pin DEEPSEEK_API_KEY as empty for this test only, preserving the
no-credential assertions without affecting parallel tests.
---
Nitpick comments:
In `@provider-deepseek/prompts/identity.txt`:
- Around line 289-304: Update the worker registration guidance in the identity
prompt to explicitly scope registerWorker, registerFunction, registerTrigger,
and trigger to Node/TypeScript only. Direct Python and Rust workers to follow
their selected SDK references and use the corresponding snake_case APIs,
including register_worker, register_function, and register_trigger.
In `@provider-deepseek/README.md`:
- Around line 32-42: Update the README pricing description near the Catalog
section to avoid claiming reported costs are exact indefinitely. Describe the
curated rates as estimates or state their effective date, and retain the
existing explanation that catalog entries use conservative defaults and cannot
represent time-of-day pricing.
In `@provider-deepseek/src/upstream.rs`:
- Around line 44-52: Update data_line to recognize both “data:” and “data: ” SSE
fields, then collect every data field in the block and join their values with
newline separators instead of returning only the last line. Preserve the
existing None result for comment-only blocks, and ensure the returned value
remains compatible with the call site’s data.as_str() comparison.
In `@provider-deepseek/tests/schemas.rs`:
- Around line 95-107: Update no_orphan_schema_goldens so read_dir errors fail
the test instead of returning silently, ensuring a missing or unreadable golden
directory cannot pass. While scanning entries, inspect only regular files with a
.json extension before comparing names against expected, leaving unrelated files
or directories out of orphan validation.
In `@provider-deepseek/tests/support/mod.rs`:
- Around line 55-90: Update diff_hint to explicitly detect when expected and
actual differ only in trailing newline or other end-of-string whitespace despite
having identical lines, and append a clear note identifying the
trailing-whitespace-only mismatch. Preserve the existing line-diff output for
all substantive differences.
🪄 Autofix
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 Plus
Run ID: 9d40156f-dfae-4eab-8022-d7c73ec88e93
⛔ Files ignored due to path filters (1)
provider-deepseek/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (38)
.github/workflows/create-tag.yml.github/workflows/release.ymlREADME.mdprovider-deepseek/.gitignoreprovider-deepseek/Cargo.tomlprovider-deepseek/README.mdprovider-deepseek/build.rsprovider-deepseek/config.yamlprovider-deepseek/iii-permissions.yamlprovider-deepseek/iii.worker.yamlprovider-deepseek/prompts/identity.txtprovider-deepseek/src/config.rsprovider-deepseek/src/curated.rsprovider-deepseek/src/discovery.rsprovider-deepseek/src/errors.rsprovider-deepseek/src/lib.rsprovider-deepseek/src/main.rsprovider-deepseek/src/manifest.rsprovider-deepseek/src/reasoning.rsprovider-deepseek/src/register.rsprovider-deepseek/src/request.rsprovider-deepseek/src/router_client.rsprovider-deepseek/src/sse.rsprovider-deepseek/src/state.rsprovider-deepseek/src/stream_fn.rsprovider-deepseek/src/surface.rsprovider-deepseek/src/upstream.rsprovider-deepseek/src/wire/messages.rsprovider-deepseek/src/wire/mod.rsprovider-deepseek/src/wire/names.rsprovider-deepseek/src/wire/tools.rsprovider-deepseek/tests/golden/schemas/provider.deepseek.abort.jsonprovider-deepseek/tests/golden/schemas/provider.deepseek.on_router_ready.jsonprovider-deepseek/tests/golden/schemas/provider.deepseek.refresh_models.jsonprovider-deepseek/tests/golden/schemas/provider.deepseek.stream.jsonprovider-deepseek/tests/integration.rsprovider-deepseek/tests/schemas.rsprovider-deepseek/tests/support/mod.rs
| pub fn models_url(api_url: &str) -> String { | ||
| api_url | ||
| .trim_end_matches('/') | ||
| .strip_suffix("/chat/completions") | ||
| .map(|base| format!("{base}/models")) | ||
| .unwrap_or_else(|| "https://api.deepseek.com/models".to_string()) | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== discovery.rs outline =="
ast-grep outline provider-deepseek/src/discovery.rs || true
echo "== discovery.rs relevant lines =="
cat -n provider-deepseek/src/discovery.rs | sed -n '1,230p'
echo "== fetch/call sites for models_url and refresh_models =="
rg -n "models_url|fetch_live_models|refresh_models|parse_live_models|reconcile" provider-deepseek/src -SRepository: iii-hq/workers
Length of output: 12736
🌐 Web query:
DeepSeek API /models endpoint credentials Bearer token origin api.deepseek.com
💡 Result:
To authenticate with the DeepSeek API, including the /models endpoint, you must use an API key obtained from the DeepSeek platform [1][2]. The authentication mechanism is a standard HTTP Bearer token [1][2]. You must include an Authorization header in every API request formatted as follows [3][2][4]: Authorization: Bearer YOUR_API_KEY Key details regarding this credential include: - Obtaining Keys: You can generate your API key by creating an account and navigating to the API Key Management page on the official DeepSeek platform (platform.deepseek.com) [5][4]. - Request Header: The Authorization header is mandatory for all endpoints at api.deepseek.com [2]. Omitting the header, using an empty value, or providing an invalid/revoked key will result in a 401 Unauthorized response [6][4]. - Compatibility: Because the DeepSeek API is designed to be compatible with the OpenAI API format, you can pass this token directly when using standard OpenAI SDKs by setting the base_url to https://api.deepseek.com and providing your key as the api_key [5][7][4]. - Security: DeepSeek API keys are sensitive strings (typically starting with sk-). They should be stored securely and never hardcoded in your source code [3][4]. If you encounter errors, a 401 status indicates an authentication issue (missing or invalid key), while a 402 status indicates that your account's balance has been exhausted [2][4].
Citations:
- 1: https://api-docs.deepseek.com/api/deepseek-api/
- 2: https://deepseekai.guide/api/deepseek-api-authentication/
- 3: https://deepseeksr1.com/api-docs/
- 4: https://deepseekai.guide/api/deepseek-api-key-setup/
- 5: https://api-docs.deepseek.com/
- 6: https://github.com/chatdeepai/deepseek-api-reproducible-tests/blob/main/api-key/TEST_PLAN.md
- 7: https://deepseekai.guide/api/deepseek-api-documentation/
Sensitive Data Exposure (CWE-201)
Reachability: Internal
Reachability path
● Entry
provider-deepseek/src/lib.rs
│
▼
● Hop
provider-deepseek/src/manifest.rs:33
json_roundtrip_has_required_fields
│
▼
● Sink
provider-deepseek/src/discovery.rs
Keep override credentials on the override origin.
models_url falls back to https://api.deepseek.com/models for unsupported api_url values, and refresh_models sends the configured bearer credential to that fallback origin. This can expose an override key to DeepSeek instead of the configured upstream and reconcile the wrong catalog.
Reject unrecognized model-generation paths or derive /models from the parsed API origin. On rejection, preserve the existing catalog and update the fallback test at provider-deepseek/src/discovery.rs:146-150.
🤖 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 `@provider-deepseek/src/discovery.rs` around lines 20 - 26, Update models_url
and its refresh_models caller so unsupported api_url values are rejected or
resolved using the configured API origin rather than falling back to
api.deepseek.com; preserve the existing catalog when URL derivation is rejected,
and update the fallback test covering models_url to assert the new behavior.
| pub fn parse_live_models(json: &Value) -> Vec<Model> { | ||
| json.get("data") | ||
| .and_then(Value::as_array) | ||
| .map(|rows| { | ||
| rows.iter() | ||
| .filter_map(|raw| { | ||
| raw.get("id") | ||
| .and_then(Value::as_str) | ||
| .filter(|s| !s.is_empty()) | ||
| }) | ||
| .map(enrich) | ||
| .collect() | ||
| }) | ||
| .unwrap_or_default() |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Do not reconcile malformed 2xx payloads as an empty catalog.
A response without a data array becomes Vec::new(). fetch_live_models treats that as success, and reconcile replaces the existing provider slice with an empty slice. A 2xx error envelope or schema change can therefore remove every catalog row.
Validate the envelope before returning FetchOutcome::Ok. Keep data: [] as the only empty-success case. Add a refresh test that verifies malformed 2xx payloads preserve the prior catalog.
Proposed fix
match resp.json::<Value>().await {
- Ok(v) => FetchOutcome::Ok(parse_live_models(&v)),
+ Ok(v) if v.get("data").and_then(Value::as_array).is_some() => {
+ FetchOutcome::Ok(parse_live_models(&v))
+ }
+ Ok(_) => FetchOutcome::Transient(
+ "models response missing a data array".to_string(),
+ ),
Err(e) => FetchOutcome::Transient(format!("models response not json: {e}")),
}Also applies to: 72-75
🤖 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 `@provider-deepseek/src/discovery.rs` around lines 31 - 44, Update
parse_live_models and the fetch_live_models success path so a 2xx payload is
considered valid only when it contains a data array; return a non-success
outcome for missing or malformed data instead of reconciling an empty catalog.
Preserve data: [] as the valid empty-success case, and add a refresh test
proving malformed 2xx responses leave the existing provider catalog unchanged.
| let Some(credential) = resolved.credential else { | ||
| // Key removed: prune the slice so the picker reflects removal | ||
| // instead of showing stale, unusable rows. | ||
| router_client::reconcile(iii, vec![], token.as_deref()).await?; | ||
| return Ok(0); | ||
| }; | ||
|
|
||
| let api_url = resolved | ||
| .api_url | ||
| .as_deref() | ||
| .map(str::trim) | ||
| .filter(|s| !s.is_empty()) | ||
| .unwrap_or(DEFAULT_API_URL); | ||
| match fetch_live_models(http, &models_url(api_url), credential_parts(&credential)).await { |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Normalize the discovery credential before constructing the header.
config_from_resolve trims credentials and rejects blank values. refresh_models sends credential_parts(&credential) without normalization. A pasted trailing newline therefore produces a different bearer value for discovery than for streaming.
Use one shared credential-normalization helper. Treat an empty normalized value as no credential before reconciliation. Add a refresh test with a trailing newline.
🤖 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 `@provider-deepseek/src/discovery.rs` around lines 83 - 96, Normalize the
credential in refresh_models using the shared helper already used by
config_from_resolve before calling credential_parts. Treat a blank normalized
result as missing, reconcile with an empty model list, and return early;
otherwise pass the normalized credential to fetch_live_models. Add a refresh
test covering a trailing-newline credential and verifying the normalized bearer
value.
| fn read_timeout() -> Duration { | ||
| std::env::var("PROVIDER_READ_TIMEOUT_SECS") | ||
| .ok() | ||
| .and_then(|s| s.parse().ok()) | ||
| .map(Duration::from_secs) | ||
| .unwrap_or(Duration::from_secs(120)) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Reject a zero value for PROVIDER_READ_TIMEOUT_SECS.
Duration::from_secs(0) is not "no timeout" for reqwest read_timeout; it bounds every read at zero and terminates each stream immediately. Operators commonly use 0 to mean "disabled". Filter it so the documented default applies.
🛡️ Proposed guard
fn read_timeout() -> Duration {
std::env::var("PROVIDER_READ_TIMEOUT_SECS")
.ok()
- .and_then(|s| s.parse().ok())
+ .and_then(|s| s.parse::<u64>().ok())
+ .filter(|secs| *secs > 0)
.map(Duration::from_secs)
.unwrap_or(Duration::from_secs(120))
}📝 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.
| fn read_timeout() -> Duration { | |
| std::env::var("PROVIDER_READ_TIMEOUT_SECS") | |
| .ok() | |
| .and_then(|s| s.parse().ok()) | |
| .map(Duration::from_secs) | |
| .unwrap_or(Duration::from_secs(120)) | |
| } | |
| fn read_timeout() -> Duration { | |
| std::env::var("PROVIDER_READ_TIMEOUT_SECS") | |
| .ok() | |
| .and_then(|s| s.parse::<u64>().ok()) | |
| .filter(|secs| *secs > 0) | |
| .map(Duration::from_secs) | |
| .unwrap_or(Duration::from_secs(120)) | |
| } |
🤖 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 `@provider-deepseek/src/register.rs` around lines 121 - 127, Update
read_timeout so a parsed PROVIDER_READ_TIMEOUT_SECS value of zero is rejected
and falls back to the existing 120-second default; preserve positive values as
configured and keep the current environment parsing flow.
| // Mid-stream error envelope (some gateways send {"error": {...}} as a | ||
| // chunk): terminal error frame carrying the partial content. | ||
| if let Some(err) = chunk.get("error") { | ||
| let msg = err | ||
| .get("message") | ||
| .and_then(Value::as_str) | ||
| .unwrap_or("upstream error") | ||
| .to_string(); | ||
| state.stop_reason = StopReason::Error; | ||
| state.error_message = Some(msg.clone()); | ||
| let mut error = build_final(state, model); | ||
| error.error_kind = Some(classify(None, &chunk.to_string())); | ||
| events.push(AssistantMessageEvent::Error { error }); | ||
| return events; | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Close the open block before the mid-stream error terminal.
The insufficient_system_resource path calls close_open_block before it pushes Error (Line 424). This path does not. A consumer that received TextStart or ThinkingStart therefore never receives the matching TextEnd/ThinkingEnd when a gateway sends an {"error": …} chunk mid-generation. The two terminal paths should emit the same block-boundary sequence.
🐛 Proposed fix
state.stop_reason = StopReason::Error;
state.error_message = Some(msg.clone());
+ close_open_block(state, model, &mut events);
let mut error = build_final(state, model);
error.error_kind = Some(classify(None, &chunk.to_string()));
events.push(AssistantMessageEvent::Error { error });
return events;📝 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.
| // Mid-stream error envelope (some gateways send {"error": {...}} as a | |
| // chunk): terminal error frame carrying the partial content. | |
| if let Some(err) = chunk.get("error") { | |
| let msg = err | |
| .get("message") | |
| .and_then(Value::as_str) | |
| .unwrap_or("upstream error") | |
| .to_string(); | |
| state.stop_reason = StopReason::Error; | |
| state.error_message = Some(msg.clone()); | |
| let mut error = build_final(state, model); | |
| error.error_kind = Some(classify(None, &chunk.to_string())); | |
| events.push(AssistantMessageEvent::Error { error }); | |
| return events; | |
| } | |
| // Mid-stream error envelope (some gateways send {"error": {...}} as a | |
| // chunk): terminal error frame carrying the partial content. | |
| if let Some(err) = chunk.get("error") { | |
| let msg = err | |
| .get("message") | |
| .and_then(Value::as_str) | |
| .unwrap_or("upstream error") | |
| .to_string(); | |
| state.stop_reason = StopReason::Error; | |
| state.error_message = Some(msg.clone()); | |
| close_open_block(state, model, &mut events); | |
| let mut error = build_final(state, model); | |
| error.error_kind = Some(classify(None, &chunk.to_string())); | |
| events.push(AssistantMessageEvent::Error { error }); | |
| return events; | |
| } |
🤖 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 `@provider-deepseek/src/sse.rs` around lines 292 - 306, Update the mid-stream
error handling in the SSE chunk processing path around the `if let Some(err) =
chunk.get("error")` block to call `close_open_block` before pushing the terminal
`AssistantMessageEvent::Error`, matching the `insufficient_system_resource`
path. Preserve the existing error classification and partial-content handling.
| let status = resp.status(); | ||
| if !status.is_success() { | ||
| let text = resp.text().await.unwrap_or_default(); | ||
| let kind = classify(Some(status.as_u16()), &text); | ||
| let msg = if text.is_empty() { | ||
| format!("deepseek http {status}") | ||
| } else { | ||
| text | ||
| }; | ||
| let _ = tx | ||
| .send(synthetic_error_event(&msg, &args.model, kind)) | ||
| .await; | ||
| return; | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Bound the error-body read.
resp.text() buffers the whole non-2xx body with no size limit. The success path streams in chunks, so it is bounded; this path is not. read_timeout bounds silence between reads, not total volume, so a broken or hostile endpoint that streams a large error body makes the worker allocate all of it. Truncate to a few kilobytes — the classifier and the operator only need the leading text.
🛡️ Proposed bound
let status = resp.status();
if !status.is_success() {
- let text = resp.text().await.unwrap_or_default();
+ const MAX_ERROR_BODY: usize = 8 * 1024;
+ let mut text = resp.text().await.unwrap_or_default();
+ if text.len() > MAX_ERROR_BODY {
+ text.truncate(text.floor_char_boundary(MAX_ERROR_BODY));
+ text.push_str("… (truncated)");
+ }
let kind = classify(Some(status.as_u16()), &text);floor_char_boundary is unstable; on stable, walk back to the nearest boundary with is_char_boundary.
🤖 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 `@provider-deepseek/src/upstream.rs` around lines 77 - 90, Update the
non-success response handling around classify and synthetic_error_event to read
the error body with a bounded size rather than calling resp.text()
unconditionally. Retain only the leading few kilobytes, trimming to a valid
UTF-8 boundary by walking backward with is_char_boundary, and use the truncated
text for classification and the emitted error message.
| while let Some(chunk) = stream.next().await { | ||
| let chunk = match chunk { | ||
| Ok(c) => c, | ||
| Err(e) => { | ||
| let _ = tx | ||
| .send(synthetic_error_event( | ||
| &format!("stream read failed: {e}"), | ||
| &args.model, | ||
| ErrorKind::Transient, | ||
| )) | ||
| .await; | ||
| return; | ||
| } | ||
| }; | ||
| append_utf8_chunk(&mut byte_buf, &mut buf, &chunk); | ||
| if drain_sse_blocks(&mut buf, &tx, &mut |block: &str| { | ||
| decode(block, &mut state, &args.model) | ||
| }) | ||
| .await | ||
| { | ||
| return; // terminal forwarded, or receiver dropped → abort upstream | ||
| } | ||
| } | ||
| // Stream ended without [DONE] (connection close framing): still terminal. | ||
| let _ = tx | ||
| .send(AssistantMessageEvent::Done { | ||
| message: build_final(&state, &args.model), | ||
| }) | ||
| .await; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect drain_sse_blocks framing and any tail-flush handling.
rg -nP -C10 '\bfn\s+drain_sse_blocks\b' --type=rs
rg -nP -C4 'drain_sse_blocks|append_utf8_chunk' --type=rs -g '!provider-deepseek/**'Repository: iii-hq/workers
Length of output: 214
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Locate drain_sse_blocks and append_utf8_chunk =="
rg -n -C10 '\b(drain_sse_blocks|append_utf8_chunk)\b' --glob '*.rs' || true
echo
echo "== Locate provider-deepseek upstream.rs relevant section =="
fd -a 'upstream\.rs' . || true
sed -n '1,220p' provider-deepseek/src/upstream.rs 2>/dev/null || trueRepository: iii-hq/workers
Length of output: 204
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Candidate upstream.rs files =="
fd 'upstream\.rs$' . || true
echo
echo "== Inspect provider-deepseek upstream.rs (if present) =="
if [ -f provider-deepseek/src/upstream.rs ]; then
sed -n '1,220p' provider-deepseek/src/upstream.rs
fi
echo
echo "== Search helpers and downstream implementations =="
for name in 'drain_sse_blocks' 'append_utf8_chunk'; do
echo "--- $name ---"
rg -n -C8 "$name" --glob '*.rs' || true
done
echo
echo "== Search for explicit tail/SSE blank-line handling =="
rg -n -C4 'DONE|sender|drop|recv|blank|tail|flush|empty_line|\\\\n\\\\n|SSE' --glob '*.rs' || trueRepository: iii-hq/workers
Length of output: 8853
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Repository files mentioning llm_router and SSE transport =="
rg -n 'llm_router|sse_transport|drain_sse_blocks|append_utf8_chunk' . || true
echo
echo "== Package definitions that may define the external crate =="
for f in Cargo.toml Cargo.lock llm-route Cargo.toml; do
[ -f "$f" ] && { echo "--- $f ---"; rg -n 'llm_router|llm-route|llm_router' "$f" -C3 || true; }
done
echo
echo "== All Rust files under llm-router if present =="
fd -a '\.rs$' llm-router 2>/dev/null | head -200 || true
echo
echo "== Search SSE transport implementations by path/name =="
rg -n 'pub\s+struct|impl|fn\s+drain|fn\s+append|event-stream|data:' llm-router 2>/dev/null || trueRepository: iii-hq/workers
Length of output: 50370
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== llm-router sse_transport implementation =="
sed -n '1,175p' llm-router/src/provider_scaffold/sse_transport.rs
echo
echo "== Any close-stream tail-flush tests in llm-router =="
sed -n '137,200p' llm-router/src/provider_scaffold/sse_transport.rsRepository: iii-hq/workers
Length of output: 6848
Flush the final SSE block before sending Done.
drain_sse_blocks only drains complete \n\n-terminated blocks, so a connection close can leave the last data: {...} line in buf. This causes Done to include the previous state without the final payload. Flush any trailing SSE block in llm-router::provider_scaffold::sse_transport before Done.
🤖 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 `@provider-deepseek/src/upstream.rs` around lines 133 - 161, After the stream
ends in the while loop, before sending AssistantMessageEvent::Done with
build_final, process any remaining incomplete SSE block left in buf. The
drain_sse_blocks function only handles complete \n\n-terminated blocks, so a
connection close can leave a final data: {...} line unprocessed. Flush this
trailing block by decoding it through the same decode flow used in
drain_sse_blocks to update state with the final payload before build_final
constructs the Done message.
| let child = std::process::Command::new(&bin) | ||
| .arg("--no-update-check") | ||
| .arg("--config") | ||
| .arg(&config_path) | ||
| .current_dir(&dir) | ||
| .stdout(std::process::Stdio::null()) | ||
| .stderr(std::process::Stdio::null()) | ||
| .spawn() | ||
| .expect("spawn engine"); | ||
|
|
||
| let url = format!("ws://127.0.0.1:{port}"); | ||
| let probe = register_worker(&url, InitOptions::default()); | ||
| let deadline = Instant::now() + Duration::from_secs(15); | ||
| loop { | ||
| let ready = probe | ||
| .trigger(TriggerRequest { | ||
| function_id: "engine::workers::list".into(), | ||
| payload: json!({}), | ||
| action: None, | ||
| timeout_ms: Some(1000), | ||
| }) | ||
| .await | ||
| .is_ok(); | ||
| if ready { | ||
| break; | ||
| } | ||
| assert!( | ||
| Instant::now() < deadline, | ||
| "engine did not become ready in 15s" | ||
| ); | ||
| tokio::time::sleep(Duration::from_millis(250)).await; | ||
| } | ||
| probe.shutdown(); | ||
|
|
||
| Some(Engine { url, child, dir }) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Own the child process before the readiness poll, or a failed poll leaks the engine.
spawn_engine spawns the engine at Line 91 but builds Engine only at Line 125. If the assert at Lines 117-120 fires, Engine::drop never runs. The engine process keeps running and the temp dir stays on disk. Each failing run then leaks another process and holds its port. Construct Engine first so Drop owns the child on every panic path.
🔒️ Proposed fix
let url = format!("ws://127.0.0.1:{port}");
+ // Own the child immediately: any panic below must still run Engine::drop.
+ let engine = Engine { url, child, dir };
let probe = register_worker(&url, InitOptions::default());
let deadline = Instant::now() + Duration::from_secs(15);
loop {
let ready = probe
.trigger(TriggerRequest {
function_id: "engine::workers::list".into(),
payload: json!({}),
action: None,
timeout_ms: Some(1000),
})
.await
.is_ok();
if ready {
break;
}
assert!(
Instant::now() < deadline,
"engine did not become ready in 15s"
);
tokio::time::sleep(Duration::from_millis(250)).await;
}
probe.shutdown();
- Some(Engine { url, child, dir })
+ Some(engine)url is used before the move, so clone it for the Engine field or read it back from engine.url.
🤖 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 `@provider-deepseek/tests/integration.rs` around lines 91 - 126, Update
spawn_engine to construct the Engine owner immediately after spawning the child
and before the readiness polling loop, so Engine::drop handles failed readiness
assertions and cleans up the process and temporary directory. Preserve the
existing URL and child ownership, cloning the URL or reusing engine.url as
needed before register_worker and the final return.
| #[tokio::test(flavor = "multi_thread")] | ||
| async fn provider_registers_with_persisted_token_and_credential_gated_catalog() { | ||
| // A real key exported on the host would leak into the in-process router's | ||
| // env-var fallback and defeat the no-credential assertions below. | ||
| std::env::remove_var("DEEPSEEK_API_KEY"); | ||
| let engine = engine_or_skip!(); | ||
| let (router_iii, provider_iii) = boot_stack(&engine.url).await; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm the crate edition and any other env mutation in the test suite.
fd -t f 'Cargo.toml' | xargs rg -n 'edition|rust-version'
rg -n --type=rust 'env::(set_var|remove_var)' -C2Repository: iii-hq/workers
Length of output: 2295
🏁 Script executed:
#!/bin/bash
set -u
echo "Files:"
fd -t f 'Cargo.toml|integration.rs' . | rg 'provider-deepseek|Cargo.toml$' | head -50
echo
echo "provider-deepseek Cargo manifest:"
fd -t f '^Cargo.toml$' provider-deepseek -x sh -c 'echo "--- $1"; sed -n "1,80p" "$1"' sh {}
echo
echo "Env mutation occurrences:"
rg -n --type=rust 'env::(set_var|remove_var|var|var_os|set_remove|getenv|setenv)' -C2 . || true
echo
echo "Target test excerpt:"
sed -n '300,360p' provider-deepseek/tests/integration.rs
echo
echo "Relevant provider-deepseek tests:"
rg -n --type=rust '(remove_var|DEEPSEEK_API_KEY|boot_stack|provider_registers_with_persisted_token|credential|token)' provider-deepseek -C3 || trueRepository: iii-hq/workers
Length of output: 50371
Remove the process-global env mutation from this test.
std::env::remove_var("DEEPSEEK_API_KEY") changes the whole test binary, so a parallel #[tokio::test] thread can clear the key while another test resolves credentials. Use an explicit configuration write that pins the empty-credential state just for this test.
🤖 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 `@provider-deepseek/tests/integration.rs` around lines 320 - 326, Remove the
process-global std::env::remove_var call from
provider_registers_with_persisted_token_and_credential_gated_catalog. Configure
the test’s engine or router setup through the existing explicit configuration
mechanism to pin DEEPSEEK_API_KEY as empty for this test only, preserving the
no-credential assertions without affecting parallel tests.
The provider-deepseek worker (#691) resolves its credential from the DEEPSEEK_API_KEY env on the router, but the E2E workflows only forwarded the anthropic/openai/zai secrets, so a deepseek subject or judge could never discover models in CI. Declare the optional secret in the reusable workflow, forward it from every caller (including the release pipeline's E2E gate from #692), add the provider to the Rust cache workspaces, and export it in both run steps.
) The provider-deepseek worker (#691) resolves its credential from the DEEPSEEK_API_KEY env on the router, but the E2E workflows only forwarded the anthropic/openai/zai secrets, so a deepseek subject or judge could never discover models in CI. Declare the optional secret in the reusable workflow, forward it from every caller (including the release pipeline's E2E gate from #692), add the provider to the Rust cache workspaces, and export it in both run steps.
2 participants
Adds
provider-deepseek, a DeepSeek Chat Completions provider behindllm-router. Paste a DeepSeek API key into the provider's slice of thellm-routerconfig anddeepseek-v4-pro/deepseek-v4-flashappear in the model picker with cost tracking, and the model's chain of thought streams into the console as it works.Modeled on
provider-zai; every DeepSeek-specific behavior is written against api-docs.deepseek.com (snapshot 2026-08). Registersprovider::deepseek::stream,::abort,::refresh_models, and::on_router_ready.Where it diverges from the sibling providers, and why
GET /models, so ids come from upstream and the local table only supplies what the listing lacks (limits, pricing). An unknown id lands on conservative defaults instead of vanishing, so a model DeepSeek ships tomorrow is routable today.usage.inputis the cache-miss slice. The spec calls these fields disjoint prompt-cache splits andfill_cost_usdadds them, so feeding theprompt_tokenstotal would bill the cached prefix twice. With DeepSeek's ~120x cache discount that roughly doubles reported cost on an agent loop. Note the OpenAI-family siblings do feed the total — a pre-existing over-report left untouched here rather than changing five workers in this PR.thinkingandreasoning_effortare omitted, so each model runs its documented default (V4: thinking at high effort) and an unconfigured chat renders its reasoning.disabledis never sent — the router has no off level to express, and a synthetic off-by-default blanks the console's thinking pane.reasoning_contentis replayed on tool-calling messages only: the API 400s a tool round whose intermediate reasoning was dropped, and documents it as ignored everywhere else, where resending would re-bill the whole chain as input every turn.provider-anthropicalready produces, and the one the*_start/*_endframes were already describing on the wire.402(out of balance) ispermanent, not a retryable rate limit.finish_reason: insufficient_system_resourceterminates astransientso the router retries instead of returning a silently truncated answer.Testing
cargo fmt --check,cargo clippy --all-targets --all-features -- -D warnings(zero warnings), and 85 tests green:AnyValue)/modelsand/chat/completions— no external API calls anywhereAlso verified
validate_worker.pypasses and the manifest emits cleanly.Notes for review
0.1.0; per repo convention the CI bot bumps after merge.create-tag.ymlandrelease.ymlperdocs/sops/new-worker.md§6. Not added toalpha-release.yml— the SOP's four-row table doesn't cover it and the omission blocks nothing; happy to add if you'd rather keep that list exhaustive.nextper §9.Fixes MOT-4348
Summary by CodeRabbit
New Features
Documentation