diff --git a/crates/arcand/src/consciousness.rs b/crates/arcand/src/consciousness.rs index 1ab3aee..ff45eba 100644 --- a/crates/arcand/src/consciousness.rs +++ b/crates/arcand/src/consciousness.rs @@ -32,6 +32,10 @@ pub struct ConsciousnessConfig { pub max_agent_iterations: u32, /// Time before transitioning from Idle to Sleeping (default: 5min). pub idle_to_sleep: Duration, + /// Whether Spaces integration is enabled (default: false). + pub spaces_enabled: bool, + /// How often to poll Spaces for new messages (default: 10s). + pub spaces_poll_interval: Duration, } impl Default for ConsciousnessConfig { @@ -42,6 +46,8 @@ impl Default for ConsciousnessConfig { idle_check_interval: Duration::from_secs(60), max_agent_iterations: 10, idle_to_sleep: Duration::from_secs(300), + spaces_enabled: false, + spaces_poll_interval: Duration::from_secs(10), } } } @@ -57,6 +63,15 @@ pub enum ConsciousnessEvent { QueryStatus { reply: oneshot::Sender }, /// Timer tick from internal intervals. TimerTick { tick_type: TimerTickType }, + /// External stimulus from the Spaces distributed networking layer. + SpacesMessage { + /// The Spaces channel where the message originated. + channel_id: String, + /// Sender identity (hex string from SpacetimeDB). + sender: String, + /// Message content. + content: String, + }, /// Graceful shutdown request. Shutdown, } @@ -189,9 +204,11 @@ impl SessionConsciousness { let mut heartbeat = tokio::time::interval(self.config.heartbeat_interval); let mut idle_check = tokio::time::interval(self.config.idle_check_interval); + let mut spaces_poll = tokio::time::interval(self.config.spaces_poll_interval); // Skip the immediate first tick (fires at interval creation). heartbeat.tick().await; idle_check.tick().await; + spaces_poll.tick().await; loop { let event = tokio::select! { @@ -213,6 +230,19 @@ impl SessionConsciousness { _ = idle_check.tick() => ConsciousnessEvent::TimerTick { tick_type: TimerTickType::IdleCheck, }, + + // Spaces polling: periodically check for new distributed messages. + // Guarded by config flag — does nothing when spaces_enabled is false. + _ = spaces_poll.tick(), if self.config.spaces_enabled => { + // Stub: actual Spaces SDK integration requires SpacetimeDB client + // which uses blocking I/O. Future work will use spawn_blocking + // to poll SpacesPort::read_messages here. + debug!( + session = %self.state.session_id, + "spaces polling tick (not yet connected)" + ); + continue; + }, }; match event { @@ -238,6 +268,14 @@ impl SessionConsciousness { ConsciousnessEvent::TimerTick { tick_type } => { self.handle_timer_tick(tick_type).await; } + ConsciousnessEvent::SpacesMessage { + channel_id, + sender, + content, + } => { + self.handle_spaces_message(channel_id, sender, content) + .await; + } } } @@ -522,6 +560,72 @@ impl SessionConsciousness { } } + /// Handle an incoming Spaces message (external distributed stimulus). + /// + /// If the actor is idle or sleeping, the message is re-injected as a + /// `UserMessage` to trigger a deliberation cycle. If active, the message + /// is logged as context. Own messages (matching the session_id) are ignored. + async fn handle_spaces_message(&mut self, channel_id: String, sender: String, content: String) { + // Filter out own messages to avoid feedback loops. + if sender == self.state.session_id.as_str() { + debug!( + session = %self.state.session_id, + %channel_id, + "ignoring own Spaces message" + ); + return; + } + + let content_preview = if content.len() > 80 { + format!("{}...", &content[..80]) + } else { + content.clone() + }; + + info!( + session = %self.state.session_id, + %channel_id, + %sender, + content_preview, + "received Spaces message" + ); + + match self.state.mode { + ConsciousnessMode::Idle | ConsciousnessMode::Sleeping => { + // Re-inject as a user message to trigger a deliberation cycle. + info!( + session = %self.state.session_id, + mode = ?self.state.mode, + "Spaces message triggering deliberation cycle" + ); + let objective = format!("[spaces:{channel_id}] @{sender}: {content}"); + self.handle_user_message( + objective, + self.state.branch.clone(), + SteeringMode::Collect, + None, + RunContext::default(), + ) + .await; + } + ConsciousnessMode::Active => { + // Don't interrupt the current run — log as context for future use. + debug!( + session = %self.state.session_id, + %channel_id, + %sender, + "Spaces message received during active run (logged as context)" + ); + } + ConsciousnessMode::ShuttingDown => { + debug!( + session = %self.state.session_id, + "ignoring Spaces message during shutdown" + ); + } + } + } + /// Build a status snapshot for the QueryStatus response. fn build_status(&self) -> ActorStatus { let queue_status = self.state.queue.status().ok(); diff --git a/crates/arcand/tests/consciousness_test.rs b/crates/arcand/tests/consciousness_test.rs index ea0dcd4..017e0c2 100644 --- a/crates/arcand/tests/consciousness_test.rs +++ b/crates/arcand/tests/consciousness_test.rs @@ -274,3 +274,104 @@ async fn multiple_messages_accepted_sequentially() { tx.send(ConsciousnessEvent::Shutdown).await.unwrap(); let _ = tokio::time::timeout(Duration::from_secs(5), handle).await; } + +// ─── Spaces integration tests (BRO-458) ──────────────────────────────────── + +#[tokio::test] +async fn spaces_message_handled_when_idle() { + let runtime = build_runtime(unique_root("consciousness-spaces-idle")); + let session_id = SessionId::from_string("test-spaces-idle".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); + + // Give actor a moment to start and be in Idle state. + tokio::time::sleep(Duration::from_millis(20)).await; + + // Send a SpacesMessage while the actor is Idle. + tx.send(ConsciousnessEvent::SpacesMessage { + channel_id: "agent-logs".to_string(), + sender: "peer-agent-abc".to_string(), + content: "Hey, can you check the latest deployment?".to_string(), + }) + .await + .unwrap(); + + // Give the actor time to process the message (it will trigger a deliberation cycle). + tokio::time::sleep(Duration::from_millis(500)).await; + + // The actor should still be alive and operational after processing. + let (reply_tx, reply_rx) = oneshot::channel(); + tx.send(ConsciousnessEvent::QueryStatus { reply: reply_tx }) + .await + .unwrap(); + + let status = tokio::time::timeout(Duration::from_secs(5), reply_rx) + .await + .expect("should get status within 5s") + .expect("status channel should not be dropped"); + + // After processing the SpacesMessage (which triggers a run cycle), the actor + // should be back to Idle. + assert_eq!(status.mode, "Idle"); + + tx.send(ConsciousnessEvent::Shutdown).await.unwrap(); + let _ = tokio::time::timeout(Duration::from_secs(5), handle).await; +} + +#[tokio::test] +async fn own_spaces_messages_ignored() { + let runtime = build_runtime(unique_root("consciousness-spaces-self")); + let session_id_str = "test-spaces-self"; + let session_id = SessionId::from_string(session_id_str.to_string()); + + let config = ConsciousnessConfig::default(); + let (handle, tx) = SessionConsciousness::spawn(session_id, BranchId::main(), runtime, config); + + // Give actor a moment to start. + tokio::time::sleep(Duration::from_millis(20)).await; + + // Send a SpacesMessage where sender matches the actor's own session_id. + // This should be silently ignored (no crash, no state change). + tx.send(ConsciousnessEvent::SpacesMessage { + channel_id: "agent-logs".to_string(), + sender: session_id_str.to_string(), + content: "I posted this myself".to_string(), + }) + .await + .unwrap(); + + // Give the actor time to process. + tokio::time::sleep(Duration::from_millis(100)).await; + + // Actor should still be Idle (own message was ignored, no run triggered). + let (reply_tx, reply_rx) = oneshot::channel(); + tx.send(ConsciousnessEvent::QueryStatus { reply: reply_tx }) + .await + .unwrap(); + + let status = tokio::time::timeout(Duration::from_secs(5), reply_rx) + .await + .expect("should get status within 5s") + .expect("status channel should not be dropped"); + + assert_eq!(status.mode, "Idle"); + assert!(!status.has_active_run); + assert_eq!(status.queue_depth, 0); + + tx.send(ConsciousnessEvent::Shutdown).await.unwrap(); + let _ = tokio::time::timeout(Duration::from_secs(5), handle).await; +}