feat: add iii-agent worker - #13
Conversation
|
Warning Rate limit exceeded
Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 52 minutes and 36 seconds. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (10)
📝 WalkthroughWalkthroughA complete new Changes
Sequence DiagramsequenceDiagram
participant Client
participant Agent as iii-agent<br/>(HTTP Handler)
participant Engine as III Engine<br/>(Trigger/State)
participant LLM as Anthropic<br/>Messages API
participant Workers as III Workers
Client->>Agent: POST /chat<br/>(session_id, message)
Agent->>Engine: state::get(agent:sessions:{id})
Engine-->>Agent: conversation history
Agent->>Engine: discover_tools()
Engine->>Workers: list_functions()
Workers-->>Engine: FunctionInfo[]
Engine-->>Agent: tools[]
Agent->>LLM: LlmRequest<br/>(messages, tools, system)
LLM-->>Agent: LlmResponse<br/>(content, stop_reason)
alt Tool Uses Present
Agent->>Engine: trigger(function_id, input)<br/>×N (parallel)
Engine->>Workers: execute function
Workers-->>Engine: result
Engine-->>Agent: tool results[]
Agent->>Agent: Append results to messages
Agent->>LLM: LlmRequest<br/>(updated messages)
LLM-->>Agent: LlmResponse
else No Tool Uses
Agent->>Agent: Extract assistant text
end
Agent->>Engine: state::set(agent:sessions:{id}, updated_history)
Engine-->>Agent: ok
Agent-->>Client: 200 OK<br/>(parsed UI elements, token usage)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 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 |
Chat orchestrator for the iii console. 7 functions: - agent::chat — synchronous chat with LLM + dynamic function discovery - agent::chat_stream — streaming chat via iii Streams - agent::discover — list available functions - agent::plan — generate execution plan without executing - agent::session_create/history/cleanup — session management Dynamically discovers functions from all connected workers via list_functions(). Uses Claude Haiku for fast responses. Tool names sanitized for Anthropic API. Uses iii primitives: State (sessions, tool cache), Streams (events), Cron (cleanup), Subscribe (functions-available). 21 tests, 0 warnings.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 10
🧹 Nitpick comments (1)
agent/config.yaml (1)
1-1: Align default model with fallback default to avoid silent behavior shifts.Line 1 uses Sonnet, but fallback defaults in
agent/src/config.rsuse Haiku. When config loading fails, model behavior changes unexpectedly (quality/cost/latency). Consider keeping one canonical default across both paths.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agent/config.yaml` at line 1, The default Anthropic model in agent/config.yaml ("anthropic_model") is different from the fallback used in agent/src/config.rs, causing unexpected behavior when config loading fails; make them identical by either updating agent/config.yaml's anthropic_model value to match the fallback constant (e.g., the string in DEFAULT_ANTHROPIC_MODEL or the fallback used in load_config()/Config::default()) or by changing the fallback constant in agent/src/config.rs to the canonical model you want; ensure the same exact model identifier string is used in both places so there is one canonical default across config file and code.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@agent/README.md`:
- Around line 40-42: Replace the inline secret in the example command so users
don't paste real keys; update the README example to show setting
ANTHROPIC_API_KEY in a separate step (e.g., export ANTHROPIC_API_KEY="sk-..." or
sourcing from an env file / .env) and then running the iii-agent binary
(./target/release/iii-agent --url ws://127.0.0.1:49134 --config ./config.yaml),
or alternatively show a placeholder like ANTHROPIC_API_KEY=<REDACTED> and a
preceding export instruction; ensure the example references the same flags
(--url, --config) but removes inline secret disclosure.
In `@agent/src/config.rs`:
- Around line 51-55: The load_config function currently deserializes YAML into
AgentConfig but doesn't validate semantic bounds; after calling
serde_yaml::from_str in load_config, add explicit runtime checks on AgentConfig
fields (e.g., ensure max_tokens > 0, max_iterations > 0, non-empty/valid cron
string, and any other domain constraints on AgentConfig) and return a
descriptive Err when any check fails so startup fails fast with a clear message;
implement these checks inside load_config immediately after deserialization and
reference the AgentConfig instance to produce context-rich errors.
In `@agent/src/discovery.rs`:
- Around line 6-9: The current hardcoded INCLUDED_PREFIXES constant makes valid
worker functions undiscoverable; replace it with a configurable exclusion-based
approach: remove or deprecate INCLUDED_PREFIXES and introduce a configurable
EXCLUDED_PREFIXES (or load prefixes from config/env) so discovery only filters
known-internal namespaces instead of allowing only six prefixes; update the
discovery logic that references INCLUDED_PREFIXES to instead consult the new
EXCLUDED_PREFIXES (or config-driven list) and accept any worker whose namespace
does not match an excluded prefix, and provide a sensible default for
EXCLUDED_PREFIXES containing only internal prefixes so behavior is unchanged
unless configured.
In `@agent/src/functions/chat_stream.rs`:
- Around line 303-326: The current load_history and save_history read the
session record via state_get but save_history overwrites the entire record with
a bare array, dropping metadata like created_at; change save_history so it
preserves the stored session object shape instead of writing the array directly:
call state_get(iii, "agent:sessions", session_id) to obtain the existing record
(or create a new object if none), set its "value" field to
serde_json::to_value(messages) (or json!([]) fallback), preserve or set
created_at if missing, then call state_set(iii, "agent:sessions", session_id,
&record) to write the full object back; keep load_history reading
val.get("value") as before but do not rely on save_history producing a bare
Vec<Message>.
- Around line 39-52: The code uses payload.get("session_id") to build
stream_group which allows empty session IDs and causes all requests to share
"agent:events:"; change the logic around the session_id binding so that if
payload.get("session_id") is absent or empty you generate a unique ID (e.g.,
Uuid::new_v4().to_string()) before computing stream_group; update the variable
created at session_id and ensure stream_group = format!("agent:events:{}",
session_id) uses that generated or provided ID (references: session_id variable,
stream_group).
In `@agent/src/functions/chat.rs`:
- Around line 188-211: The current load_history and save_history replace the
entire session record with a bare Vec<Message>, which drops fields like
created_at; update them to deserialize the stored value into a SessionState
struct (e.g., struct SessionState { created_at: String, messages: Vec<Message>
}), have load_history extract and return session.messages (or empty vec if
missing), and have save_history load the existing SessionState (or create a new
one with created_at set if missing), set its messages = messages, then serialize
and persist the full SessionState via state::state_set so other fields are
preserved for session_cleanup and session_history.
In `@agent/src/functions/plan.rs`:
- Around line 41-42: The planner currently uses
discovery::build_capabilities_summary(&tools) which composes capabilities from
ToolDef.name (sanitized names like eval__metrics) causing generated
steps[*].function_id to be invalid; update the logic so
build_capabilities_summary (or the call site in the planner where tools is
passed) uses each tool's original function_id (e.g., eval::metrics) instead of
ToolDef.name, or alternatively post-process the model's returned Plan in the
function that calls discovery::discover_tools (and/or the planner function) to
normalize/translate any sanitized names back to the original function_id before
returning the plan to executors; reference ToolDef.name, ToolDef.function_id (or
the field that holds the original id), discovery::discover_tools,
discovery::build_capabilities_summary, and ensure steps[*].function_id matches
those original function_ids.
In `@agent/src/functions/session.rs`:
- Around line 73-78: The cleanup handler currently hardcodes a 24-hour TTL in
build_cleanup_handler; change build_cleanup_handler to accept the configured
session_ttl_hours (pass the value from your config/manifest into
build_cleanup_handler) and replace all literal `24` comparisons inside the
returned closure (and the other occurrences around the second block at lines
103-107) with comparisons against that injected session_ttl_hours value (use the
same time-unit semantics as the config and the same identifier
`session_ttl_hours` inside the closure) so the handler honors the configured
TTL.
In `@agent/src/llm.rs`:
- Around line 109-113: The reqwest client is created with Client::new() (in the
new(api_key: String) constructor) which leaves no timeout and can hang; replace
Client::new() with a configured client from
Client::builder().timeout(Duration::from_secs(30)) (or read timeout from
config/env) and return that as the client field in new; also add the necessary
std::time::Duration import and ensure any per-request builders (where you call
client.post/... or methods around the client) do not override this with infinite
timeouts.
- Around line 168-193: The SSE parsing closure over byte_stream (the
byte_stream.map that builds event_stream) currently parses each TCP chunk
independently and drops JSON that spans chunks; fix by implementing buffering of
partial lines inside that closure (or replace with an eventsource_stream parser)
so chunks are appended to a persistent buffer, then split on '\n' to extract
complete "data: " lines, keep any trailing partial line in the buffer for the
next chunk, parse those complete lines into StreamEvent (the
serde_json::from_str::<StreamEvent>) and surface parse errors instead of
silently ignoring them; update the closure that produces event_stream (and the
subsequent flat_map) to use this buffered approach or swap in eventsource_stream
to ensure tool calls and text_delta events aren’t lost.
---
Nitpick comments:
In `@agent/config.yaml`:
- Line 1: The default Anthropic model in agent/config.yaml ("anthropic_model")
is different from the fallback used in agent/src/config.rs, causing unexpected
behavior when config loading fails; make them identical by either updating
agent/config.yaml's anthropic_model value to match the fallback constant (e.g.,
the string in DEFAULT_ANTHROPIC_MODEL or the fallback used in
load_config()/Config::default()) or by changing the fallback constant in
agent/src/config.rs to the canonical model you want; ensure the same exact model
identifier string is used in both places so there is one canonical default
across config file and code.
🪄 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: ba16c178-3335-4cfb-94af-476c03e6b491
📒 Files selected for processing (17)
agent/Cargo.tomlagent/README.mdagent/SPEC.mdagent/build.rsagent/config.yamlagent/src/config.rsagent/src/discovery.rsagent/src/functions/chat.rsagent/src/functions/chat_stream.rsagent/src/functions/discover.rsagent/src/functions/mod.rsagent/src/functions/plan.rsagent/src/functions/session.rsagent/src/llm.rsagent/src/main.rsagent/src/manifest.rsagent/src/state.rs
| pub fn new(api_key: String) -> Self { | ||
| Self { | ||
| client: Client::new(), | ||
| api_key, | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, let's read the full llm.rs file to understand the structure
wc -l agent/src/llm.rsRepository: iii-hq/workers
Length of output: 78
🏁 Script executed:
# Read the file to see the code structure and verify the line numbers
cat -n agent/src/llm.rsRepository: iii-hq/workers
Length of output: 7722
🏁 Script executed:
# Check if reqwest is in Cargo.toml to understand version/features
grep -A 5 "reqwest" Cargo.toml | head -20Repository: iii-hq/workers
Length of output: 101
🏁 Script executed:
# Check for Cargo.toml in agent directory
find agent -name "Cargo.toml"Repository: iii-hq/workers
Length of output: 74
🏁 Script executed:
# Let's also check the repository structure to understand dependencies
ls -la | grep -E "Cargo|package"Repository: iii-hq/workers
Length of output: 40
🌐 Web query:
reqwest::Client default timeout behavior http requests
💡 Result:
The default timeout behavior for reqwest::Client (async) created via Client::new or ClientBuilder::new.build is no timeout (None). A timeout must be explicitly set using ClientBuilder::timeout or RequestBuilder::timeout. For the blocking client (reqwest::blocking::Client), the default is 30 seconds, which can be disabled by setting .timeout(None). This applies to the latest version as of 2026-04-14. The async client intentionally has no default to avoid interrupting long-running streaming responses, while users can wrap futures in tokio::time::timeout if needed. Per-request timeouts override client-level settings.
Citations:
- 1: https://docs.rs/reqwest/latest/reqwest/struct.ClientBuilder.html
- 2: https://docs.rs/reqwest/latest/reqwest/struct.Client.html
- 3: https://deepwiki.com/seanmonstar/reqwest/3.1-client-configuration
- 4: set default timeout to 30 s for async client as well seanmonstar/reqwest#2044
- 5: Mismatch between docs and actual defaults for 'timeout' seanmonstar/reqwest#983
Set explicit request timeouts for Anthropic calls.
reqwest::Client::new() has no default timeout and will hang indefinitely on network stalls or upstream delays (lines 125, 152), pinning worker resources and degrading availability.
Suggested patch
+use std::time::Duration;
+
pub struct LlmClient {
client: Client,
api_key: String,
}
@@
pub async fn send(&self, request: &LlmRequest) -> Result<LlmResponse> {
let response = self
.client
.post(ANTHROPIC_API_URL)
+ .timeout(Duration::from_secs(60))
.header("x-api-key", &self.api_key)
.header("anthropic-version", ANTHROPIC_VERSION)
.header("content-type", "application/json")
.json(request)
.send()
.await?;
@@
let response = self
.client
.post(ANTHROPIC_API_URL)
+ .timeout(Duration::from_secs(60))
.header("x-api-key", &self.api_key)
.header("anthropic-version", ANTHROPIC_VERSION)
.header("content-type", "application/json")
.json(&stream_request)
.send()
.await?;Also applies to: 122-131, 150-158
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@agent/src/llm.rs` around lines 109 - 113, The reqwest client is created with
Client::new() (in the new(api_key: String) constructor) which leaves no timeout
and can hang; replace Client::new() with a configured client from
Client::builder().timeout(Duration::from_secs(30)) (or read timeout from
config/env) and return that as the client field in new; also add the necessary
std::time::Duration import and ensure any per-request builders (where you call
client.post/... or methods around the client) do not override this with infinite
timeouts.
| let event_stream = byte_stream.map(|chunk_result| { | ||
| let chunk = chunk_result.map_err(|e| anyhow!("stream read error: {}", e))?; | ||
| let text = String::from_utf8_lossy(&chunk); | ||
| let mut events = Vec::new(); | ||
|
|
||
| for line in text.lines() { | ||
| if let Some(data) = line.strip_prefix("data: ") { | ||
| if data == "[DONE]" { | ||
| continue; | ||
| } | ||
| if let Ok(event) = serde_json::from_str::<StreamEvent>(data) { | ||
| events.push(event); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| Ok(events) | ||
| }); | ||
|
|
||
| Ok(event_stream.flat_map(|result| { | ||
| let items: Vec<Result<StreamEvent>> = match result { | ||
| Ok(events) => events.into_iter().map(Ok).collect(), | ||
| Err(e) => vec![Err(e)], | ||
| }; | ||
| futures_util::stream::iter(items) | ||
| })) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# Find the file and check its structure
find . -type f -name "llm.rs" | head -20Repository: iii-hq/workers
Length of output: 76
🏁 Script executed:
# Check the specific lines in question
if [ -f "agent/src/llm.rs" ]; then
wc -l agent/src/llm.rs
fiRepository: iii-hq/workers
Length of output: 78
🏁 Script executed:
# Read the relevant section of the file
if [ -f "agent/src/llm.rs" ]; then
sed -n '160,200p' agent/src/llm.rs
fiRepository: iii-hq/workers
Length of output: 1534
🏁 Script executed:
# Look for StreamEvent definition and context
if [ -f "agent/src/llm.rs" ]; then
head -100 agent/src/llm.rs
fiRepository: iii-hq/workers
Length of output: 2454
🏁 Script executed:
# Check Cargo.toml for dependencies, specifically eventsource_stream
find . -name "Cargo.toml" -type f | head -5Repository: iii-hq/workers
Length of output: 123
🏁 Script executed:
# Check agent/Cargo.toml for dependencies
cat agent/Cargo.tomlRepository: iii-hq/workers
Length of output: 824
🏁 Script executed:
# Search for any existing eventsource or similar crates in the codebase
rg "eventsource" --type tomlRepository: iii-hq/workers
Length of output: 40
🏁 Script executed:
# Look at the full context of the stream parsing function
sed -n '130,220p' agent/src/llm.rsRepository: iii-hq/workers
Length of output: 3002
🏁 Script executed:
# Check if there are other parts of the codebase that use this stream
rg "event_stream\|StreamEvent" agent/src/Repository: iii-hq/workers
Length of output: 40
🏁 Script executed:
# Check how the stream is consumed in the codebase
rg "send_stream" agent/src/ -A 5Repository: iii-hq/workers
Length of output: 871
🏁 Script executed:
# Look for any buffering or frame handling logic elsewhere
rg "lines()\|chunk\|frame" agent/src/ -B 2 -A 2Repository: iii-hq/workers
Length of output: 40
🏁 Script executed:
# Check what version of reqwest bytes_stream returns and if there's documentation
rg "bytes_stream" . --type rustRepository: iii-hq/workers
Length of output: 127
🏁 Script executed:
# Look for tests that might show the expected behavior
find . -name "*test*" -type f | head -20Repository: iii-hq/workers
Length of output: 97
🏁 Script executed:
# Look at the chat_stream consumer to understand impact
cat -n agent/src/functions/chat_stream.rs | head -100Repository: iii-hq/workers
Length of output: 3560
🏁 Script executed:
# Check if there are any integration tests or examples that show streaming
find . -type f -name "*.rs" | xargs grep -l "send_stream" | head -10Repository: iii-hq/workers
Length of output: 113
🏁 Script executed:
# Verify the specific concern: can JSON split across chunks?
# Look at how reqwest structures the stream
cat > /tmp/analyze_chunk_split.py << 'EOF'
# Check if a JSON event can reasonably split across TCP/HTTP chunks
# Typical chunks are 4-16KB, but SSE can have smaller message sizes
# Anthropic's streaming format is Server-Sent Events (SSE)
# Each event looks like:
# data: {"type": "content_block_delta", "delta": {"type": "text_delta", "text": "hello"}}
#
# If the text content is long, a single event can easily exceed typical frame sizes
# and get split across multiple TCP packets/chunks
print("Analysis: SSE events with long content can exceed chunk boundaries")
print("Example: A text_delta with 1000+ character text would span chunks")
print("Risk: Current code processes chunk-by-chunk, can miss split events")
EOF
python /tmp/analyze_chunk_split.pyRepository: iii-hq/workers
Length of output: 257
🏁 Script executed:
# Check if there's any buffer handling in the streaming logic
sed -n '1,220p' agent/src/llm.rs | grep -i "buffer\|accumulate\|vec\|collect"Repository: iii-hq/workers
Length of output: 518
🏁 Script executed:
# One more check: look at the actual stream consumption to see if tool_use loss would be critical
sed -n '1,150p' agent/src/functions/chat_stream.rsRepository: iii-hq/workers
Length of output: 4623
🏁 Script executed:
# Verify: can we see where tool_use events are actually processed?
sed -n '100,200p' agent/src/functions/chat_stream.rsRepository: iii-hq/workers
Length of output: 3573
🏁 Script executed:
# Check if there's any accumulation or buffering of partial data anywhere
rg "content_block\|tool_use\|delta" agent/src/functions/chat_stream.rs -B 2 -A 2Repository: iii-hq/workers
Length of output: 40
SSE stream parsing loses events when JSON frames split across chunks, silently dropping tool calls and truncating text.
The code processes each TCP chunk independently and attempts to parse complete JSON lines. When a JSON event spans multiple chunks (common with longer content), both chunks fail to parse individually—the silent if let Ok(event) discards them without error. This breaks tool invocation: collected_tool_uses will be missing entries, preventing function execution. Similarly, text content gets truncated when deltas are lost.
Example: A content_block_delta event with long text_delta can exceed typical chunk sizes and split across two TCP frames. The first frame ends mid-JSON; the second frame has only the tail. Neither parses, both drop silently.
Accumulate frames into a proper SSE parser (e.g., eventsource_stream crate) or buffer incomplete lines across chunks before parsing.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@agent/src/llm.rs` around lines 168 - 193, The SSE parsing closure over
byte_stream (the byte_stream.map that builds event_stream) currently parses each
TCP chunk independently and drops JSON that spans chunks; fix by implementing
buffering of partial lines inside that closure (or replace with an
eventsource_stream parser) so chunks are appended to a persistent buffer, then
split on '\n' to extract complete "data: " lines, keep any trailing partial line
in the buffer for the next chunk, parse those complete lines into StreamEvent
(the serde_json::from_str::<StreamEvent>) and surface parse errors instead of
silently ignoring them; update the closure that produces event_stream (and the
subsequent flat_map) to use this buffered approach or swap in eventsource_stream
to ensure tool calls and text_delta events aren’t lost.
- Drop features = ["otel"] (OTel always-on in 0.11.0) - Add metadata: None to RegisterTriggerInput literals (new required field) - 21 tests pass
|
Heads up — main is being bumped to
Release notes: https://github.com/iii-hq/iii/releases/tag/iii/v0.11.3 |
…y allowlist, planner ids
- chat.rs + chat_stream.rs: preserve the {created_at, messages} envelope
instead of overwriting the session record with a bare messages array.
save_history now reads the existing created_at (or mints a fresh RFC3339
one) and writes the full envelope; load_history unwraps either shape
(envelope or legacy bare array) so old sessions still open.
- chat_stream.rs: mint a fresh uuid session_id when the caller omits one
so concurrent session-less requests don't collide on the shared
agent:events: stream group.
- config.rs: validate() after deserialization — reject empty
anthropic_model / cron string and zero max_tokens / max_iterations /
session_ttl_hours at startup instead of breaking behaviour silently.
- session.rs: build_cleanup_handler now takes ttl_hours and honours
config.session_ttl_hours instead of the hardcoded 24h.
- discovery.rs: flip from a hardcoded include-list of six worker
namespaces to an exclude-list of infrastructure prefixes (agent::,
engine::, state::, stream::, iii.). New worker namespaces are
auto-discoverable without a code change. Add discover_tools_with(iii,
excluded) + DEFAULT_EXCLUDED_PREFIXES so deployments can tighten.
- discovery.rs: add build_planner_capabilities(iii) that emits engine
function_ids (eval::metrics) instead of sanitized tool names
(eval__metrics). plan.rs uses it, so the LLM fills steps[*].function_id
with ids downstream executors can actually invoke.
- README.md: replace 'ANTHROPIC_API_KEY=sk-ant-... ./iii-agent ...' with
an export loaded from the keychain so the literal key doesn't land in
shell history or ps output.
1 participant
Summary
Functions
agent::chatagent::chat_streamagent::discoveragent::planagent::session_createagent::session_historyagent::session_cleanupTest plan
cargo test— 21 tests passingcargo check— 0 warningsSummary by CodeRabbit
Release Notes
New Features
Documentation
Configuration