feat(consciousness): POST /messages and GET /queue endpoints (BRO-456) - #50
Merged
Merged
Conversation
…-456)
Phase 2: Non-blocking HTTP integration for consciousness sessions.
New endpoints (require ARCAN_CONSCIOUSNESS=true):
- POST /sessions/{id}/messages — lightweight push with steering mode
- GET /sessions/{id}/queue — queue introspection (mode, depth, pending)
New types: PushMessageRequest, PushMessageResponse, QueueStatusResponse
Actor additions: QueryStatus event, ActorStatus snapshot, build_status()
Handle additions: query_status() helper method
7 integration tests passing, full workspace green (1087 tests, 0 failures).
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
📝 Walkthrough🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
crates/arcand/src/consciousness.rs (1)
524-549: Consider exposing queue status errors rather than silently defaulting.
self.state.queue.status().ok()silently swallows any errors from the queue status call. While the fallback to defaults is safe, this could mask underlying issues in production. Consider logging a warning when status retrieval fails.🔧 Optional: Log queue status retrieval failures
fn build_status(&self) -> ActorStatus { - let queue_status = self.state.queue.status().ok(); + let queue_status = match self.state.queue.status() { + Ok(s) => Some(s), + Err(err) => { + tracing::warn!(%err, "failed to retrieve queue status"); + None + } + }; let pending = queue_status🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crates/arcand/src/consciousness.rs` around lines 524 - 549, In build_status, do not silently ignore errors from self.state.queue.status(); capture the Result, and when it is Err log a warning including the error details (e.g., via the crate logger or processLogger) before falling back to defaults; keep the existing ActorStatus construction (queue_depth, queue_pending, has_active_run, oldest_message_age_ms) but use the logged Result to decide fallback, referencing self.state.queue.status(), queue_status, and ActorStatus so reviewers can locate the change.crates/arcand/tests/consciousness_test.rs (1)
207-223: Consider replacing the fixed sleep with a retry loop for test stability.The 20ms sleep on line 215 assumes the actor starts within that window. On heavily loaded CI runners or slower systems, this could lead to flaky test failures. A small retry loop would be more robust.
🔧 Suggested: Use retry loop instead of fixed sleep
- // Give actor a moment to start. - tokio::time::sleep(Duration::from_millis(20)).await; - - let status = handle.query_status().await.expect("should get status"); + // Retry a few times if the actor hasn't started yet. + let mut status = None; + for _ in 0..10 { + tokio::time::sleep(Duration::from_millis(20)).await; + if let Some(s) = handle.query_status().await { + status = Some(s); + break; + } + } + let status = status.expect("should get status within retries");🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crates/arcand/tests/consciousness_test.rs` around lines 207 - 223, The fixed 20ms sleep in the test function query_status_returns_idle_mode is brittle; replace it with a small retry loop that repeatedly calls handle.query_status() (or another lightweight readiness probe on the actor created via registry.get_or_create) until it succeeds or a short max timeout/attempt count is reached, sleeping a few milliseconds between attempts; keep the eventual assertions (status.mode == "Idle", queue_depth == 0, !has_active_run) and still call registry.shutdown_all().await on exit so the test is robust on slower CI runners.crates/arcand/src/canonical.rs (1)
2740-2745: Steering mode parsing does not reject unknown values.Unknown steering values silently default to
Collectinstead of returning a 400 Bad Request. This may mask client errors where a typo like"steeer"is sent. Consider either documenting this behavior or rejecting unknown values explicitly.🔧 Option A: Reject unknown steering modes
let steering = match request.steering.as_deref() { Some("steer" | "Steer") => aios_protocol::SteeringMode::Steer, Some("interrupt" | "Interrupt") => aios_protocol::SteeringMode::Interrupt, Some("followup" | "Followup") => aios_protocol::SteeringMode::Followup, - _ => aios_protocol::SteeringMode::Collect, + Some("collect" | "Collect") | None => aios_protocol::SteeringMode::Collect, + Some(unknown) => { + return Err(( + StatusCode::BAD_REQUEST, + Json(json!({ "error": "invalid steering mode", "value": unknown })), + )); + } };🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crates/arcand/src/canonical.rs` around lines 2740 - 2745, The current match for request.steering in canonical.rs silently maps unrecognized strings to aios_protocol::SteeringMode::Collect; change this to validate the input and return a 400-style error for unknown values instead of defaulting. Update the logic around request.steering (the steering variable assignment) to explicitly match known strings ("steer"/"Steer", "interrupt"/"Interrupt", "followup"/"Followup") and, in the Some(other) branch, return an error/result indicating "invalid steering mode" (including the bad value and the list of accepted values) so callers receive a Bad Request rather than silently getting Collect. Ensure the function's return type and call sites handle the new error path.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@crates/arcand/src/canonical.rs`:
- Around line 2769-2776: The response currently hardcodes queue_position:
Some(1) for ConsciousnessAck::Accepted which is inaccurate; update the match arm
handling crate::consciousness::ConsciousnessAck::Accepted in canonical.rs so
that when queued is true the PushMessageResponse sets queue_position to None (or
simply always set queue_position: None) instead of Some(1); if you intend to
return a real position later, extend the ConsciousnessAck::Accepted variant to
carry the position and map that value into PushMessageResponse.queue_position
instead.
---
Nitpick comments:
In `@crates/arcand/src/canonical.rs`:
- Around line 2740-2745: The current match for request.steering in canonical.rs
silently maps unrecognized strings to aios_protocol::SteeringMode::Collect;
change this to validate the input and return a 400-style error for unknown
values instead of defaulting. Update the logic around request.steering (the
steering variable assignment) to explicitly match known strings
("steer"/"Steer", "interrupt"/"Interrupt", "followup"/"Followup") and, in the
Some(other) branch, return an error/result indicating "invalid steering mode"
(including the bad value and the list of accepted values) so callers receive a
Bad Request rather than silently getting Collect. Ensure the function's return
type and call sites handle the new error path.
In `@crates/arcand/src/consciousness.rs`:
- Around line 524-549: In build_status, do not silently ignore errors from
self.state.queue.status(); capture the Result, and when it is Err log a warning
including the error details (e.g., via the crate logger or processLogger) before
falling back to defaults; keep the existing ActorStatus construction
(queue_depth, queue_pending, has_active_run, oldest_message_age_ms) but use the
logged Result to decide fallback, referencing self.state.queue.status(),
queue_status, and ActorStatus so reviewers can locate the change.
In `@crates/arcand/tests/consciousness_test.rs`:
- Around line 207-223: The fixed 20ms sleep in the test function
query_status_returns_idle_mode is brittle; replace it with a small retry loop
that repeatedly calls handle.query_status() (or another lightweight readiness
probe on the actor created via registry.get_or_create) until it succeeds or a
short max timeout/attempt count is reached, sleeping a few milliseconds between
attempts; keep the eventual assertions (status.mode == "Idle", queue_depth == 0,
!has_active_run) and still call registry.shutdown_all().await on exit so the
test is robust on slower CI runners.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 2dcbb21e-6237-4415-8f52-b77361d9f7aa
📒 Files selected for processing (3)
crates/arcand/src/canonical.rscrates/arcand/src/consciousness.rscrates/arcand/tests/consciousness_test.rs
Comment on lines
+2769
to
+2776
| Ok(Ok(crate::consciousness::ConsciousnessAck::Accepted { queued })) => Ok(( | ||
| StatusCode::ACCEPTED, | ||
| Json(PushMessageResponse { | ||
| session_id: sid, | ||
| queued, | ||
| queue_position: if queued { Some(1) } else { None }, | ||
| }), | ||
| )), |
There was a problem hiding this comment.
Hardcoded queue_position: Some(1) is inaccurate.
When queued is true, the response always reports queue_position: Some(1), but this doesn't reflect the actual position in the queue. The ConsciousnessAck::Accepted variant doesn't provide the queue position. Either remove this field, make it None when queued, or extend the ConsciousnessAck to include the actual position.
🐛 Option: Return None for queue_position until accurate data is available
Ok(Ok(crate::consciousness::ConsciousnessAck::Accepted { queued })) => Ok((
StatusCode::ACCEPTED,
Json(PushMessageResponse {
session_id: sid,
queued,
- queue_position: if queued { Some(1) } else { None },
+ queue_position: None, // TODO: Extend ConsciousnessAck to include actual position
}),
)),📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| Ok(Ok(crate::consciousness::ConsciousnessAck::Accepted { queued })) => Ok(( | |
| StatusCode::ACCEPTED, | |
| Json(PushMessageResponse { | |
| session_id: sid, | |
| queued, | |
| queue_position: if queued { Some(1) } else { None }, | |
| }), | |
| )), | |
| Ok(Ok(crate::consciousness::ConsciousnessAck::Accepted { queued })) => Ok(( | |
| StatusCode::ACCEPTED, | |
| Json(PushMessageResponse { | |
| session_id: sid, | |
| queued, | |
| queue_position: None, // TODO: Extend ConsciousnessAck to include actual position | |
| }), | |
| )), |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/arcand/src/canonical.rs` around lines 2769 - 2776, The response
currently hardcodes queue_position: Some(1) for ConsciousnessAck::Accepted which
is inaccurate; update the match arm handling
crate::consciousness::ConsciousnessAck::Accepted in canonical.rs so that when
queued is true the PushMessageResponse sets queue_position to None (or simply
always set queue_position: None) instead of Some(1); if you intend to return a
real position later, extend the ConsciousnessAck::Accepted variant to carry the
position and map that value into PushMessageResponse.queue_position instead.
4 tasks
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
POST /sessions/{id}/messages— lightweight message push with steering mode (Collect/Steer/Followup/Interrupt)GET /sessions/{id}/queue— queue introspection showing actor mode, depth, pending messagesQueryStatusevent to consciousness actor withActorStatussnapshot responsequery_status()helper onConsciousnessHandleARCAN_CONSCIOUSNESS=trueand return 404 when disabledNew types
PushMessageRequest,PushMessageResponse,QueueStatusResponse,QueuePendingEntryActorStatus,PendingMessage(in consciousness.rs)Test plan
cargo clippy --workspace— zero warningsARCAN_CONSCIOUSNESS=true cargo run -p arcan+ curl POST /messages + GET /queueGenerated with Claude Code
Summary by CodeRabbit
POST /sessions/{session_id}/messagesendpoint to submit messages to consciousness sessions with queue acknowledgment and position tracking.GET /sessions/{session_id}/queueendpoint to retrieve session queue details including depth, mode, and pending message information.