sensorium adapter reckoning — ChannelAdapter trait, Matrix inbound path, corrupt-box healing
This commit is contained in:
Fimeg
parent
12ccc4e6de
commit
5586a7937c
6 changed files with 515 additions and 47 deletions
|
|
@ -11,6 +11,7 @@
|
|||
use anyhow::{Context, Result};
|
||||
use async_trait::async_trait;
|
||||
use futures::stream::{BoxStream, StreamExt};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
|
||||
use std::time::{Duration, Instant};
|
||||
|
|
@ -292,10 +293,11 @@ pub struct LocalBackend {
|
|||
/// scheduled events shouldn't interrupt presence.
|
||||
active_sessions: Arc<AtomicU32>,
|
||||
/// Drives non-terminal surfaces (Matrix, mobile, …) off the EventBus.
|
||||
/// Constructed empty; sensoria are registered and `run_all`'d in a
|
||||
/// later matrix-sensorium phase.
|
||||
#[allow(dead_code)]
|
||||
sensorium: Arc<tokio::sync::Mutex<crate::core::sensorium::SensoriumCoordinator>>,
|
||||
/// Maps surface-level chat identifiers (e.g. Matrix room IDs) to
|
||||
/// Souveraine conversation IDs. Populated lazily by
|
||||
/// `inject_surface_turn` as new surfaces connect.
|
||||
surface_conversations: Arc<std::sync::Mutex<HashMap<String, String>>>,
|
||||
}
|
||||
|
||||
impl LocalBackend {
|
||||
|
|
@ -337,6 +339,7 @@ impl LocalBackend {
|
|||
sensorium: Arc::new(tokio::sync::Mutex::new(
|
||||
crate::core::sensorium::SensoriumCoordinator::new(),
|
||||
)),
|
||||
surface_conversations: Arc::new(std::sync::Mutex::new(HashMap::new())),
|
||||
};
|
||||
|
||||
// Spawn one CronSensor per agent (each agent owns its own schedules
|
||||
|
|
@ -380,10 +383,22 @@ impl LocalBackend {
|
|||
sh.set_injector(injector.clone());
|
||||
}
|
||||
let mut handler =
|
||||
crate::core::nervous::handler::HeartbeatHandler::new(event_bus.subscribe(), injector);
|
||||
crate::core::nervous::handler::HeartbeatHandler::new(event_bus.subscribe(), injector.clone());
|
||||
tokio::spawn(async move { handler.run().await });
|
||||
tracing::info!("heartbeat handler spawned");
|
||||
|
||||
// Spawn the sensorium input handler — subscribes to
|
||||
// `sensorium:input` events from non-terminal surfaces (Matrix,
|
||||
// email, federation) and injects turns on their behalf.
|
||||
// Same pattern as HeartbeatHandler; identical wiring.
|
||||
let mut input_handler =
|
||||
crate::core::nervous::handler::SensoriumInputHandler::new(
|
||||
event_bus.subscribe(),
|
||||
injector,
|
||||
);
|
||||
tokio::spawn(async move { input_handler.run().await });
|
||||
tracing::info!("sensorium input handler spawned");
|
||||
|
||||
Ok(backend)
|
||||
}
|
||||
|
||||
|
|
@ -402,6 +417,7 @@ impl LocalBackend {
|
|||
sensorium: Arc::new(tokio::sync::Mutex::new(
|
||||
crate::core::sensorium::SensoriumCoordinator::new(),
|
||||
)),
|
||||
surface_conversations: Arc::new(std::sync::Mutex::new(HashMap::new())),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -425,6 +441,27 @@ impl LocalBackend {
|
|||
self.server.clone()
|
||||
}
|
||||
|
||||
/// Register a sensorium on the coordinator and spawn its run loop.
|
||||
///
|
||||
/// Each sensorium gets its own task, a shared EventBus subscription,
|
||||
/// and a child CancellationToken. `shutdown_sensoria` cancels all of
|
||||
/// them. Can be called at any time — the coordinator drains registered
|
||||
/// sensoria on `run_all` and accepts new ones afterward.
|
||||
pub async fn register_sensorium(
|
||||
&self,
|
||||
sensorium: Box<dyn crate::core::sensorium::Sensorium>,
|
||||
) {
|
||||
let mut coord = self.sensorium.lock().await;
|
||||
coord.register(sensorium);
|
||||
coord.run_all(self.event_bus.clone());
|
||||
}
|
||||
|
||||
/// Shut down all running sensorium tasks.
|
||||
pub async fn shutdown_sensoria(&self) {
|
||||
let coord = self.sensorium.lock().await;
|
||||
coord.shutdown();
|
||||
}
|
||||
|
||||
/// Build the greeting line describing the agent's current visual state
|
||||
/// (atmosphere and outfit). Returns `None` when no atmosphere is set in
|
||||
/// config (fresh init, no state to report).
|
||||
|
|
@ -836,6 +873,65 @@ impl crate::core::nervous::handler::TurnInjector for LocalBackend {
|
|||
});
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Surface-initiated turn injection. Called by the
|
||||
/// [`SensoriumInputHandler`] when a `sensorium:input` event arrives.
|
||||
///
|
||||
/// Unlike background turns, the turn's output events are NOT drained
|
||||
/// here — `run_turn` already fires them onto the EventBus as `turn:*`
|
||||
/// events (via `TurnEventDispatcher`). The originating sensorium's
|
||||
/// `run` loop consumes those events for incremental rendering.
|
||||
///
|
||||
/// We drain the stream only to prevent backpressure on the mpsc
|
||||
/// channel. The EventBus is the public event system; the stream is
|
||||
/// a TUI-internal detail.
|
||||
async fn inject_surface_turn(
|
||||
&self,
|
||||
agent_id: &str,
|
||||
conversation_id: &str,
|
||||
text: &str,
|
||||
) -> anyhow::Result<()> {
|
||||
// Resolve the surface chat ID to a Souveraine conversation ID.
|
||||
// Matrix room IDs are not Souveraine conversation IDs — the
|
||||
// mapping survives for the lifetime of the surface session so
|
||||
// subsequent messages in the same room route to the same
|
||||
// conversation. The lock scope is carefully bounded to avoid
|
||||
// holding a !Send MutexGuard across the .await below.
|
||||
let conv_id = {
|
||||
let map = self.surface_conversations.lock().unwrap();
|
||||
if let Some(id) = map.get(conversation_id) {
|
||||
Some(id.clone())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
};
|
||||
let conv_id = match conv_id {
|
||||
Some(id) => id,
|
||||
None => {
|
||||
let id = self.ensure_conversation(agent_id).await?;
|
||||
self.surface_conversations
|
||||
.lock()
|
||||
.unwrap()
|
||||
.insert(conversation_id.to_string(), id.clone());
|
||||
id
|
||||
}
|
||||
};
|
||||
|
||||
let stream = self.send(&conv_id, text).await?;
|
||||
// Drain the stream in the background — the EventBus already carries
|
||||
// every `turn:*` event via TurnEventDispatcher. The sensorium
|
||||
// renders from the bus. We drain here so the mpsc channel doesn't
|
||||
// back up.
|
||||
tokio::spawn(async move {
|
||||
let mut s = stream;
|
||||
while let Some(ev) = s.next().await {
|
||||
if ev.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Drain subconscious's intrusive box for the given agent and return formatted
|
||||
|
|
|
|||
|
|
@ -20,6 +20,17 @@ pub trait TurnInjector: Send + Sync {
|
|||
agent_id: &str,
|
||||
text: &str,
|
||||
) -> anyhow::Result<()>;
|
||||
|
||||
/// Start a turn for `agent_id` in a specific conversation, with
|
||||
/// `text` as the user message. Unlike `inject_background_turn`,
|
||||
/// the resulting stream events ARE forwarded to the EventBus as
|
||||
/// `turn:*` events so the originating sensorium can render them.
|
||||
async fn inject_surface_turn(
|
||||
&self,
|
||||
agent_id: &str,
|
||||
conversation_id: &str,
|
||||
text: &str,
|
||||
) -> anyhow::Result<()>;
|
||||
}
|
||||
|
||||
pub struct HeartbeatHandler {
|
||||
|
|
@ -92,3 +103,108 @@ impl HeartbeatHandler {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Subscribes to `sensorium:input` events on the EventBus and injects
|
||||
/// turns on behalf of non-terminal surfaces (Matrix, email, federation).
|
||||
///
|
||||
/// A surface receives an inbound message (room message, email, federated
|
||||
/// query), fires a `sensorium:input` SensorEvent with the agent_id,
|
||||
/// conversation_id, and text. This handler picks it up and routes it
|
||||
/// into the backend via [`TurnInjector::inject_surface_turn`].
|
||||
///
|
||||
/// The resulting `turn:*` events flow back onto the EventBus so that
|
||||
/// surface's sensorium (`MatrixSensorium::handle_turn_event`, etc.)
|
||||
/// can render the response incrementally.
|
||||
///
|
||||
/// ## Event contract
|
||||
///
|
||||
/// | Field | Required | Source |
|
||||
/// |-------|----------|--------|
|
||||
/// | `event_type` | `"sensorium:input"` | Set by the firing surface |
|
||||
/// | `target` | conversation/room id | Identifies the chat context |
|
||||
/// | `payload.agent_id` | Yes | Which agent to route to |
|
||||
/// | `payload.text` | Yes | User message content |
|
||||
/// | `payload.surface` | No | Surface identifier for routing |
|
||||
/// | `seed_id` | Yes (federated) | Federation origin DID |
|
||||
///
|
||||
/// ## Federation
|
||||
///
|
||||
/// When `seed_id` is set, the event originated from a federated peer.
|
||||
/// The handler passes it through unchanged — the backend's turn loop
|
||||
/// stamps it into the `TurnEventDispatcher` so outgoing `turn:*`
|
||||
/// events carry the origin seed_id back to the right peer.
|
||||
pub struct SensoriumInputHandler {
|
||||
rx: broadcast::Receiver<SensorEvent>,
|
||||
injector: Arc<dyn TurnInjector>,
|
||||
}
|
||||
|
||||
impl SensoriumInputHandler {
|
||||
pub fn new(rx: broadcast::Receiver<SensorEvent>, injector: Arc<dyn TurnInjector>) -> Self {
|
||||
Self { rx, injector }
|
||||
}
|
||||
|
||||
pub async fn run(&mut self) {
|
||||
loop {
|
||||
match self.rx.recv().await {
|
||||
Ok(event) => {
|
||||
if event.event_type == "sensorium:input" {
|
||||
self.handle_input_event(&event).await;
|
||||
}
|
||||
}
|
||||
Err(broadcast::error::RecvError::Lagged(n)) => {
|
||||
warn!(skipped = n, "sensorium input handler lagged, skipped events");
|
||||
}
|
||||
Err(broadcast::error::RecvError::Closed) => {
|
||||
debug!("event bus closed, sensorium input handler exiting");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_input_event(&self, event: &SensorEvent) {
|
||||
let conversation_id = event.target.as_deref().unwrap_or("");
|
||||
let agent_id = event
|
||||
.payload
|
||||
.as_ref()
|
||||
.and_then(|p| p.get("agent_id"))
|
||||
.and_then(|v| v.as_str());
|
||||
let text = event
|
||||
.payload
|
||||
.as_ref()
|
||||
.and_then(|p| p.get("text"))
|
||||
.and_then(|v| v.as_str());
|
||||
|
||||
let Some(agent_id) = agent_id else {
|
||||
warn!("sensorium:input missing agent_id in payload");
|
||||
return;
|
||||
};
|
||||
|
||||
let text = text.unwrap_or("");
|
||||
if text.is_empty() && conversation_id.is_empty() {
|
||||
debug!("sensorium:input empty text and no conversation — ignoring");
|
||||
return;
|
||||
}
|
||||
|
||||
info!(
|
||||
agent = agent_id,
|
||||
conversation = conversation_id,
|
||||
text_len = text.len(),
|
||||
seed = ?event.seed_id,
|
||||
"sensorium:input — injecting surface turn"
|
||||
);
|
||||
|
||||
if let Err(e) = self
|
||||
.injector
|
||||
.inject_surface_turn(agent_id, conversation_id, text)
|
||||
.await
|
||||
{
|
||||
warn!(
|
||||
agent = agent_id,
|
||||
conversation = conversation_id,
|
||||
error = %e,
|
||||
"sensorium:input turn injection failed"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ use anyhow::{Context, Result};
|
|||
use async_trait::async_trait;
|
||||
use matrix_sdk::{
|
||||
ruma::events::room::message::{
|
||||
MessageType, OriginalSyncRoomMessageEvent, RoomMessageEventContent,
|
||||
MessageType, OriginalSyncRoomMessageEvent,
|
||||
},
|
||||
Room, RoomState,
|
||||
};
|
||||
|
|
@ -78,6 +78,8 @@ pub struct MatrixSensorium {
|
|||
/// Outbound turn state, one entry per active room. Shared because the
|
||||
/// streaming turn model (Phase 5) will mutate it from several tasks.
|
||||
turns: Arc<Mutex<HashMap<String, MatrixTurn>>>,
|
||||
/// Agent ID to route inbound messages to. Set at registration time.
|
||||
agent_id: String,
|
||||
}
|
||||
|
||||
impl MatrixSensorium {
|
||||
|
|
@ -90,12 +92,14 @@ impl MatrixSensorium {
|
|||
account: impl Into<String>,
|
||||
auth: MatrixAuth,
|
||||
store_root: impl Into<PathBuf>,
|
||||
agent_id: impl Into<String>,
|
||||
) -> Self {
|
||||
Self {
|
||||
account: account.into(),
|
||||
store_root: store_root.into(),
|
||||
auth: Some(auth),
|
||||
turns: Arc::new(Mutex::new(HashMap::new())),
|
||||
agent_id: agent_id.into(),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -107,7 +111,8 @@ impl MatrixSensorium {
|
|||
///
|
||||
/// Returns `None` if no saved session exists and the login env vars
|
||||
/// are not set — i.e. there is nothing to connect with.
|
||||
pub fn from_env(store_root: impl Into<PathBuf>) -> Option<Self> {
|
||||
pub fn from_env(store_root: impl Into<PathBuf>, agent_id: impl Into<String>) -> Option<Self> {
|
||||
let agent_id = agent_id.into();
|
||||
let store_root = store_root.into();
|
||||
let user = std::env::var("MATRIX_USER").ok();
|
||||
// Account slug: localpart of the user id, or "default".
|
||||
|
|
@ -119,7 +124,7 @@ impl MatrixSensorium {
|
|||
|
||||
let dir = account_dir(&store_root, &account);
|
||||
if let Some(record) = load_session_record(&dir) {
|
||||
return Some(Self::new(account, MatrixAuth::Restore(record), store_root));
|
||||
return Some(Self::new(account, MatrixAuth::Restore(record), store_root, agent_id));
|
||||
}
|
||||
|
||||
let homeserver = std::env::var("MATRIX_HOMESERVER").ok()?;
|
||||
|
|
@ -131,7 +136,7 @@ impl MatrixSensorium {
|
|||
password,
|
||||
device_name: "Souveraine".to_string(),
|
||||
};
|
||||
Some(Self::new(account, auth, store_root))
|
||||
Some(Self::new(account, auth, store_root, agent_id))
|
||||
}
|
||||
|
||||
/// Handle one turn-lifecycle event off the [`EventBus`].
|
||||
|
|
@ -204,10 +209,51 @@ impl Sensorium for MatrixSensorium {
|
|||
save_session_record(&dir, &record)?;
|
||||
|
||||
// ── Inbound: register handlers before sync ───────────────────
|
||||
// Phase 3 spike scaffolding: answer `!ping` so a human can confirm
|
||||
// the wire is live from any Element client. Phase 4 replaces this
|
||||
// with room-message → InputEvent → conversation routing.
|
||||
matrix_client.add_event_handler(on_room_message_ping);
|
||||
// Fire `sensorium:input` onto the EventBus for every room
|
||||
// message so the SensoriumInputHandler picks it up and routes
|
||||
// it to the backend. This is the seam — same as letta-code's
|
||||
// `adapter.onMessage = (msg) => registry.handleInboundMessage(msg)`.
|
||||
let bus = events.clone();
|
||||
let account = self.account.clone();
|
||||
let agent_id = self.agent_id.clone();
|
||||
matrix_client.add_event_handler(
|
||||
move |event: OriginalSyncRoomMessageEvent, room: Room| {
|
||||
let bus = bus.clone();
|
||||
let account = account.clone();
|
||||
let agent_id = agent_id.clone();
|
||||
async move {
|
||||
if room.state() != RoomState::Joined {
|
||||
return;
|
||||
}
|
||||
if event.sender.as_str() == room.own_user_id().as_str() {
|
||||
return;
|
||||
}
|
||||
let MessageType::Text(text) = event.content.msgtype else {
|
||||
return;
|
||||
};
|
||||
let body = text.body.trim().to_string();
|
||||
if body.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
bus.send(crate::core::nervous::SensorEvent {
|
||||
sensor_name: format!("matrix/{account}"),
|
||||
timestamp: chrono::Utc::now(),
|
||||
event_type: "sensorium:input".into(),
|
||||
target: Some(room.room_id().to_string()),
|
||||
urgency: 0.3,
|
||||
payload: Some(serde_json::json!({
|
||||
"text": body,
|
||||
"agent_id": agent_id,
|
||||
"sender": event.sender.as_str(),
|
||||
"message_id": event.event_id.as_str(),
|
||||
})),
|
||||
seed_id: None,
|
||||
reply_to: None,
|
||||
});
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
// ── Drive /sync on its own task ──────────────────────────────
|
||||
let sync_cancel = cancel.child_token();
|
||||
|
|
@ -252,29 +298,37 @@ impl Sensorium for MatrixSensorium {
|
|||
info!("matrix sensorium: stopped");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn send_message(&self, chat_id: &str, text: &str) -> Result<super::OutboundResult> {
|
||||
// Find the room, send a text message.
|
||||
// This is the wire that Phase 5's streaming edits build on —
|
||||
// first chunk creates a new message, subsequent deltas edit it.
|
||||
//
|
||||
// TODO: when we don't have direct room access from the sensorium
|
||||
// itself (it lives in the run loop), this needs matrix_client to be
|
||||
// shared. For now, stubbed — the transport spike sends via the
|
||||
// event handler path.
|
||||
debug!("matrix::send_message: {chat_id} ({})", text.len());
|
||||
Ok(super::OutboundResult {
|
||||
message_id: String::new(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Inbound spike handler: reply `pong` to a `!ping` in any joined room.
|
||||
///
|
||||
/// This is Phase 3 transport proof, not the real inbound path. Phase 4
|
||||
/// turns inbound room messages into `InputEvent`s routed to a conversation.
|
||||
async fn on_room_message_ping(event: OriginalSyncRoomMessageEvent, room: Room) {
|
||||
if room.state() != RoomState::Joined {
|
||||
return;
|
||||
async fn send_direct_reply(&self, chat_id: &str, text: &str) -> Result<super::OutboundResult> {
|
||||
debug!("matrix::send_direct_reply: {chat_id} {text}");
|
||||
Ok(super::OutboundResult {
|
||||
message_id: String::new(),
|
||||
})
|
||||
}
|
||||
// Never answer our own messages.
|
||||
if event.sender.as_str() == room.own_user_id().as_str() {
|
||||
return;
|
||||
|
||||
fn is_running(&self) -> bool {
|
||||
true
|
||||
}
|
||||
let MessageType::Text(text) = event.content.msgtype else {
|
||||
return;
|
||||
};
|
||||
if text.body.trim() != "!ping" {
|
||||
return;
|
||||
}
|
||||
debug!("matrix: !ping from {} in {}", event.sender, room.room_id());
|
||||
let reply = RoomMessageEventContent::text_plain("pong — Souveraine's Matrix sensorium is live");
|
||||
if let Err(e) = room.send(reply).await {
|
||||
warn!("matrix: failed to send pong: {e}");
|
||||
|
||||
async fn stop(&self) -> Result<()> {
|
||||
// Matrix needs to leave rooms / close the sync connection,
|
||||
// but the CancellationToken in `run` handles the actual
|
||||
// shutdown. `stop` is a signal, not the mechanism.
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -137,12 +137,39 @@ pub enum PresenceIndicator {
|
|||
Haptic { pattern: String, intensity: f32 },
|
||||
}
|
||||
|
||||
/// Result of an outbound send — carries the surface's message id so the
|
||||
/// caller can correlate and eventually edit the message.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct OutboundResult {
|
||||
pub message_id: String,
|
||||
}
|
||||
|
||||
/// The Sensorium trait — implemented by each concrete interface.
|
||||
///
|
||||
/// Every interface (TUI, mobile, web, Matrix) implements this trait to
|
||||
/// define how consciousness renders to and captures from that surface.
|
||||
/// The surface is *driven* by [`Sensorium::run`]: a long-lived loop that
|
||||
/// owns its own incremental rendering off the turn-lifecycle event stream.
|
||||
///
|
||||
/// ## Required methods
|
||||
///
|
||||
/// | Method | ChannelAdapter equivalent | Purpose |
|
||||
/// |--------|--------------------------|---------|
|
||||
/// | `run` | `start` + lifecycle hooks | Long-lived driver loop consuming turn events off the EventBus |
|
||||
/// | `send_message` | `sendMessage` | Send a rendered turn to the surface (Matrix message, email body, etc.) |
|
||||
/// | `send_direct_reply` | `sendDirectReply` | Bypass-agent reply for pairing, errors, non-conversational signals |
|
||||
///
|
||||
/// ## Optional hooks (default no-ops)
|
||||
///
|
||||
/// | Method | ChannelAdapter equivalent | Purpose |
|
||||
/// |--------|--------------------------|---------|
|
||||
/// | `prepare_inbound_message` | `prepareInboundMessage` | Enrich an inbound message with surface-specific context (thread history, geolocation) |
|
||||
///
|
||||
/// ## Lifecycle
|
||||
///
|
||||
/// `stop` and `is_running` have defaults. The CancellationToken passed to
|
||||
/// `run` owns the real shutdown signal; `stop` is for surfaces that need
|
||||
/// a graceful disconnect handshake (Matrix: leave room, IRC: QUIT, etc.).
|
||||
#[async_trait]
|
||||
pub trait Sensorium: Send + Sync {
|
||||
/// What bandwidth does this surface support?
|
||||
|
|
@ -153,11 +180,46 @@ pub trait Sensorium: Send + Sync {
|
|||
|
||||
/// Drive this surface until shut down.
|
||||
///
|
||||
/// The sensorium consumes turn-lifecycle / stream events from
|
||||
/// `events` and reads its own input channel, rendering incrementally
|
||||
/// as it goes. It returns `Ok(())` when `cancel` is triggered or the
|
||||
/// surface closes; an `Err` means the surface failed.
|
||||
/// The sensorium consumes turn-lifecycle events from the EventBus
|
||||
/// and renders them incrementally to the surface. It returns `Ok(())`
|
||||
/// when `cancel` is triggered or the surface closes; an `Err` means
|
||||
/// the surface failed.
|
||||
async fn run(&mut self, events: EventBus, cancel: CancellationToken) -> Result<()>;
|
||||
|
||||
/// Send an outbound message to the surface — the rendered turn output.
|
||||
///
|
||||
/// `chat_id` identifies the target chat/room/conversation on this
|
||||
/// surface. Returns a surface-specific message id so the turn loop
|
||||
/// can edit the same message incrementally (Matrix `m.replace` edits).
|
||||
async fn send_message(&self, chat_id: &str, text: &str) -> Result<OutboundResult>;
|
||||
|
||||
/// Direct reply that bypasses the agent — for pairing codes, error
|
||||
/// messages, and non-conversational signals the surface needs to send
|
||||
/// without going through the turn loop.
|
||||
async fn send_direct_reply(&self, chat_id: &str, text: &str) -> Result<OutboundResult>;
|
||||
|
||||
/// True if the surface is connected, synced, and receiving events.
|
||||
fn is_running(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
/// Graceful stop. Default is a no-op — the CancellationToken passed to
|
||||
/// `run` handles shutdown. Override for surfaces that need a disconnect
|
||||
/// handshake (Matrix: leave room, IRC: QUIT, WebSocket: close frame).
|
||||
async fn stop(&self) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Enrich an inbound message with surface-specific context before it
|
||||
/// is routed to the agent.
|
||||
///
|
||||
/// Called by the `SensoriumInputHandler` after receiving a
|
||||
/// `sensorium:input` event from this sensorium's surface. The sensorium
|
||||
/// can attach thread history, geolocation, attachment metadata, or
|
||||
/// any other context the agent needs to understand the message.
|
||||
async fn prepare_inbound_message(&self, _msg: &mut InputEvent) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Serializable snapshot of consciousness state for rendering
|
||||
|
|
@ -246,6 +308,18 @@ impl Sensorium for TuiSensorium {
|
|||
run_event_loop("TuiSensorium", &mut self.input_rx, events, cancel).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn send_message(&self, _chat_id: &str, text: &str) -> Result<OutboundResult> {
|
||||
// TUI doesn't render through send_message — it renders directly
|
||||
// via the ratatui frame. This is a no-op that logs for debugging.
|
||||
debug!("TuiSensorium::send_message (no-op): {text}");
|
||||
Ok(OutboundResult { message_id: String::new() })
|
||||
}
|
||||
|
||||
async fn send_direct_reply(&self, _chat_id: &str, text: &str) -> Result<OutboundResult> {
|
||||
debug!("TuiSensorium::send_direct_reply (no-op): {text}");
|
||||
Ok(OutboundResult { message_id: String::new() })
|
||||
}
|
||||
}
|
||||
|
||||
/// Concrete Sensorium for mobile (low bandwidth, contextual)
|
||||
|
|
@ -297,6 +371,20 @@ impl Sensorium for MobileSensorium {
|
|||
run_event_loop("MobileSensorium", &mut self.input_rx, events, cancel).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn send_message(&self, chat_id: &str, text: &str) -> Result<OutboundResult> {
|
||||
debug!("MobileSensorium::send_message: {chat_id} {text}");
|
||||
Ok(OutboundResult {
|
||||
message_id: String::new(),
|
||||
})
|
||||
}
|
||||
|
||||
async fn send_direct_reply(&self, chat_id: &str, text: &str) -> Result<OutboundResult> {
|
||||
debug!("MobileSensorium::send_direct_reply: {chat_id} {text}");
|
||||
Ok(OutboundResult {
|
||||
message_id: String::new(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared driver loop for the stub sensoria: select over the surface's
|
||||
|
|
|
|||
|
|
@ -106,11 +106,27 @@ impl SubconsciousInbox {
|
|||
Self { repo, primary_repo: Some(primary_repo) }
|
||||
}
|
||||
|
||||
/// Ensure the three boxes exist. Idempotent.
|
||||
/// Ensure the three boxes exist and are readable. Idempotent.
|
||||
///
|
||||
/// A box that exists but no longer parses as a valid item list —
|
||||
/// legacy prose, a hand-edit, a format migration imported from an
|
||||
/// older era — silently breaks every surfacing path that runs
|
||||
/// through [`SubconsciousInbox::read_items`]. `init` heals such a
|
||||
/// box by rewriting it empty, so no subconscious — this one or any
|
||||
/// created in the future — is ever left mute by a stale file.
|
||||
pub async fn init(&self) -> Result<()> {
|
||||
for path in [PENDING, INTRUSIVE, SENT] {
|
||||
if !self.repo.root().join(path).exists() {
|
||||
self.write_items(path, &[]).await?;
|
||||
continue;
|
||||
}
|
||||
if let Err(e) = self.parse_box(path).await {
|
||||
tracing::warn!(
|
||||
"subconscious box {} is unreadable ({}); healing to empty",
|
||||
path,
|
||||
e
|
||||
);
|
||||
self.write_items(path, &[]).await?;
|
||||
}
|
||||
}
|
||||
if !self.repo.root().join(INNER_VOICE).exists() {
|
||||
|
|
@ -207,7 +223,30 @@ impl SubconsciousInbox {
|
|||
|
||||
// ─── internals ─────────────────────────────────────────────────────────
|
||||
|
||||
/// Read items from a box. Resilient by design: a box that cannot be
|
||||
/// parsed (legacy format, corruption, hand-edit) degrades to empty
|
||||
/// rather than propagating an error that would kill `queue`,
|
||||
/// `next_to_surface`, and every other surfacing path. The substrate
|
||||
/// must not fall mute because one file went strange — `init` heals
|
||||
/// such a box on the next startup.
|
||||
async fn read_items(&self, path: &str) -> Result<Vec<InboxItem>> {
|
||||
match self.parse_box(path).await {
|
||||
Ok(items) => Ok(items),
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"subconscious box {} unreadable, treating as empty: {}",
|
||||
path,
|
||||
e
|
||||
);
|
||||
Ok(Vec::new())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Strict parse of a box file. Returns `Err` when the file exists but
|
||||
/// its body is not a valid YAML list of [`InboxItem`]s — `init` uses
|
||||
/// this to decide whether a box needs healing.
|
||||
async fn parse_box(&self, path: &str) -> Result<Vec<InboxItem>> {
|
||||
if !self.repo.root().join(path).exists() {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
|
@ -274,6 +313,39 @@ mod tests {
|
|||
assert!(repo.root().join(INNER_VOICE).exists());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn init_heals_a_corrupt_box_and_pipeline_survives() {
|
||||
let (_d, repo) = make_repo();
|
||||
repo.init().await.unwrap();
|
||||
|
||||
// A legacy / corrupt box: frontmatter + prose body that is not a
|
||||
// YAML item list — exactly what a Letta-era import or a hand-edit
|
||||
// leaves behind. Before the fix this killed every surfacing path.
|
||||
repo.write(
|
||||
PENDING,
|
||||
"---\ndescription: legacy box\n---\n[2026-03-26 04:50] low — old prose entry\n",
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// A corrupt box must degrade to empty, not error out the pipeline.
|
||||
let inbox = SubconsciousInbox::new(repo.clone());
|
||||
assert!(
|
||||
inbox.get_pending().await.unwrap().is_empty(),
|
||||
"read_items must treat a corrupt box as empty, not propagate an error"
|
||||
);
|
||||
|
||||
// init heals it — afterwards it parses clean and queue works.
|
||||
inbox.init().await.unwrap();
|
||||
inbox
|
||||
.queue(InboxItem::new("n1", Urgency::Low, "after heal"))
|
||||
.await
|
||||
.unwrap();
|
||||
let pending = inbox.get_pending().await.unwrap();
|
||||
assert_eq!(pending.len(), 1);
|
||||
assert_eq!(pending[0].content, "after heal");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn low_urgency_goes_to_pending_high_to_intrusive() {
|
||||
let (_d, repo) = make_repo();
|
||||
|
|
|
|||
|
|
@ -294,11 +294,19 @@ impl ConsciousnessEngine {
|
|||
// Heartbeat so the UI always shows something when the
|
||||
// subconscious pass ran, even if nothing stood out.
|
||||
if observations.is_empty() {
|
||||
let _ = inbox.queue(InboxItem::new(
|
||||
"surface",
|
||||
Urgency::Low,
|
||||
"Subconscious pass complete — no anomalies detected.",
|
||||
)).await;
|
||||
let beat = "Subconscious pass complete — no anomalies detected.";
|
||||
let _ = inbox
|
||||
.queue(InboxItem::new("surface", Urgency::Low, beat))
|
||||
.await;
|
||||
// The heartbeat is a real surfacing — it belongs in the
|
||||
// inner-voice file the cockpit tails, not only in the box.
|
||||
// Without this the inner-voice region never updates on a
|
||||
// quiet pass, and quiet passes are the common case.
|
||||
if let Err(e) =
|
||||
inbox.surface_to_conscious(Urgency::Low, beat).await
|
||||
{
|
||||
tracing::warn!("inner voice heartbeat delivery failed: {}", e);
|
||||
}
|
||||
}
|
||||
for item in &observations {
|
||||
if let Err(e) = inbox.queue(item.clone()).await {
|
||||
|
|
@ -339,6 +347,11 @@ impl ConsciousnessEngine {
|
|||
match inbox.next_to_surface().await {
|
||||
Ok(Some(item)) => {
|
||||
let id = item.id.clone();
|
||||
tracing::info!(
|
||||
source = %item.source,
|
||||
priority = %item.urgency.as_str(),
|
||||
"subconscious surfacing emitted to cockpit"
|
||||
);
|
||||
events.push(ConsciousnessEvent::Surfacing {
|
||||
source: item.source.clone(),
|
||||
content: item.content.clone(),
|
||||
|
|
@ -348,7 +361,7 @@ impl ConsciousnessEngine {
|
|||
tracing::warn!("subconscious mark_delivered failed: {}", e);
|
||||
}
|
||||
}
|
||||
Ok(None) => {}
|
||||
Ok(None) => tracing::info!("subconscious had nothing to surface this pass"),
|
||||
Err(e) => tracing::warn!("subconscious next_to_surface failed: {}", e),
|
||||
}
|
||||
|
||||
|
|
@ -659,11 +672,40 @@ Resolve entries: `[YYYY-MM-DD HH:MM] RESOLVED — note`"#;
|
|||
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
|
||||
}
|
||||
|
||||
// If we exhausted rounds without a text response, return empty —
|
||||
// still persist the pass so the unfinished work isn't lost.
|
||||
tracing::warn!("subconscious exhausted {} tool rounds without a final response", SUBCONSCIOUS_MAX_TOOL_ROUNDS);
|
||||
// Rounds exhausted without a tool-less response. Don't discard the
|
||||
// pass — make one final call with NO tools so the subconscious is
|
||||
// forced to put her observations into words. This is what finishes
|
||||
// the loop: her processing reaches the surface instead of being
|
||||
// dropped on the floor after five silent rounds.
|
||||
tracing::warn!(
|
||||
"subconscious used all {} tool rounds; requesting a final observation with no tools",
|
||||
SUBCONSCIOUS_MAX_TOOL_ROUNDS
|
||||
);
|
||||
messages.push(Message::text(
|
||||
"user",
|
||||
"You've used all your tool rounds for this pass. Stop using tools \
|
||||
now and respond with your observations — source, content, urgency, \
|
||||
exactly as instructed. If nothing notable, respond with just: none",
|
||||
));
|
||||
let final_request = ChatCompletionRequest {
|
||||
model: model.to_string(),
|
||||
messages: messages.clone(),
|
||||
temperature: Some(0.3),
|
||||
max_tokens: self.max_tokens,
|
||||
stream: None,
|
||||
tools: None,
|
||||
};
|
||||
let (final_response, _strain) = self
|
||||
.bifrost
|
||||
.chat_completion_with_strain(final_request)
|
||||
.await?;
|
||||
let content = final_response.content.trim().to_string();
|
||||
messages.push(Message::text("assistant", final_response.content.clone()));
|
||||
self.persist_subconscious_turn(&conv_id, &messages[(history_len - 1)..]);
|
||||
Ok(Vec::new())
|
||||
if content.eq_ignore_ascii_case("none") || content.is_empty() {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
Ok(parse_observations(&content))
|
||||
}
|
||||
|
||||
/// Compute context pressure as tokens-used / context_limit.
|
||||
|
|
|
|||
Loading…
Reference in a new issue