feat(consciousness): Spaces integration as external stimulus (BRO-458)

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<ActorStatus> },
     /// 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();

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;
+}