diff --git a/crates/arcand/src/canonical.rs b/crates/arcand/src/canonical.rs index 8cfee77..8e3507b 100644 --- a/crates/arcand/src/canonical.rs +++ b/crates/arcand/src/canonical.rs @@ -359,6 +359,48 @@ struct ProviderResponse { available: Vec, } +// ─── Consciousness endpoints (BRO-456) ────────────────────────────────────── + +/// Request body for POST /sessions/{id}/messages — lightweight message push. +#[derive(Debug, Deserialize, ToSchema)] +struct PushMessageRequest { + /// The message content to send. + message: String, + /// How this message should interact with an active run (default: Collect). + steering: Option, +} + +/// Response for POST /sessions/{id}/messages. +#[derive(Debug, Serialize, ToSchema)] +struct PushMessageResponse { + #[schema(value_type = String)] + session_id: SessionId, + queued: bool, + queue_position: Option, +} + +/// Response for GET /sessions/{id}/queue. +#[derive(Debug, Serialize, ToSchema)] +struct QueueStatusResponse { + #[schema(value_type = String)] + session_id: SessionId, + mode: String, + queue_depth: usize, + has_active_run: bool, + oldest_message_age_ms: Option, + pending: Vec, +} + +/// A single pending entry in the queue response. +#[derive(Debug, Serialize, ToSchema)] +struct QueuePendingEntry { + id: String, + steering_mode: String, + content: String, +} + +// ─── End consciousness types ───────────────────────────────────────────────── + #[derive(Debug, Serialize, ToSchema)] struct RunResponse { #[schema(value_type = String, example = "sess-abc123")] @@ -452,6 +494,8 @@ struct ErrorResponse { get_context, get_cost, list_mcp_servers, + push_message, + get_queue_status, ), components(schemas( // Request / response types @@ -477,6 +521,11 @@ struct ErrorResponse { AutonomicResponse, ContextResponse, CostResponse, + // Consciousness endpoint types (BRO-456) + PushMessageRequest, + PushMessageResponse, + QueueStatusResponse, + QueuePendingEntry, // Mirror schemas for external aios-protocol types OperatingModeSchema, AgentStateVectorSchema, @@ -499,6 +548,7 @@ struct ErrorResponse { (name = "provider", description = "Live provider switching"), (name = "autonomic", description = "Autonomic context regulation and homeostatic state"), (name = "mcp", description = "MCP server registry"), + (name = "consciousness", description = "Consciousness session management and message queue"), ) )] struct ApiDoc; @@ -740,6 +790,15 @@ pub fn create_canonical_router_with_skills( "/sessions/{session_id}/approvals/{approval_id}", post(resolve_approval), ) + // BRO-456: Consciousness message push and queue introspection. + .route( + "/sessions/{session_id}/messages", + post(push_message), + ) + .route( + "/sessions/{session_id}/queue", + get(get_queue_status), + ) .route("/provider", get(get_provider).put(set_provider)) .route("/autonomic", get(get_autonomic)) .route("/context", get(get_context)) @@ -2641,6 +2700,141 @@ async fn get_cost(State(state): State) -> Json { }) } +// ─── Consciousness endpoints (BRO-456) ────────────────────────────────────── + +/// Push a message to an active consciousness session. +/// +/// Lightweight alternative to `POST /runs` for follow-ups, interrupts, or steers. +/// Requires consciousness mode (`ARCAN_CONSCIOUSNESS=true`). +#[utoipa::path( + post, + path = "/sessions/{session_id}/messages", + tag = "consciousness", + params(("session_id" = String, Path, description = "Session identifier")), + request_body = PushMessageRequest, + responses( + (status = 202, description = "Message accepted", body = PushMessageResponse), + (status = 404, description = "Session not found or consciousness disabled", body = ErrorResponse), + (status = 503, description = "Message rejected", body = ErrorResponse) + ) +)] +async fn push_message( + Path(session_id): Path, + State(state): State, + Json(request): Json, +) -> Result<(StatusCode, Json), (StatusCode, Json)> { + let registry = state.consciousness_registry.as_ref().ok_or_else(|| { + ( + StatusCode::NOT_FOUND, + Json(json!({ "error": "consciousness mode is not enabled" })), + ) + })?; + + let handle = registry.get(&session_id).ok_or_else(|| { + ( + StatusCode::NOT_FOUND, + Json(json!({ "error": "no active consciousness session", "session_id": session_id })), + ) + })?; + + 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, + }; + + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel(); + let send_result = handle + .send(crate::consciousness::ConsciousnessEvent::UserMessage( + Box::new(crate::consciousness::UserMessageEvent { + objective: request.message, + branch: BranchId::main(), + steering, + ack: Some(ack_tx), + run_context: crate::consciousness::RunContext::default(), + }), + )) + .await; + + if send_result.is_err() { + return Err(internal_error(format!( + "consciousness actor for session {} is not running", + session_id + ))); + } + + let sid = SessionId::from_string(session_id); + match tokio::time::timeout(Duration::from_secs(5), ack_rx).await { + 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::Rejected { reason })) => Err(( + StatusCode::SERVICE_UNAVAILABLE, + Json(json!({ "error": "rejected", "reason": reason })), + )), + _ => Err(internal_error("consciousness actor did not respond")), + } +} + +/// Get the queue status for a consciousness session. +#[utoipa::path( + get, + path = "/sessions/{session_id}/queue", + tag = "consciousness", + params(("session_id" = String, Path, description = "Session identifier")), + responses( + (status = 200, description = "Queue status", body = QueueStatusResponse), + (status = 404, description = "Session not found or consciousness disabled", body = ErrorResponse) + ) +)] +async fn get_queue_status( + Path(session_id): Path, + State(state): State, +) -> Result, (StatusCode, Json)> { + let registry = state.consciousness_registry.as_ref().ok_or_else(|| { + ( + StatusCode::NOT_FOUND, + Json(json!({ "error": "consciousness mode is not enabled" })), + ) + })?; + + let handle = registry.get(&session_id).ok_or_else(|| { + ( + StatusCode::NOT_FOUND, + Json(json!({ "error": "no active consciousness session", "session_id": session_id })), + ) + })?; + + let status = handle + .query_status() + .await + .ok_or_else(|| internal_error("consciousness actor did not respond to status query"))?; + + let sid = SessionId::from_string(session_id); + Ok(Json(QueueStatusResponse { + session_id: sid, + mode: status.mode, + queue_depth: status.queue_depth, + has_active_run: status.has_active_run, + oldest_message_age_ms: status.oldest_message_age_ms, + pending: status + .queue_pending + .into_iter() + .map(|m| QueuePendingEntry { + id: m.id, + steering_mode: m.mode, + content: m.content, + }) + .collect(), + })) +} + // ─── Error helpers ─────────────────────────────────────────────────────────── fn internal_error(error: impl std::fmt::Display) -> (StatusCode, Json) { diff --git a/crates/arcand/src/consciousness.rs b/crates/arcand/src/consciousness.rs index 9947cb4..1ab3aee 100644 --- a/crates/arcand/src/consciousness.rs +++ b/crates/arcand/src/consciousness.rs @@ -53,12 +53,32 @@ impl Default for ConsciousnessConfig { pub enum ConsciousnessEvent { /// User message from HTTP POST /runs or /messages. UserMessage(Box), + /// Query the actor's current status (mode + queue snapshot). + QueryStatus { reply: oneshot::Sender }, /// Timer tick from internal intervals. TimerTick { tick_type: TimerTickType }, /// Graceful shutdown request. Shutdown, } +/// Snapshot of the actor's status for the GET /queue endpoint. +#[derive(Debug, Clone, serde::Serialize)] +pub struct ActorStatus { + pub mode: String, + pub queue_depth: usize, + pub queue_pending: Vec, + pub has_active_run: bool, + pub oldest_message_age_ms: Option, +} + +/// A pending message in the queue (serializable for API responses). +#[derive(Debug, Clone, serde::Serialize)] +pub struct PendingMessage { + pub id: String, + pub mode: String, + pub content: String, +} + /// Payload for a user message event (boxed to keep enum small). #[derive(Debug)] pub struct UserMessageEvent { @@ -211,6 +231,10 @@ impl SessionConsciousness { ) .await; } + ConsciousnessEvent::QueryStatus { reply } => { + let status = self.build_status(); + let _ = reply.send(status); + } ConsciousnessEvent::TimerTick { tick_type } => { self.handle_timer_tick(tick_type).await; } @@ -497,6 +521,32 @@ impl SessionConsciousness { } } } + + /// Build a status snapshot for the QueryStatus response. + fn build_status(&self) -> ActorStatus { + let queue_status = self.state.queue.status().ok(); + let pending = queue_status + .as_ref() + .map(|s| { + s.pending + .iter() + .map(|m| PendingMessage { + id: m.id.clone(), + mode: format!("{:?}", m.mode), + content: m.content.clone(), + }) + .collect() + }) + .unwrap_or_default(); + + ActorStatus { + mode: format!("{:?}", self.state.mode), + queue_depth: queue_status.as_ref().map(|s| s.depth).unwrap_or(0), + queue_pending: pending, + has_active_run: queue_status.as_ref().is_some_and(|s| s.has_active_run), + oldest_message_age_ms: queue_status.and_then(|s| s.oldest_message_age_ms), + } + } } // ─── Handle ───────────────────────────────────────────────────────────────── @@ -534,6 +584,19 @@ impl ConsciousnessHandle { self.tx.send(event).await } + /// Query the actor's current status (mode + queue snapshot). + pub async fn query_status(&self) -> Option { + let (reply_tx, reply_rx) = oneshot::channel(); + self.tx + .send(ConsciousnessEvent::QueryStatus { reply: reply_tx }) + .await + .ok()?; + tokio::time::timeout(Duration::from_secs(2), reply_rx) + .await + .ok()? + .ok() + } + /// Send a shutdown event and wait for the actor to stop. pub async fn shutdown(self) { let _ = self.tx.send(ConsciousnessEvent::Shutdown).await; diff --git a/crates/arcand/tests/consciousness_test.rs b/crates/arcand/tests/consciousness_test.rs index 4c5f675..ea0dcd4 100644 --- a/crates/arcand/tests/consciousness_test.rs +++ b/crates/arcand/tests/consciousness_test.rs @@ -22,6 +22,7 @@ use arcand::consciousness::{ SessionConsciousness, UserMessageEvent, }; use async_trait::async_trait; +use tokio::sync::oneshot; use aios_protocol::{BranchId, SessionId}; @@ -202,3 +203,74 @@ async fn channel_close_stops_actor() { let result = tokio::time::timeout(Duration::from_secs(5), handle).await; assert!(result.is_ok(), "actor should stop when channel closes"); } + +#[tokio::test] +async fn query_status_returns_idle_mode() { + let runtime = build_runtime(unique_root("consciousness-status")); + let registry = ConsciousnessRegistry::new(ConsciousnessConfig::default()); + + let handle = registry.get_or_create("test-status", BranchId::main(), runtime); + + // Give actor a moment to start. + tokio::time::sleep(Duration::from_millis(20)).await; + + let status = handle.query_status().await.expect("should get status"); + assert_eq!(status.mode, "Idle"); + assert_eq!(status.queue_depth, 0); + assert!(!status.has_active_run); + + registry.shutdown_all().await; +} + +#[tokio::test] +async fn multiple_messages_accepted_sequentially() { + let runtime = build_runtime(unique_root("consciousness-multi-msg")); + let session_id = SessionId::from_string("test-multi".to_string()); + + runtime + .create_session_with_id( + session_id.clone(), + "test", + PolicySet::default(), + aios_protocol::ModelRouting::default(), + ) + .await + .unwrap(); + + let config = ConsciousnessConfig { + max_agent_iterations: 1, + ..Default::default() + }; + let (handle, tx) = SessionConsciousness::spawn(session_id, BranchId::main(), runtime, config); + + // Send two messages — both should be accepted (either queued or immediate). + for i in 0..2 { + let (ack_tx, ack_rx) = oneshot::channel(); + tx.send(ConsciousnessEvent::UserMessage(Box::new( + UserMessageEvent { + objective: format!("message {i}"), + branch: BranchId::main(), + steering: SteeringMode::Collect, + ack: Some(ack_tx), + run_context: RunContext::default(), + }, + ))) + .await + .unwrap(); + + let ack = tokio::time::timeout(Duration::from_secs(10), ack_rx) + .await + .expect("should get ack") + .expect("channel should not be dropped"); + + match ack { + ConsciousnessAck::Accepted { .. } => {} // queued or immediate — both OK + ConsciousnessAck::Rejected { reason } => { + panic!("message {i} rejected: {reason}"); + } + } + } + + tx.send(ConsciousnessEvent::Shutdown).await.unwrap(); + let _ = tokio::time::timeout(Duration::from_secs(5), handle).await; +}