feat(rig): align streaming, semantics, and worker concurrency#340
Open
vsumner wants to merge 3 commits intospacedriveapp:mainfrom
Open
feat(rig): align streaming, semantics, and worker concurrency#340vsumner wants to merge 3 commits intospacedriveapp:mainfrom
vsumner wants to merge 3 commits intospacedriveapp:mainfrom
Conversation
Contributor
WalkthroughAdds Rig Alignment: configuration, runtime hot-reload, semantic decision logic for tool_choice/output_schema, streaming support across LLM and messaging layers, per-worker tool concurrency, secret-scrubbing in deltas, five new telemetry metrics, API surface extensions for streaming events, and extensive documentation and rollout plan. Changes
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 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 |
![tembo[bot]](https://avatars.githubusercontent.com/in/1223475?s=60&v=4){kind=small}
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 15
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/api/cortex.rs (1)
147-159:⚠️ Potential issue | 🟠 MajorUpdate the shared client SSE type before shipping these new event kinds.
This stream now emits
text_delta,tool_started, andtool_completed, butinterface/src/api/client.tsstill narrowsCortexChatSSEEventtothinking | done | error. That leaves the typed client contract stale and makes exhaustiveness checks lie about what can arrive over SSE.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/api/cortex.rs` around lines 147 - 159, The client-side SSE union type CortexChatSSEEvent (in interface/src/api/client.ts) is missing the new event variants emitted by the server ("text_delta", "tool_started", "tool_completed"); update that discriminated union to include these three event kinds and their payload shapes so the typed client matches the server's CortexChatEvent stream (also verify the mapping of event names used in the SSE handler: "thinking", "text_delta", "tool_started", "tool_completed", "done", "error"). Ensure any switch/exhaustiveness checks and helper types that pattern-match on CortexChatSSEEvent are updated accordingly.
🧹 Nitpick comments (4)
src/tools.rs (1)
516-545: Derive this worker surface from tool constants, not duplicated literals.This helper is part of the concurrency safety gate, so mirroring
create_worker_tool_server()by hand is brittle. If a worker tool is renamed or added there without updating this list, concurrency decisions start evaluating the wrong surface.♻️ Suggested direction
pub fn list_worker_tool_names( browser_enabled: bool, brave_search_enabled: bool, secrets_enabled: bool, mcp_tools: &[McpToolAdapter], ) -> Vec<String> { let mut names = vec![ - "shell".to_string(), - "file".to_string(), - "exec".to_string(), - "task_update".to_string(), - "set_status".to_string(), - "read_skill".to_string(), + ShellTool::NAME.to_string(), + FileTool::NAME.to_string(), + ExecTool::NAME.to_string(), + TaskUpdateTool::NAME.to_string(), + SetStatusTool::NAME.to_string(), + ReadSkillTool::NAME.to_string(), ]; if secrets_enabled { - names.push("secret_set".to_string()); + names.push(SecretSetTool::NAME.to_string()); } if browser_enabled { - names.push("browser".to_string()); + names.push(BrowserTool::NAME.to_string()); } if brave_search_enabled { - names.push("web_search".to_string()); + names.push(WebSearchTool::NAME.to_string()); }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/tools.rs` around lines 516 - 545, The list_worker_tool_names function currently hardcodes tool name literals, which can diverge from the canonical tool registry used by create_worker_tool_server(); update list_worker_tool_names to derive its returned Vec<String> from the same constants/registry used by create_worker_tool_server() (e.g., reuse the tool name constants or a shared function/enum that exposes names) rather than duplicating literals, and remove the manual pushes for "shell","file","exec","task_update","set_status","read_skill","secret_set","browser","web_search" so the worker surface stays consistent with create_worker_tool_server() and any MCP tools are still appended by mapping mcp_tools.iter().map(|t| t.name()).src/main.rs (1)
1854-1854: Unreachableelsebranch intokio::select!.With a single unconditional branch (
response = response_rx.recv()), theelsearm never triggers—it only fires when all branches are disabled. TheNonecase on line 1700 already handles the channel-closed scenario. This is harmless dead code, but removing it reduces noise.🧹 Proposed cleanup
} } - else => break, } } });🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/main.rs` at line 1854, The else branch inside the tokio::select! that follows the single unconditional branch awaiting response_rx.recv() is unreachable and should be removed: locate the select! that waits on response = response_rx.recv() and delete the trailing else => break arm (the None handling for channel closure is already implemented elsewhere), leaving only the active recv branch to simplify the control flow.src/config/types.rs (1)
802-830: Consider consolidating the duplicated read-only tool list.
is_builtin_read_only_worker_toolanddefault_worker_read_only_tool_allowlistmaintain identical tool lists independently. If a tool is added to one but not the other, the validation logic inresolve_worker_tool_concurrency(which checks if allowlist entries are actually read-only) could silently reject valid configurations or accept invalid ones.♻️ Proposed consolidation
+const BUILTIN_READ_ONLY_WORKER_TOOLS: &[&str] = &[ + "read_skill", + "spacebot_docs", + "memory_recall", + "channel_recall", + "web_search", + "worker_inspect", + "email_search", + "config_inspect", +]; + pub fn is_builtin_read_only_worker_tool(tool_name: &str) -> bool { - matches!( - tool_name, - "read_skill" - | "spacebot_docs" - | "memory_recall" - | "channel_recall" - | "web_search" - | "worker_inspect" - | "email_search" - | "config_inspect" - ) + BUILTIN_READ_ONLY_WORKER_TOOLS.contains(&tool_name) } pub fn default_worker_read_only_tool_allowlist() -> Vec<String> { - [ - "read_skill", - "spacebot_docs", - "memory_recall", - "channel_recall", - "web_search", - "worker_inspect", - "email_search", - "config_inspect", - ] - .into_iter() - .map(str::to_string) - .collect() + BUILTIN_READ_ONLY_WORKER_TOOLS + .iter() + .map(|s| s.to_string()) + .collect() }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/config/types.rs` around lines 802 - 830, Extract the shared list of read-only tool names into a single source of truth (e.g., a const BUILTIN_READ_ONLY_TOOLS: &[&str] or a fn builtin_read_only_tools() -> &'static [&str]) and update both is_builtin_read_only_worker_tool and default_worker_read_only_tool_allowlist to use that single definition; ensure is_builtin_read_only_worker_tool uses the shared slice with matches or contains and default_worker_read_only_tool_allowlist maps the same slice to Vec<String>, so the validation in resolve_worker_tool_concurrency always consults the same canonical list.src/llm/model.rs (1)
356-360: Fallback attempts lose the new semantic labels.These derived models keep
rig_alignmentbut dropagent_id/process_type, so the new semantic-decision and drift metrics for fallback attempts will be emitted underunknown.♻️ Suggested follow-up
let mut model = SpacebotModel::make(&self.llm_manager, model_name); + if let (Some(agent_id), Some(process_type)) = (&self.agent_id, &self.process_type) { + model = model.with_context(agent_id.clone(), process_type.clone()); + } if let Some(rig_alignment) = self.rig_alignment.clone() { model = model.with_rig_alignment(rig_alignment); }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/llm/model.rs` around lines 356 - 360, The derived model created via SpacebotModel::make currently only reapplies rig_alignment (using with_rig_alignment) which drops agent_id and process_type, causing metrics to be emitted as "unknown"; update the derivation so agent_id and process_type are preserved — either extend with_rig_alignment to retain/propagate agent_id and process_type, or explicitly call the existing setters (e.g., with_agent_id, with_process_type or equivalent) on the model after make and before returning, ensuring rig_alignment, agent_id and process_type all remain set on the returned SpacebotModel.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/content/docs/`(configuration)/config.mdx:
- Around line 534-538: Clarify and tighten the paragraph about worker
concurrency by removing the repetition and making the single constraint
explicit: state that worker concurrency increases only when every tool exposed
to a request is on the allowlist and the allowlist contains only read‑only
tools; if the allowlist includes any mutable tools, or the default worker
surface exposes mutating tools (e.g., shell, file, exec, task_update,
set_status), Spacebot forces sequential execution and ignores
worker_read_only_tool_concurrency > 1. Update the text to a single concise
sentence (or two short sentences) that conveys those three points in order:
requirement (all exposed tools allowlisted), restriction (allowlist must be
read‑only), and default behavior (common mutating tools prevent concurrency).
In `@interface/src/hooks/useChannelLiveState.ts`:
- Around line 94-131: The applyOutboundMessageDelta function is appending
aggregated_text as if it were a delta which causes duplicated content; change
the update logic so that when a streamingMessageId exists you only append
event.text_delta, and if text_delta is absent but event.aggregated_text is
present you replace the stream item's content with aggregated_text (rather than
appending). Update the branch that finds streamIndex (using streamingMessageId
and timeline) to discriminate between event.text_delta and event.aggregated_text
and use append for text_delta but assignment/replacement for aggregated_text;
keep creation of a new message with assistantMessageItem(messageId,
event.agent_id, event.aggregated_text) unchanged.
In `@METRICS.md`:
- Around line 15-25: Update the "Total Cardinality" and "Feature Gate
Consistency" summary sections to include the new rig metrics and the worker
callsite described by the `spacebot_rig_*` series (specifically
`spacebot_rig_request_semantics_total`, `spacebot_rig_stream_sessions_total`,
`spacebot_rig_time_to_first_delta_ms`, `spacebot_rig_tool_concurrency_total`,
and `spacebot_rig_drift_detected_total`), ensuring the text reflects that these
metrics augment existing request/tool metrics rather than replace them and that
the cardinality and gating discussion accounts for the additional worker
callsite and its semantics/labels.
In `@README.md`:
- Line 232: The sentence about worker_read_only_tool_concurrency is awkwardly
worded with a repeated "only"; update the README sentence that currently
mentions the default worker surface and the mutating tools (`shell`, `file`,
`exec`, `task_update`, `set_status`) so it reads clearly, e.g. state that the
default surface still exposes those mutating tools and that
worker_read_only_tool_concurrency > 1 applies only to specialized worker
surfaces that expose only allowlisted read-only tools; edit the sentence
mentioning worker_read_only_tool_concurrency to use "applies only to" or
"applies to ... that expose only" to remove the extra "only" and improve
clarity.
In `@rig-spacebot-alignment-plan.md`:
- Around line 28-31: The doc uses non-hyphenated compound modifiers; update the
affected lines (including the bullet containing "real time-to-first-token" and
related user-facing phrasing) to use hyphenated compound modifiers—e.g., change
"real time-to-first-token" to "real-time-to-first-token" and make "end user
visible" or similar phrases "end-user-visible" (also apply the same hyphenation
fixes to the occurrences referenced around lines 917–920); ensure all compound
adjectives modifying nouns are consistently hyphenated throughout the listed
bullets.
In `@src/agent/channel.rs`:
- Around line 289-305: The current code in channel.rs uses
self.response_tx.try_send(OutboundResponse::StreamChunk(...)) which drops
streamed chunks on TrySendError::Full; replace the non-blocking try_send with
the async send path so chunks are not lost: call
self.response_tx.send(OutboundResponse::StreamChunk(text_delta.to_string())).await
(or await with a bounded timeout if you need backpressure protection) and handle
the SendError (channel closed) by logging via tracing::debug/error; keep using
OutboundResponse::StreamChunk and the same message text, and remove the
TrySendError::Full branch so backpressure will suspend the sender instead of
dropping data.
- Around line 356-376: The streaming fallback currently leaves final_history
empty so retries lose context; initialize final_history from the incoming
history before iterating the stream (i.e., seed final_history with the input
history's contents) so that if the stream ends without
MultiTurnStreamItem::FinalResponse.history() the recovery branch (and functions
like streaming_error_to_prompt_result / missing_stream_final_response_error)
will use the original history; locate this logic around stream_prompt,
MultiTurnStreamItem::FinalResponse handling, and the final_history variable and
replace the empty Vec::new() initialization with a clone/copy of the input
history.
In `@src/agent/cortex_chat.rs`:
- Around line 121-135: The helper persist_and_emit_cortex_chat_done currently
discards the Result from CortexChatStore::save_message and still emits
CortexChatEvent::Done; change this to handle failures: await save_message(), and
if it returns Err, log the error (e.g., tracing::error! with context including
thread_id) and/or propagate it by sending a CortexChatEvent::Error (or return
early) instead of emitting Done; only use .ok() when sending to event_tx if you
intentionally ignore send failures. Ensure you replace the two "let _ =" uses
with proper error handling around save_message and a conditional send of
CortexChatEvent::Done.
In `@src/agent/worker.rs`:
- Around line 206-215: The builtin worker surface is registering the full
mutable tool set (e.g. shell, file, exec, task_update, set_status) so
resolve_worker_tool_concurrency(...) always detects non-allowlisted mutable
tools and clamps concurrency to 1; change the builtin worker path that calls
list_worker_tool_names(...) to use a reduced read-only tool surface (or add a
flag/variant like list_worker_readonly_tool_names) when constructing
worker_tool_names for builtin workers so mutable tools are not included,
ensuring resolve_worker_tool_concurrency(...) can compute higher concurrency;
update every similar call site (including the other occurrence around the
819-870 region) to use the read-only list or pass the explicit readonly flag.
In `@src/hooks/spacebot.rs`:
- Around line 501-513: The on_text_delta handler currently calls
crate::secrets::scrub::scrub_leaks on each incoming text_delta and then uses
that scrubbed chunk as the aggregated_text; replace this with the stream-aware
scrubber so secrets split across chunk boundaries are removed and ensure you
emit a scrubbed version of the true cumulative buffer as aggregated_text in the
ProcessEvent::TextDelta; update the on_text_delta method to accept/maintain the
real aggregate (use the stream-aware scrubber API from src/secrets/scrub.rs) and
send that scrubbed aggregate in the event, and also change src/agent/channel.rs
around the existing forwarding logic (the block referenced at lines ~270-274) to
forward the actual aggregated text into this hook; make the identical change for
the other on_text_delta-like handler around lines 1177-1204.
In `@src/llm/model.rs`:
- Around line 1873-1876: The current branch that matches `if let
Value::Object(object) = schema` retains only `"$ref"` which drops local
`"$defs"` needed to resolve references; change the retention logic in that block
so the object keeps both `"$ref"` and `"$defs"` (e.g., retain keys where key ==
"$ref" || key == "$defs") so schemas like `{ "$ref": "#/$defs/Foo", "$defs": {
... } }` still have their definitions available for resolution.
- Around line 160-175: The logic currently treats any explicit tool_choice as
requiring request.tools, which wrongly rejects Some(ToolChoice::None) and
Some(ToolChoice::Auto); update the check that computes tool_choice_supported so
that it only enforces non-empty request.tools when the explicit tool_choice is a
forcing variant (i.e., variants that actually require tools), otherwise allow
empty request.tools; locate the block using tool_choice_requested,
request.tools, and rig_alignment.tool_choice_provider_allowlist and change the
empty-tools branch to only trigger for forcing ToolChoice variants (skip the
check for ToolChoice::None and ToolChoice::Auto).
- Around line 1861-1868: The schema_name function currently forwards the raw
schemars title; update it to normalize the title to OpenAI's constraints:
extract the title from output_schema as now (function schema_name), then
remove/replace any character not in [A-Za-z0-9_-] (e.g., replace runs of invalid
chars with a single underscore), truncate the resulting string to a maximum of
64 characters, and if the normalized result is empty use "response_schema" as a
fallback; ensure this normalized value is returned by schema_name so OpenAI
receives a compliant schema name.
In `@src/messaging/slack.rs`:
- Around line 59-70: ActiveStream's last_edit is seeded to Instant::now() in
StreamStart which causes the first StreamChunk to be throttled; change the
initialization and/or throttling check so the first edit is always allowed:
either initialize last_edit in StreamStart to Instant::now() -
STREAM_EDIT_INTERVAL (so the first chunk passes) or make last_edit an
Option<Instant> (None means "no prior edit") and update
should_throttle_stream_edit to return false when there is no previous edit;
ensure you update ActiveStream, StreamStart, StreamChunk handling, and the
should_throttle_stream_edit logic to use the chosen approach.
In `@src/messaging/telegram.rs`:
- Around line 355-375: The fallback branch that deletes the streaming
placeholder and calls send_formatted currently passes reply_to = None, which
loses reply threading; update the call to send_formatted(&self.bot, chat_id,
&text, None).await to pass the original reply message id instead (e.g., use
stream.message_id or message.reply_to as appropriate) so the recovery message is
posted as a reply; keep the rest of the logic (delete_message, active_messages
removal) unchanged and ensure you reference send_formatted, stream.message_id,
message.conversation_id and self.active_messages while making this change.
---
Outside diff comments:
In `@src/api/cortex.rs`:
- Around line 147-159: The client-side SSE union type CortexChatSSEEvent (in
interface/src/api/client.ts) is missing the new event variants emitted by the
server ("text_delta", "tool_started", "tool_completed"); update that
discriminated union to include these three event kinds and their payload shapes
so the typed client matches the server's CortexChatEvent stream (also verify the
mapping of event names used in the SSE handler: "thinking", "text_delta",
"tool_started", "tool_completed", "done", "error"). Ensure any
switch/exhaustiveness checks and helper types that pattern-match on
CortexChatSSEEvent are updated accordingly.
---
Nitpick comments:
In `@src/config/types.rs`:
- Around line 802-830: Extract the shared list of read-only tool names into a
single source of truth (e.g., a const BUILTIN_READ_ONLY_TOOLS: &[&str] or a fn
builtin_read_only_tools() -> &'static [&str]) and update both
is_builtin_read_only_worker_tool and default_worker_read_only_tool_allowlist to
use that single definition; ensure is_builtin_read_only_worker_tool uses the
shared slice with matches or contains and
default_worker_read_only_tool_allowlist maps the same slice to Vec<String>, so
the validation in resolve_worker_tool_concurrency always consults the same
canonical list.
In `@src/llm/model.rs`:
- Around line 356-360: The derived model created via SpacebotModel::make
currently only reapplies rig_alignment (using with_rig_alignment) which drops
agent_id and process_type, causing metrics to be emitted as "unknown"; update
the derivation so agent_id and process_type are preserved — either extend
with_rig_alignment to retain/propagate agent_id and process_type, or explicitly
call the existing setters (e.g., with_agent_id, with_process_type or equivalent)
on the model after make and before returning, ensuring rig_alignment, agent_id
and process_type all remain set on the returned SpacebotModel.
In `@src/main.rs`:
- Line 1854: The else branch inside the tokio::select! that follows the single
unconditional branch awaiting response_rx.recv() is unreachable and should be
removed: locate the select! that waits on response = response_rx.recv() and
delete the trailing else => break arm (the None handling for channel closure is
already implemented elsewhere), leaving only the active recv branch to simplify
the control flow.
In `@src/tools.rs`:
- Around line 516-545: The list_worker_tool_names function currently hardcodes
tool name literals, which can diverge from the canonical tool registry used by
create_worker_tool_server(); update list_worker_tool_names to derive its
returned Vec<String> from the same constants/registry used by
create_worker_tool_server() (e.g., reuse the tool name constants or a shared
function/enum that exposes names) rather than duplicating literals, and remove
the manual pushes for
"shell","file","exec","task_update","set_status","read_skill","secret_set","browser","web_search"
so the worker surface stays consistent with create_worker_tool_server() and any
MCP tools are still appended by mapping mcp_tools.iter().map(|t| t.name()).
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: c476898d-4733-4678-8f43-1271c9e95976
📒 Files selected for processing (33)
METRICS.mdREADME.mddocs/content/docs/(configuration)/config.mdxinterface/src/api/client.tsinterface/src/hooks/useChannelLiveState.tsinterface/tests/useChannelLiveState.test.tsrig-spacebot-alignment-plan.mdsrc/agent/branch.rssrc/agent/channel.rssrc/agent/compactor.rssrc/agent/cortex.rssrc/agent/cortex_chat.rssrc/agent/ingestion.rssrc/agent/worker.rssrc/api/agents.rssrc/api/cortex.rssrc/api/state.rssrc/api/system.rssrc/config/load.rssrc/config/runtime.rssrc/config/toml_schema.rssrc/config/types.rssrc/hooks/spacebot.rssrc/lib.rssrc/llm/anthropic.rssrc/llm/anthropic/params.rssrc/llm/model.rssrc/main.rssrc/messaging/discord.rssrc/messaging/slack.rssrc/messaging/telegram.rssrc/telemetry/registry.rssrc/tools.rs
vsumner
force-pushed
the
docs/rig-design-sync
branch
from
423f943 to
d1c2e70
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/agent/worker.rs (1)
257-304:⚠️ Potential issue | 🟠 MajorSnapshot
rig_alignmentonce before building the worker.
runtime_config.rig_alignmentis loaded twice here. A hot-reload between those calls can resolvetool_concurrencyfrom one config and buildSpacebotModelwith another, so the same worker run can mix semantics/concurrency settings and emit misleading telemetry.💡 Suggested fix
let worker_tool_names = crate::tools::list_worker_tool_names( self.browser_config.enabled, self.brave_search_key.is_some(), self.deps.runtime_config.secrets.load().as_ref().is_some(), &mcp_tools, ); - let tool_concurrency = resolve_worker_tool_concurrency( - self.deps.runtime_config.rig_alignment.load().as_ref(), - &worker_tool_names, - ); + let rig_alignment = (**self.deps.runtime_config.rig_alignment.load()).clone(); + let tool_concurrency = + resolve_worker_tool_concurrency(&rig_alignment, &worker_tool_names); @@ let model = SpacebotModel::make(&self.deps.llm_manager, &model_name) .with_context(&*self.deps.agent_id, "worker") - .with_rig_alignment((**self.deps.runtime_config.rig_alignment.load()).clone()) + .with_rig_alignment(rig_alignment.clone()) .with_routing((**routing).clone());🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/agent/worker.rs` around lines 257 - 304, Snapshot the runtime rig_alignment once into a local variable and use that single snapshot both when resolving tool_concurrency and when building the model to avoid a race between the two loads; specifically, call self.deps.runtime_config.rig_alignment.load() once into a local (cloned) value, pass that value into resolve_worker_tool_concurrency (instead of loading again) and into SpacebotModel::make/.with_rig_alignment (use the cloned content for with_rig_alignment), ensuring tool_concurrency, telemetry and model construction all use the same rig_alignment snapshot.src/main.rs (1)
1980-2143:⚠️ Potential issue | 🟠 MajorReuse this outbound router for resumed idle-worker channels too.
Only the fresh inbound path gets the new streaming state machine. The pre-created channel path at Lines 1748-1822 still uses the legacy outbound loop, so streamed replies after a restart skip
StreamStartsynthesis, buffered-final cleanup, andOutboundStreamEndfanout.Please extract this task into a shared helper and use it from both channel creation sites.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/main.rs` around lines 1980 - 2143, The outbound streaming loop (uses response_rx.recv, stream_active/stream_message/last_stream_text, pending_stream_cleanup_response, api_event_for_stream_cleanup, api_event_for_outbound_response, messaging_for_outbound, outbound_message, api_event_tx, outbound_conversation_id, outbound_terminal_reason, should_end_stream_before_terminal_response) is duplicated: extract it into a shared helper (e.g., spawn_outbound_router or run_outbound_router) that accepts the channel-specific handles (response_rx, outbound_message, messaging_for_outbound, api_event_tx, sse_agent_id, sse_channel_id, outbound_conversation_id, etc.) and encapsulates the tokio::select loop and all stream state and cleanup logic, then replace both the fresh inbound creation site and the pre-created/resumed idle-worker channel site to call that helper so resumed channels get the same StreamStart/StreamChunk/StreamEnd and cleanup behavior.
♻️ Duplicate comments (3)
METRICS.md (1)
15-25:⚠️ Potential issue | 🟡 MinorUpdate the downstream summary sections to cover the new rig series.
These additions document the
spacebot_rig_*metrics, butTotal CardinalityandFeature Gate Consistencybelow still read like the pre-rig inventory. That leaves this reference internally inconsistent and understates the added telemetry surface for operators sizing or auditing metrics.Also applies to: 75-117, 179-189
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@METRICS.md` around lines 15 - 25, Update the downstream summary sections "Total Cardinality" and "Feature Gate Consistency" to include the new spacebot_rig_* metrics (e.g., spacebot_rig_request_semantics_total, spacebot_rig_stream_sessions_total, spacebot_rig_time_to_first_delta_ms, spacebot_rig_tool_concurrency_total, spacebot_rig_drift_detected_total) so the cardinality counts and gate-matrix reflect the added series and labels, and ensure any statements about telemetry surface, operator sizing, or audit scope mention the separate rig metrics and their native vs. synthetic streaming distinction; apply the same updates to the other referenced ranges (lines corresponding to 75-117 and 179-189) so all summaries are consistent with the new rig inventory.src/agent/cortex_chat.rs (1)
126-147:⚠️ Potential issue | 🟠 MajorDon't emit
Doneafter a failed save.
save_message()failures are only logged here, but the client still getsDone, so the live SSE view can show an assistant turn that disappears on the nextload_history(). This helper also keeps SQLite on the terminal-event hot path by awaiting the insert before sending the final event. ReturnErroron save failure, and move the write off the response path if eventual consistency is acceptable.As per coding guidelines "Implement fire-and-forget DB writes using
tokio::spawnfor conversation history, memory writes, and worker logs so users get responses immediately".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/agent/cortex_chat.rs` around lines 126 - 147, The helper persist_and_emit_cortex_chat_done currently awaits CortexChatStore::save_message and emits CortexChatEvent::Done regardless of save result; change it to return a Result and do not send Done when save_message returns Err (so the client won't see a transient assistant turn), and if eventual consistency is acceptable convert the DB write to a fire-and-forget background task (use tokio::spawn to call store.save_message without awaiting and log any error inside that task) so the final Done event (sent via event_tx.send with CortexChatEvent::Done { full_text: response }) only occurs after a successful save or is emitted immediately while the spawned task handles persistence depending on the chosen behavior; update the function signature of persist_and_emit_cortex_chat_done and its call sites accordingly.src/hooks/spacebot.rs (1)
517-518:⚠️ Potential issue | 🟠 MajorPer-delta scrubbing still leaks secrets across chunk boundaries.
This fixes whole-token matches, but a secret split across two
text_deltaevents still passes through unchanged in each chunk, so any subscriber that appends deltas can reconstruct the raw token before the scrubbed aggregate arrives. Use the stream-aware scrubber / carry-buffer path here instead of callingscrub_leaks()on each chunk independently.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/hooks/spacebot.rs` around lines 517 - 518, The current per-chunk calls to crate::secrets::scrub::scrub_leaks on text_delta and aggregated_text leak secrets that cross chunk boundaries; replace these calls with the stream-aware scrubber (the carry-buffer path) so deltas are scrubbed with state across events. Update the code that sets scrubbed_text_delta and scrubbed_aggregated_text to feed text_delta into the stream-aware scrubber API (preserving and advancing its carry buffer) rather than calling scrub_leaks() directly, ensuring the scrubber instance/state is stored and reused across deltas so cross-boundary secrets are removed before emitting chunks to subscribers.
🧹 Nitpick comments (5)
src/llm/anthropic/params.rs (2)
320-326: Consider assertingadditionalProperties: falseis set.The test verifies that
requiredis preserved correctly, but doesn't assert thatadditionalProperties: falsewas actually added by the sanitizer—the primary purpose of the function.💚 Suggested assertion addition
assert_eq!(schema["required"], serde_json::json!(["answer"])); + assert_eq!(schema["additionalProperties"], serde_json::json!(false));🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/llm/anthropic/params.rs` around lines 320 - 326, The test anthropic_schema_sanitizer_preserves_optional_fields should also assert that sanitize_anthropic_schema enforces no extra properties: after calling sanitize_anthropic_schema(&mut schema) on the schemars::schema_for!(AnthropicOptionalSchema) result, add an assertion that schema["additionalProperties"] equals serde_json::json!(false) to verify the sanitizer set additionalProperties: false.
207-238: Consider array-formtypefield edge case.The object detection logic checks for
"type": "object"as a string, but JSON Schema also permits"type": ["object", "null"](array form) for nullable objects. If such schemas are possible in your codebase, they won't receive theadditionalProperties: falsetreatment.This is a minor edge case—flag if nullable object schemas are expected.
♻️ Optional: Handle array-form type
if let Value::Object(object) = schema { - let is_object_schema = object.get("type") == Some(&Value::String("object".to_string())) - || object.contains_key("properties"); + let is_object_schema = match object.get("type") { + Some(Value::String(s)) => s == "object", + Some(Value::Array(arr)) => arr.iter().any(|v| v == &Value::String("object".to_string())), + _ => false, + } || object.contains_key("properties");🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/llm/anthropic/params.rs` around lines 207 - 238, The schema detection in sanitize_anthropic_schema only treats "type" when it's a string "object"; update the logic to also handle the JSON Schema array-form type (e.g., ["object","null"]) by checking if object.get("type") is a Value::Array containing the string "object" and treat that as an object schema so that additionalProperties is set to false; modify the is_object_schema condition inside sanitize_anthropic_schema (referencing the "type" key and the function name) to include this array-case check while leaving the recursive traversal for "$defs", "properties", "items", "anyOf", "allOf", and "oneOf" unchanged.rig-spacebot-alignment-plan.md (1)
1-24: Prefer folding this into the existing docs surfaces instead of a new top-level file.This rollout/runbook material overlaps with the
README.mdanddocs/updates in the same PR, so keeping it here creates another source of truth that can drift as the Rig knobs change.Based on learnings, "When a feature change affects user-facing configuration, behavior, or architecture, update relevant existing documentation (
README.md,docs/) in the same commit or PR; don't create new doc files for this".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rig-spacebot-alignment-plan.md` around lines 1 - 24, This new top-level file (rig-spacebot-alignment-plan.md) duplicates rollout/runbook material; remove it from the PR and fold its YAML plan_contract content (including plan_id: rig-spacebot-alignment-2026-03-05 and related owner/assumptions/handoff_state/risk_level blocks) into the existing README.md or appropriate docs/ pages instead, placing the metadata under the existing "Rollout" or "Runbook" section so there is a single source of truth and update references in the PR to point to those updated README/docs sections; ensure the content remains verbatim and delete rig-spacebot-alignment-plan.md from the commit.src/messaging/slack.rs (1)
59-64: Use a descriptive timestamp field name here.
tsis hard to scan in this file because it now sits next tothread_ts,message_ts, and the stream placeholder timestamp.✏️ Suggested rename
struct ActiveStream { - ts: String, + timestamp: String, last_edit: Instant, text: String, }As per coding guidelines, "Use non-abbreviated variable names in Rust code:
queuenotq,messagenotmsg,channelnotch; common abbreviations likeconfigare acceptable".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/messaging/slack.rs` around lines 59 - 64, Rename the non-descriptive field ts in the ActiveStream struct to a clear name (e.g., stream_ts or stream_timestamp) to match surrounding fields like thread_ts and message_ts; update the struct definition (ActiveStream { ts -> stream_ts/stream_timestamp }) and all usages of ActiveStream::ts (constructors, pattern matches, field access, tests, and any serde/derive code) to the new name to avoid ambiguity with other timestamp fields. Ensure any serialization/deserialization, Debug/Clone derives, and references in related functions are updated accordingly so the code compiles and the intent is clear.src/tools.rs (1)
512-545: Keep worker-surface classification derived from the same source as tool registration.
list_worker_tool_names()re-encodes the worker tool surface separately fromcreate_worker_tool_server(). That means the next tool add/remove here can silently change concurrency mode unless both places stay perfectly in sync. I’d extract a shared worker-surface descriptor and use it for both registration and classification.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/tools.rs` around lines 512 - 545, list_worker_tool_names currently duplicates the list of worker-exposed tools independently of create_worker_tool_server, risking drift when tools change; extract a single shared descriptor (e.g., a static/shared Vec or const WORKER_TOOL_DESCRIPTOR) that defines the worker surface and use that descriptor both in list_worker_tool_names and create_worker_tool_server (or whichever function registers tools) so the allowed tool names are derived from the same source; update list_worker_tool_names to build names from the shared descriptor (and ensure mcp_tools handling is unified) and change create_worker_tool_server to reference the same descriptor rather than hardcoding its own list.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/agent/channel.rs`:
- Around line 167-177: The current logic appends incoming chunk into
self.leak_tail and scans immediately, which only blocks the chunk that completes
a secret and allows earlier bytes to be forwarded; change to
"scrub-before-fanout" by retaining a trailing window (STREAM_LEAK_TAIL_BYTES)
and only flushing bytes that are older than that window. Concretely, in the
method that currently uses self.leak_tail.push_str(text_delta) /
trim_to_last_bytes(...) and crate::secrets::scrub::scan_for_leaks(...), treat
self.leak_tail as a pending buffer: append the incoming text to it, run
scan_for_leaks on the combined buffer, and only forward (emit) the prefix of the
buffer that is longer than STREAM_LEAK_TAIL_BYTES (leaving the last
STREAM_LEAK_TAIL_BYTES bytes unflushed for future chunks); if scan_for_leaks
finds a match in the pending buffer mark self.blocked and drop/unforward the
entire pending buffer. Ensure trim_to_last_bytes still bounds the pending buffer
to STREAM_LEAK_TAIL_BYTES to prevent unbounded growth.
In `@src/agent/worker.rs`:
- Around line 1001-1017: The test
worker_concurrency_allowlist_rejects_default_worker_surface never hits the
branch that inspects tool_names because it uses RigAlignmentConfig::default()
where configured <= 1; update the test to pass a RigAlignmentConfig with its
configured concurrency set > 1 (e.g., set configured = 2 via a non-default
constructor or by overriding the default with ..Default::default()) so
resolve_worker_tool_concurrency(&your_config, &tool_names) actually evaluates
the allowlist/rejection logic and then assert the expected result.
- Around line 937-955: The current runtime check in worker.rs that uses
rig_alignment.worker_read_only_tool_allowlist and is_read_only_worker_tool
forces sequential execution for any non-builtin name; either enforce
builtin-only at config load or trust the configured names—so pick one: (1) Fail
fast: during config parsing/validation iterate
rig_alignment.worker_read_only_tool_allowlist and call
is_builtin_read_only_worker_tool (or an equivalent builtin-set predicate) and
return a config error if any entry is not builtin, removing the runtime
warning/return in the worker concurrency gate; or (2) Trust config: change the
runtime gate to consult the active configured surface (e.g. remove
is_read_only_worker_tool usage and check membership against the configured
allowlist/active tool registry instead) so custom names can enable concurrency.
Ensure you update the use sites referencing worker_read_only_tool_allowlist,
unsafe_allowlist_entries, and the concurrency decision logic accordingly.
In `@src/llm/model.rs`:
- Around line 691-699: The current ApiType::Anthropic | ApiType::OpenAiResponses
branch calls attempt_completion() and then wraps the finished response with
stream_from_completion_response, which blocks upstream streaming and hides live
TextDelta/ToolCallDelta events; change the logic in the stream() path so that
for OpenAiResponses/Codex you either (A) call the native streaming path
(call_openai_responses or equivalent) and return its live stream directly
instead of awaiting attempt_completion(), or (B) if you must keep a synthetic
wrapper, explicitly exclude TTFT observation for stream_from_completion_response
so timing metrics aren’t recorded for synthetic streams; update the branches
that mirror this behavior (also adjust the similar blocks around the referenced
ranges) and keep references to attempt_completion,
stream_from_completion_response, call_openai_responses, and stream() when making
the change.
In `@src/main.rs`:
- Around line 2073-2080: The synthetic and real
spacebot::OutboundResponse::StreamEnd cases aren't routed through the cleanup
path, so stream_active, stream_message, and last_stream_text aren't cleared and
the UI never receives OutboundStreamEnd; update the match/handling for
OutboundResponse::StreamEnd (the branch that currently calls
messaging_for_outbound.respond(¤t_message,
spacebot::OutboundResponse::StreamEnd)) to call the same cleanup routine/path
used for RichMessage/ThreadReply termination (clearing stream_active,
stream_message, last_stream_text and emitting OutboundStreamEnd via
messaging_for_outbound.respond), and apply the same change to the other
occurrence around the second region referenced (the later matching block) so
StopTyping or channel shutdown do not trigger a duplicate cleanup.
In `@src/messaging/slack.rs`:
- Around line 1201-1230: last_edit is advanced to now before the Slack edit, so
transient failures keep the throttle active; capture the previous
stream.last_edit (e.g., let prev_last_edit = stream.last_edit) before assigning
stream.last_edit = now, perform the session.chat_update as written, and if
session.chat_update returns Err restore stream.last_edit = prev_last_edit so the
throttle window is not kept after a failed edit; reference symbols:
active_messages, stream.last_edit, should_throttle_stream_edit,
session.chat_update, SlackApiChatUpdateRequest.
---
Outside diff comments:
In `@src/agent/worker.rs`:
- Around line 257-304: Snapshot the runtime rig_alignment once into a local
variable and use that single snapshot both when resolving tool_concurrency and
when building the model to avoid a race between the two loads; specifically,
call self.deps.runtime_config.rig_alignment.load() once into a local (cloned)
value, pass that value into resolve_worker_tool_concurrency (instead of loading
again) and into SpacebotModel::make/.with_rig_alignment (use the cloned content
for with_rig_alignment), ensuring tool_concurrency, telemetry and model
construction all use the same rig_alignment snapshot.
In `@src/main.rs`:
- Around line 1980-2143: The outbound streaming loop (uses response_rx.recv,
stream_active/stream_message/last_stream_text, pending_stream_cleanup_response,
api_event_for_stream_cleanup, api_event_for_outbound_response,
messaging_for_outbound, outbound_message, api_event_tx,
outbound_conversation_id, outbound_terminal_reason,
should_end_stream_before_terminal_response) is duplicated: extract it into a
shared helper (e.g., spawn_outbound_router or run_outbound_router) that accepts
the channel-specific handles (response_rx, outbound_message,
messaging_for_outbound, api_event_tx, sse_agent_id, sse_channel_id,
outbound_conversation_id, etc.) and encapsulates the tokio::select loop and all
stream state and cleanup logic, then replace both the fresh inbound creation
site and the pre-created/resumed idle-worker channel site to call that helper so
resumed channels get the same StreamStart/StreamChunk/StreamEnd and cleanup
behavior.
---
Duplicate comments:
In `@METRICS.md`:
- Around line 15-25: Update the downstream summary sections "Total Cardinality"
and "Feature Gate Consistency" to include the new spacebot_rig_* metrics (e.g.,
spacebot_rig_request_semantics_total, spacebot_rig_stream_sessions_total,
spacebot_rig_time_to_first_delta_ms, spacebot_rig_tool_concurrency_total,
spacebot_rig_drift_detected_total) so the cardinality counts and gate-matrix
reflect the added series and labels, and ensure any statements about telemetry
surface, operator sizing, or audit scope mention the separate rig metrics and
their native vs. synthetic streaming distinction; apply the same updates to the
other referenced ranges (lines corresponding to 75-117 and 179-189) so all
summaries are consistent with the new rig inventory.
In `@src/agent/cortex_chat.rs`:
- Around line 126-147: The helper persist_and_emit_cortex_chat_done currently
awaits CortexChatStore::save_message and emits CortexChatEvent::Done regardless
of save result; change it to return a Result and do not send Done when
save_message returns Err (so the client won't see a transient assistant turn),
and if eventual consistency is acceptable convert the DB write to a
fire-and-forget background task (use tokio::spawn to call store.save_message
without awaiting and log any error inside that task) so the final Done event
(sent via event_tx.send with CortexChatEvent::Done { full_text: response }) only
occurs after a successful save or is emitted immediately while the spawned task
handles persistence depending on the chosen behavior; update the function
signature of persist_and_emit_cortex_chat_done and its call sites accordingly.
In `@src/hooks/spacebot.rs`:
- Around line 517-518: The current per-chunk calls to
crate::secrets::scrub::scrub_leaks on text_delta and aggregated_text leak
secrets that cross chunk boundaries; replace these calls with the stream-aware
scrubber (the carry-buffer path) so deltas are scrubbed with state across
events. Update the code that sets scrubbed_text_delta and
scrubbed_aggregated_text to feed text_delta into the stream-aware scrubber API
(preserving and advancing its carry buffer) rather than calling scrub_leaks()
directly, ensuring the scrubber instance/state is stored and reused across
deltas so cross-boundary secrets are removed before emitting chunks to
subscribers.
---
Nitpick comments:
In `@rig-spacebot-alignment-plan.md`:
- Around line 1-24: This new top-level file (rig-spacebot-alignment-plan.md)
duplicates rollout/runbook material; remove it from the PR and fold its YAML
plan_contract content (including plan_id: rig-spacebot-alignment-2026-03-05 and
related owner/assumptions/handoff_state/risk_level blocks) into the existing
README.md or appropriate docs/ pages instead, placing the metadata under the
existing "Rollout" or "Runbook" section so there is a single source of truth and
update references in the PR to point to those updated README/docs sections;
ensure the content remains verbatim and delete rig-spacebot-alignment-plan.md
from the commit.
In `@src/llm/anthropic/params.rs`:
- Around line 320-326: The test
anthropic_schema_sanitizer_preserves_optional_fields should also assert that
sanitize_anthropic_schema enforces no extra properties: after calling
sanitize_anthropic_schema(&mut schema) on the
schemars::schema_for!(AnthropicOptionalSchema) result, add an assertion that
schema["additionalProperties"] equals serde_json::json!(false) to verify the
sanitizer set additionalProperties: false.
- Around line 207-238: The schema detection in sanitize_anthropic_schema only
treats "type" when it's a string "object"; update the logic to also handle the
JSON Schema array-form type (e.g., ["object","null"]) by checking if
object.get("type") is a Value::Array containing the string "object" and treat
that as an object schema so that additionalProperties is set to false; modify
the is_object_schema condition inside sanitize_anthropic_schema (referencing the
"type" key and the function name) to include this array-case check while leaving
the recursive traversal for "$defs", "properties", "items", "anyOf", "allOf",
and "oneOf" unchanged.
In `@src/messaging/slack.rs`:
- Around line 59-64: Rename the non-descriptive field ts in the ActiveStream
struct to a clear name (e.g., stream_ts or stream_timestamp) to match
surrounding fields like thread_ts and message_ts; update the struct definition
(ActiveStream { ts -> stream_ts/stream_timestamp }) and all usages of
ActiveStream::ts (constructors, pattern matches, field access, tests, and any
serde/derive code) to the new name to avoid ambiguity with other timestamp
fields. Ensure any serialization/deserialization, Debug/Clone derives, and
references in related functions are updated accordingly so the code compiles and
the intent is clear.
In `@src/tools.rs`:
- Around line 512-545: list_worker_tool_names currently duplicates the list of
worker-exposed tools independently of create_worker_tool_server, risking drift
when tools change; extract a single shared descriptor (e.g., a static/shared Vec
or const WORKER_TOOL_DESCRIPTOR) that defines the worker surface and use that
descriptor both in list_worker_tool_names and create_worker_tool_server (or
whichever function registers tools) so the allowed tool names are derived from
the same source; update list_worker_tool_names to build names from the shared
descriptor (and ensure mcp_tools handling is unified) and change
create_worker_tool_server to reference the same descriptor rather than
hardcoding its own list.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 7b4c7744-0f08-4204-b3d5-ea2dafc5b244
📒 Files selected for processing (33)
METRICS.mdREADME.mddocs/content/docs/(configuration)/config.mdxinterface/src/api/client.tsinterface/src/hooks/useChannelLiveState.tsinterface/tests/useChannelLiveState.test.tsrig-spacebot-alignment-plan.mdsrc/agent/branch.rssrc/agent/channel.rssrc/agent/compactor.rssrc/agent/cortex.rssrc/agent/cortex_chat.rssrc/agent/ingestion.rssrc/agent/worker.rssrc/api/agents.rssrc/api/cortex.rssrc/api/state.rssrc/api/system.rssrc/config/load.rssrc/config/runtime.rssrc/config/toml_schema.rssrc/config/types.rssrc/hooks/spacebot.rssrc/lib.rssrc/llm/anthropic.rssrc/llm/anthropic/params.rssrc/llm/model.rssrc/main.rssrc/messaging/discord.rssrc/messaging/slack.rssrc/messaging/telegram.rssrc/telemetry/registry.rssrc/tools.rs
🚧 Files skipped from review as they are similar to previous changes (6)
- src/agent/branch.rs
- src/agent/ingestion.rs
- src/agent/compactor.rs
- src/api/system.rs
- interface/src/hooks/useChannelLiveState.ts
- src/agent/cortex.rs
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
1 participant
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
outbound_stream_endevent for empty-terminal cleanupspacebot_rig_*telemetry, update metrics/docs/config guidance, and close the rollout/runbook gaps in the Rig alignment planTesting
cargo fmt --allcargo clippy --all-targetscargo test stream_cleanup_api_event_uses_terminal_contract -- --nocapturecargo test pending_stream_cleanup_response_prefers_buffered_text_on_shutdown -- --nocapturecargo test channel_streaming_reply_terminal -- --nocapturecargo test cortex_chat_streaming_persists_final -- --nocapturecargo test text_delta_backpressure_does_not_timeout_prompt -- --nocapturecargo test adapter_streaming_throttle -- --nocapturecargo test stream_end_releases_active_messages_lock_before_delete -- --nocapturecargo test worker_concurrency_allowlist -- --nocapturecargo test worker_concurrency_reduces_wall_clock -- --nocapturecargo test rig_parity -- --nocapturecd interface && bun test tests/useChannelLiveState.test.tscd interface && bunx tsc --noEmitcd interface && bun run buildjust preflightjust gate-prNotes (Optional)
bun run buildstill emits the existing Vite chunk-size warning, but it is non-blocking and the build succeeds.