Fimeg/souveraine
Watch
1
0
Fork
You've already forked souveraine
 0
Code Releases Activity

api+remote: ambient sensorium, cancel/interject verbs, SSE event mirror

Browse source 
SendMessageRequest gains ambient (surface sensorium note). RemoteBackend
gains spawn_cancel_watch + spawn_interject_pump so TUI remote mode has the
same interrupt/type-while-busy semantics as local. StreamEvent is now the
full wire mirror of BackendEvent — a new engine event is a compile error
at the SSE seam, not a silent skip.
This commit is contained in:
Fimeg 2026-07-12 14:56:48 -04:00
commit 165d814492
8 changed files with 439 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

158
src/api/handlers.rs
View file

@ -154,6 +154,26 @@ pub async fn stream_messages(
message: format!("Conversation {} not found", conversation_id),
})))?;
// Ambient context first — the room she is being spoken to in. A system
// note, same register as interjections, so it reads as perception rather
// than instruction.
if let Some(ambient) = request.ambient.as_deref().map(str::trim).filter(|a| !a.is_empty()) {
let stamp = chrono::Local::now().format("%H:%M");
let note = crate::core::session::ConversationMessage {
role: crate::core::session::MessageRole::System,
blocks: vec![crate::core::session::ContentBlock::Text {
text: format!("[ambient at {stamp} — what the surface senses]\n{ambient}"),
}],
usage: None,
timestamp: Some(chrono::Utc::now()),
};
server.sessions.add_message(&conversation_id, note)
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, Json(ErrorResponse {
error: "message_store_failed".to_string(),
message: e.to_string(),
})))?;
}
// Convert API messages to ConversationMessages and add to session
for msg in &request.messages {
let conv_msg = msg.to_conversation_message();
@ -194,9 +214,16 @@ async fn handle_conversation_stream(
use tokio_util::sync::CancellationToken;
let event_bus = server.event_bus.clone();
let empty_interject: crate::backend::InterjectionQueue =
std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
// Turn backchannel: fresh cancel token each turn, persistent
// interjection queue. POST .../cancel and .../interject reach in here.
let signals = server.turn_signals
.entry(conversation_id.clone())
.or_default();
let cancel = CancellationToken::new();
*signals.cancel.lock().unwrap() = cancel.clone();
let interject = signals.interject.clone();
drop(signals); // release the DashMap ref before the turn runs
let (be_tx, mut be_rx) = mpsc::channel::<anyhow::Result<BackendEvent>>(64);
let server_clone = server.clone();
@ -205,63 +232,106 @@ async fn handle_conversation_stream(
tokio::spawn(async move {
let _ = crate::server::turn::run_turn(
server_clone, conv, &be_tx, event_bus,
cancel, empty_interject,
cancel, interject,
).await;
});
// Drain the turn's BackendEvent stream and map to SSE events.
// run_turn already handles: message storage, surficing injection,
// EventBus dispatch, N+1 pass — none of that is needed here.
// Drain the turn's BackendEvent stream. run_turn already handles message
// storage, surfacing injection, EventBus dispatch, N+1 pass. Every event
// crosses the boundary via the exhaustive From impl — no silent skips;
// the full personification channel (subconscious, interstitials,
// atmosphere, strain, pressure) reaches every surface.
while let Some(result) = be_rx.recv().await {
let be = match result {
Ok(be) => be,
Err(_) => break,
};
let sse_event = match be {
BackendEvent::Token(content) => StreamEvent::AssistantMessage { content },
BackendEvent::Reasoning(content) => StreamEvent::ReasoningMessage { content },
BackendEvent::Surfacing { source, content, priority } => {
StreamEvent::Surfacing { source, content, priority }
}
BackendEvent::Reflection(content) => StreamEvent::Reflection { content },
BackendEvent::Archivist { synthesis, pressure } => {
StreamEvent::Archivist { synthesis, pressure }
}
BackendEvent::CompactionWarning { pressure, tier } => {
StreamEvent::Archivist {
synthesis: format!("compaction warning tier {} at {:.0}%", tier, pressure * 100.0),
pressure,
}
}
BackendEvent::ToolCall { id, name, arguments, .. } => {
StreamEvent::ToolCallMessage {
tool_call: ToolCall {
id,
function: ToolFunction { name, arguments },
},
}
}
BackendEvent::ToolResult { output, is_error, .. } => {
StreamEvent::ToolReturnMessage {
tool_return: ToolReturn {
status: if is_error { "error".into() } else { "success".into() },
output,
},
}
}
// Silently skip events that have no SSE counterpart
_ => continue,
};
if tx.send(sse_event).await.is_err() {
if tx.send(StreamEvent::from(be)).await.is_err() {
break;
}
}
let _ = tx.send(StreamEvent::Ping).await;
Ok(())
}
/// GET /v1/conversations/:id/messages — full transcript backfill for resume.
/// Returns the session's `ConversationMessage`s verbatim; `RemoteBackend`
/// and any surface use this to restore a conversation after restart.
pub async fn get_conversation_messages(
State(server): State<Arc<SouveraineServer>>,
Path(id): Path<String>,
) -> Result<Json<Vec<crate::core::session::ConversationMessage>>, ApiError> {
let session = server.sessions.get(&id)
.ok_or_else(|| (StatusCode::NOT_FOUND, Json(ErrorResponse {
error: "conversation_not_found".to_string(),
message: format!("Conversation {} not found", id),
})))?;
Ok(Json(session.messages.clone()))
}
/// POST /v1/conversations/:id/fork — deep-clone into a fresh conversation.
/// The `/btw` side-quest verb, now reachable from any surface.
pub async fn fork_conversation(
State(server): State<Arc<SouveraineServer>>,
Path(id): Path<String>,
) -> Result<(StatusCode, Json<Conversation>), ApiError> {
let forked_id = server.sessions.fork(&id)
.map_err(|e| (StatusCode::NOT_FOUND, Json(ErrorResponse {
error: "fork_failed".to_string(),
message: e.to_string(),
})))?;
let session = server.sessions.get(&forked_id)
.ok_or_else(|| (StatusCode::INTERNAL_SERVER_ERROR, Json(ErrorResponse {
error: "fork_failed".to_string(),
message: "forked session vanished".to_string(),
})))?;
Ok((StatusCode::CREATED, Json(Conversation {
id: forked_id.clone(),
agent_id: session.agent_id.clone(),
created_at: session.created_at,
updated_at: Some(session.updated_at),
})))
}
/// POST /v1/conversations/:id/cancel — interrupt the turn in flight.
/// The token is a signal, not enforcement: the engine lets the current
/// tool finish, stops new LLM calls, and commits partial text with the
/// `*[raised hand]*` marker so she reads the interrupt in her own history.
pub async fn cancel_turn(
State(server): State<Arc<SouveraineServer>>,
Path(id): Path<String>,
) -> Result<StatusCode, ApiError> {
match server.turn_signals.get(&id) {
Some(signals) => {
signals.cancel.lock().unwrap().cancel();
Ok(StatusCode::ACCEPTED)
}
None => Err((StatusCode::NOT_FOUND, Json(ErrorResponse {
error: "no_active_turn".to_string(),
message: format!("No turn signals for conversation {}", id),
}))),
}
}
/// POST /v1/conversations/:id/interject — slip a note into the running turn.
/// Drained between LLM rounds and read as `[interjected at …]` system notes.
/// If she's idle, the note waits in the queue and lands at next turn start.
pub async fn interject(
State(server): State<Arc<SouveraineServer>>,
Path(id): Path<String>,
Json(request): Json<InterjectRequest>,
) -> Result<StatusCode, ApiError> {
if request.text.trim().is_empty() {
return Err((StatusCode::BAD_REQUEST, Json(ErrorResponse {
error: "empty_interjection".to_string(),
message: "text must be non-empty".to_string(),
})));
}
let signals = server.turn_signals.entry(id).or_default();
signals.interject.lock().unwrap().push(request.text);
Ok(StatusCode::ACCEPTED)
}
// ─── Memory (memfs HTTP write path) ───────────────────────────────────────
//
// Routes:

8
src/api/mod.rs
View file

@ -39,7 +39,13 @@ pub fn create_routes(state: Arc<SouveraineServer>) -> Router {
// Protected conversation routes — require bearer token for the conversation's agent.
let conversation_routes = Router::new()
.route("/v1/conversations/:id", get(handlers::get_conversation))
.route("/v1/conversations/:id/messages", post(handlers::stream_messages))
.route(
"/v1/conversations/:id/messages",
get(handlers::get_conversation_messages).post(handlers::stream_messages),
)
.route("/v1/conversations/:id/fork", post(handlers::fork_conversation))
.route("/v1/conversations/:id/cancel", post(handlers::cancel_turn))
.route("/v1/conversations/:id/interject", post(handlers::interject))
.route_layer(middleware::from_fn_with_state(
state.clone(),
auth::require_conversation_token,

169
src/api/models.rs
View file

@ -214,9 +214,26 @@ pub struct SendMessageRequest {
pub messages: Vec<Message>,
#[serde(default)]
pub stream: bool,
/// Ambient context from the sending surface — what the environment
/// senses at the moment of speaking: active window, open apps, cursor
/// position, device sensors. Injected as a system note before the user
/// message so the agent perceives the room she is being spoken to in.
#[serde(default)]
pub ambient: Option<String>,
}
#[derive(Debug, Clone, Serialize)]
#[derive(Debug, Deserialize)]
pub struct InterjectRequest {
pub text: String,
}
/// The full wire mirror of [`crate::backend::BackendEvent`].
///
/// Every engine event crosses the SSE boundary — no silent skips. The
/// exhaustive `From` impls in both directions mean a new `BackendEvent`
/// variant is a compile error here, not an invisible hole in every
/// non-TUI surface. Tag values are the wire contract; never rename.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "message_type")]
pub enum StreamEvent {
#[serde(rename = "assistant_message")]
@ -224,7 +241,11 @@ pub enum StreamEvent {
#[serde(rename = "reasoning_message")]
ReasoningMessage { content: String },
#[serde(rename = "tool_call_message")]
ToolCallMessage { tool_call: ToolCall },
ToolCallMessage {
tool_call: ToolCall,
#[serde(default)]
round: u32,
},
#[serde(rename = "tool_return_message")]
ToolReturnMessage { tool_return: ToolReturn },
#[serde(rename = "souveraine_surfacing")]
@ -233,6 +254,38 @@ pub enum StreamEvent {
Reflection { content: String },
#[serde(rename = "souveraine_archivist")]
Archivist { synthesis: String, pressure: f32 },
#[serde(rename = "compaction_warning")]
CompactionWarning { pressure: f32, tier: u8 },
#[serde(rename = "context_pressure")]
ContextPressure { pressure: f32, tokens: usize },
#[serde(rename = "inference_strain")]
InferenceStrain { attempt: u32, status: u16, model: String },
#[serde(rename = "schedule_active")]
ScheduleActive { name: String },
#[serde(rename = "schedule_complete")]
ScheduleComplete { name: String, silent: bool },
#[serde(rename = "subconscious_token")]
SubconsciousToken { content: String },
#[serde(rename = "subconscious_tool_call")]
SubconsciousToolCall { name: String, arguments: String },
#[serde(rename = "subconscious_tool_result")]
SubconsciousToolResult { name: String, output: String, is_error: bool },
#[serde(rename = "subconscious_halt")]
SubconsciousHalt { reason: String, severity: String },
#[serde(rename = "subconscious_pass")]
SubconsciousPass { active: bool },
#[serde(rename = "atmosphere")]
Atmosphere { preset: String },
#[serde(rename = "itinerary")]
Itinerary { route: String },
#[serde(rename = "outfit")]
Outfit { name: String },
#[serde(rename = "interstitial")]
Interstitial { text: String, register: crate::backend::Register },
#[serde(rename = "primary_complete")]
PrimaryComplete,
#[serde(rename = "done")]
Done,
#[serde(rename = "ping")]
Ping,
}
@ -247,15 +300,125 @@ impl StreamEvent {
StreamEvent::Surfacing { .. } => "souveraine_surfacing",
StreamEvent::Reflection { .. } => "souveraine_reflection",
StreamEvent::Archivist { .. } => "souveraine_archivist",
StreamEvent::CompactionWarning { .. } => "compaction_warning",
StreamEvent::ContextPressure { .. } => "context_pressure",
StreamEvent::InferenceStrain { .. } => "inference_strain",
StreamEvent::ScheduleActive { .. } => "schedule_active",
StreamEvent::ScheduleComplete { .. } => "schedule_complete",
StreamEvent::SubconsciousToken { .. } => "subconscious_token",
StreamEvent::SubconsciousToolCall { .. } => "subconscious_tool_call",
StreamEvent::SubconsciousToolResult { .. } => "subconscious_tool_result",
StreamEvent::SubconsciousHalt { .. } => "subconscious_halt",
StreamEvent::SubconsciousPass { .. } => "subconscious_pass",
StreamEvent::Atmosphere { .. } => "atmosphere",
StreamEvent::Itinerary { .. } => "itinerary",
StreamEvent::Outfit { .. } => "outfit",
StreamEvent::Interstitial { .. } => "interstitial",
StreamEvent::PrimaryComplete => "primary_complete",
StreamEvent::Done => "done",
StreamEvent::Ping => "ping",
}
}
}
#[derive(Debug, Clone, Serialize)]
impl From<crate::backend::BackendEvent> for StreamEvent {
fn from(be: crate::backend::BackendEvent) -> Self {
use crate::backend::BackendEvent as BE;
match be {
BE::Token(content) => Self::AssistantMessage { content },
BE::Reasoning(content) => Self::ReasoningMessage { content },
BE::Surfacing { source, content, priority } => Self::Surfacing { source, content, priority },
BE::Reflection(content) => Self::Reflection { content },
BE::Archivist { synthesis, pressure } => Self::Archivist { synthesis, pressure },
BE::CompactionWarning { pressure, tier } => Self::CompactionWarning { pressure, tier },
BE::ContextPressure(pressure, tokens) => Self::ContextPressure { pressure, tokens },
BE::InferenceStrain { attempt, status, model } => Self::InferenceStrain { attempt, status, model },
BE::ScheduleActive { name } => Self::ScheduleActive { name },
BE::ScheduleComplete { name, silent } => Self::ScheduleComplete { name, silent },
BE::ToolCall { id, name, arguments, round } => Self::ToolCallMessage {
tool_call: ToolCall { id, function: ToolFunction { name, arguments } },
round,
},
BE::ToolResult { id, name, output, is_error } => Self::ToolReturnMessage {
tool_return: ToolReturn {
status: if is_error { "error".into() } else { "success".into() },
output,
id,
name,
},
},
BE::SubconsciousToken(content) => Self::SubconsciousToken { content },
BE::SubconsciousToolCall { name, arguments } => Self::SubconsciousToolCall { name, arguments },
BE::SubconsciousToolResult { name, output, is_error } => {
Self::SubconsciousToolResult { name, output, is_error }
}
BE::SubconsciousHalt { reason, severity } => Self::SubconsciousHalt { reason, severity },
BE::SubconsciousPass(active) => Self::SubconsciousPass { active },
BE::Atmosphere(preset) => Self::Atmosphere { preset },
BE::Itinerary(route) => Self::Itinerary { route },
BE::Outfit(name) => Self::Outfit { name },
BE::Interstitial { text, register } => Self::Interstitial { text, register },
BE::Keepalive => Self::Ping,
BE::PrimaryComplete => Self::PrimaryComplete,
BE::Done => Self::Done,
}
}
}
impl From<StreamEvent> for crate::backend::BackendEvent {
fn from(se: StreamEvent) -> Self {
use crate::backend::BackendEvent as BE;
match se {
StreamEvent::AssistantMessage { content } => BE::Token(content),
StreamEvent::ReasoningMessage { content } => BE::Reasoning(content),
StreamEvent::Surfacing { source, content, priority } => BE::Surfacing { source, content, priority },
StreamEvent::Reflection { content } => BE::Reflection(content),
StreamEvent::Archivist { synthesis, pressure } => BE::Archivist { synthesis, pressure },
StreamEvent::CompactionWarning { pressure, tier } => BE::CompactionWarning { pressure, tier },
StreamEvent::ContextPressure { pressure, tokens } => BE::ContextPressure(pressure, tokens),
StreamEvent::InferenceStrain { attempt, status, model } => BE::InferenceStrain { attempt, status, model },
StreamEvent::ScheduleActive { name } => BE::ScheduleActive { name },
StreamEvent::ScheduleComplete { name, silent } => BE::ScheduleComplete { name, silent },
StreamEvent::ToolCallMessage { tool_call, round } => BE::ToolCall {
id: tool_call.id,
name: tool_call.function.name,
arguments: tool_call.function.arguments,
round,
},
StreamEvent::ToolReturnMessage { tool_return } => BE::ToolResult {
id: tool_return.id,
name: tool_return.name,
is_error: tool_return.status == "error",
output: tool_return.output,
},
StreamEvent::SubconsciousToken { content } => BE::SubconsciousToken(content),
StreamEvent::SubconsciousToolCall { name, arguments } => BE::SubconsciousToolCall { name, arguments },
StreamEvent::SubconsciousToolResult { name, output, is_error } => {
BE::SubconsciousToolResult { name, output, is_error }
}
StreamEvent::SubconsciousHalt { reason, severity } => BE::SubconsciousHalt { reason, severity },
StreamEvent::SubconsciousPass { active } => BE::SubconsciousPass(active),
StreamEvent::Atmosphere { preset } => BE::Atmosphere(preset),
StreamEvent::Itinerary { route } => BE::Itinerary(route),
StreamEvent::Outfit { name } => BE::Outfit(name),
StreamEvent::Interstitial { text, register } => BE::Interstitial { text, register },
StreamEvent::Ping => BE::Keepalive,
StreamEvent::PrimaryComplete => BE::PrimaryComplete,
StreamEvent::Done => BE::Done,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ToolReturn {
pub status: String,
pub output: String,
/// Tool-call id this return answers. Empty on frames from pre-widening servers.
#[serde(default)]
pub id: String,
/// Tool name. Empty on frames from pre-widening servers.
#[serde(default)]
pub name: String,
}
#[derive(Debug, Serialize)]

3
src/backend/mod.rs
View file

@ -52,7 +52,8 @@ pub struct ConversationInfo {
/// kind of utterance. A few words attached to a gesture ("checking the
/// ledger…") is ambient — a *cenno*. A full paragraph of reasoning mid-turn
/// is her actual voice and deserves to read as such, not as a quiet aside.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Register {
/// Short ambient aside attached to tool work. Terse, quiet.
Cenno,

146
src/backend/remote.rs
View file

@ -4,7 +4,6 @@ use anyhow::{Context, Result};
use async_trait::async_trait;
use futures::stream::{BoxStream, StreamExt};
use serde::Deserialize;
use serde_json::Value;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
@ -71,6 +70,66 @@ impl RemoteBackend {
fn url(&self, path: &str) -> String {
format!("{}{}", self.base_url, path)
}
/// Watch a local cancel token and relay it as POST .../cancel.
/// Esc in the TUI fires the same interrupt semantics as local mode.
fn spawn_cancel_watch(&self, conversation_id: &str, cancel: tokio_util::sync::CancellationToken) {
let client = self.client.clone();
let url = self.url(&format!("/v1/conversations/{}/cancel", conversation_id));
let token = self.token.clone();
tokio::spawn(async move {
tokio::select! {
_ = cancel.cancelled() => {
let mut req = client.post(url);
if let Some(t) = token {
req = req.bearer_auth(t);
}
let _ = req.send().await;
}
// ponytail: task self-reaps after an hour so uncancelled
// turns don't accumulate watchers; scope to stream lifetime
// if turns ever legitimately exceed this
_ = tokio::time::sleep(Duration::from_secs(3600)) => {}
}
});
}
/// Relay locally-queued interjections to the server mid-turn.
/// ponytail: 250ms polling pump bounded by cancel/1h; replace with a
/// notify-driven channel if the queue ever grows a waker.
fn spawn_interject_pump(
&self,
conversation_id: &str,
interject: super::InterjectionQueue,
cancel: tokio_util::sync::CancellationToken,
) {
let client = self.client.clone();
let url = self.url(&format!("/v1/conversations/{}/interject", conversation_id));
let token = self.token.clone();
tokio::spawn(async move {
let deadline = tokio::time::Instant::now() + Duration::from_secs(3600);
loop {
let drained: Vec<String> = {
let mut q = interject.lock().unwrap();
q.drain(..).collect()
};
for text in drained {
let mut req = client.post(&url).json(&serde_json::json!({ "text": text }));
if let Some(ref t) = token {
req = req.bearer_auth(t);
}
let _ = req.send().await;
}
tokio::select! {
_ = cancel.cancelled() => break,
_ = tokio::time::sleep(Duration::from_millis(250)) => {}
}
if tokio::time::Instant::now() >= deadline {
break;
}
}
});
}
}
#[async_trait]
@ -156,17 +215,37 @@ impl Backend for RemoteBackend {
&self,
conversation_id: &str,
) -> Result<Vec<crate::core::session::ConversationMessage>> {
let url = format!("/v1/conversations/{}", conversation_id);
let url = format!("/v1/conversations/{}/messages", conversation_id);
let resp = self
.auth_req(self.client.get(self.url(&url)))
.send()
.await
.context("GET /v1/conversations/:id")?
.context("GET /v1/conversations/:id/messages")?
.error_for_status()?;
let messages: Vec<crate::core::session::ConversationMessage> = resp.json().await?;
Ok(messages)
}
async fn fork_conversation(
&self,
_agent_id: &str,
source_conversation_id: &str,
) -> Result<String> {
#[derive(Deserialize)]
struct Wire {
id: String,
}
let url = format!("/v1/conversations/{}/fork", source_conversation_id);
let resp = self
.auth_req(self.client.post(self.url(&url)))
.send()
.await
.context("POST /v1/conversations/:id/fork")?
.error_for_status()?;
let conv: Wire = resp.json().await?;
Ok(conv.id)
}
async fn ensure_conversation(&self, agent_id: &str) -> Result<String> {
#[derive(Deserialize)]
struct Wire {
@ -199,6 +278,9 @@ impl Backend for RemoteBackend {
self.client
.post(self.url(&format!("/v1/conversations/{}/messages", conversation_id))),
)
// The client-wide 60s timeout covers the whole body — it would
// sever any turn longer than a minute. Turns run long by design.
.timeout(Duration::from_secs(3600))
.json(&body)
.send()
.await
@ -246,6 +328,28 @@ impl Backend for RemoteBackend {
Ok(ReceiverStream::new(rx).boxed())
}
async fn send_with_cancel(
&self,
conversation_id: &str,
text: &str,
cancel: tokio_util::sync::CancellationToken,
) -> Result<BoxStream<'static, Result<BackendEvent>>> {
self.spawn_cancel_watch(conversation_id, cancel);
self.send(conversation_id, text).await
}
async fn send_with_signals(
&self,
conversation_id: &str,
text: &str,
cancel: tokio_util::sync::CancellationToken,
interject: super::InterjectionQueue,
) -> Result<BoxStream<'static, Result<BackendEvent>>> {
self.spawn_cancel_watch(conversation_id, cancel.clone());
self.spawn_interject_pump(conversation_id, interject, cancel);
self.send(conversation_id, text).await
}
}
/// Index of the start of the SSE frame delimiter (`\n\n` or `\r\n\r\n`).
@ -273,40 +377,22 @@ fn find_frame_end(buf: &[u8]) -> Option<usize> {
fn parse_frame(bytes: &[u8]) -> Option<BackendEvent> {
let s = std::str::from_utf8(bytes).ok()?;
let mut event_type: Option<String> = None;
let mut data = String::new();
for line in s.split('\n') {
let line = line.trim_end_matches('\r');
if let Some(v) = line.strip_prefix("event:") {
event_type = Some(v.trim().to_string());
} else if let Some(v) = line.strip_prefix("data:") {
if let Some(v) = line.strip_prefix("data:") {
if !data.is_empty() {
data.push('\n');
}
data.push_str(v.trim_start());
}
}
let event_type = event_type?;
let json: Value = serde_json::from_str(&data).ok()?;
map_event(&event_type, &json)
}
fn map_event(event_type: &str, v: &Value) -> Option<BackendEvent> {
let s = |k: &str| v.get(k).and_then(|x| x.as_str()).map(String::from);
Some(match event_type {
"message" => BackendEvent::Token(s("content")?),
"reasoning" => BackendEvent::Reasoning(s("content")?),
"souveraine_surfacing" => BackendEvent::Surfacing {
source: s("source").unwrap_or_default(),
content: s("content").unwrap_or_default(),
priority: s("priority").unwrap_or_default(),
},
"souveraine_reflection" => BackendEvent::Reflection(s("content").unwrap_or_default()),
"souveraine_archivist" => BackendEvent::Archivist {
synthesis: s("synthesis").unwrap_or_default(),
pressure: v.get("pressure").and_then(|x| x.as_f64()).unwrap_or(0.0) as f32,
},
"ping" => BackendEvent::Keepalive,
_ => return None,
})
if data.is_empty() {
return None;
}
// The data payload is self-describing (message_type tag) — the SSE
// event: line is redundant. Full round-trip through the wire mirror:
// every event the server emits lands here as its BackendEvent self.
let ev: crate::api::models::StreamEvent = serde_json::from_str(&data).ok()?;
Some(ev.into())
}

24
src/server/mod.rs
View file

@ -66,6 +66,12 @@ pub struct SouveraineServer {
/// This instance's Ed25519 public key hex — used to filter self-announcements
/// from the device registry. Loaded at construction; None if seed unavailable.
pub local_seed_id: Option<String>,
/// Per-conversation turn backchannel. Holds the cancel token for the
/// turn in flight and a persistent interjection queue, so any surface
/// can interrupt (`POST /v1/conversations/:id/cancel`) or slip a note
/// mid-turn (`POST .../interject`). The queue survives between turns —
/// an interjection landed while she's idle is read at the next turn start.
pub turn_signals: Arc<dashmap::DashMap<String, TurnSignals>>,
/// Manages every active surface (TUI, Matrix, mobile, etc.).
/// Each sensorium runs in its own task, sharing the EventBus.
/// Register via `register_sensorium()`; `shutdown()` cancels all.
@ -84,6 +90,23 @@ pub struct ServerConfig {
pub gitea_url: Option<String>,
}
/// Backchannel signals for one conversation's turns. The cancel token is
/// replaced at each turn start (fire-once semantics); the interjection
/// queue is shared with `run_turn`, which drains it between LLM rounds.
pub struct TurnSignals {
pub cancel: StdMutex<tokio_util::sync::CancellationToken>,
pub interject: crate::backend::InterjectionQueue,
}
impl Default for TurnSignals {
fn default() -> Self {
Self {
cancel: StdMutex::new(tokio_util::sync::CancellationToken::new()),
interject: Arc::new(StdMutex::new(Vec::new())),
}
}
}
impl SouveraineServer {
pub async fn new(mut config: ConsciousnessConfig) -> anyhow::Result<Self> {
config.normalize_providers();
@ -328,6 +351,7 @@ impl SouveraineServer {
device_registry,
summon_handler,
local_seed_id,
turn_signals: Arc::new(dashmap::DashMap::new()),
sensorium: Arc::new(Mutex::new(crate::core::sensorium::SensoriumCoordinator::new())),
surface_conversations: Arc::new(StdMutex::new(HashMap::new())),
})

5
src/ui/screens/settings/mod.rs
View file

@ -1159,6 +1159,11 @@ mod tests {
virtual_key: "vk-test".into(),
primary_model: String::new(),
timeout_secs: 30,
credential_file: None,
cc_version: None,
account_uuid: None,
device_id: None,
extra_metadata: None,
},
);
let mut s = SettingsScreen::new(&config, std::path::PathBuf::from("souveraine.toml"), buffer);

5
src/ui/settings/view.rs
View file

@ -579,6 +579,11 @@ impl SettingsView {
virtual_key: String::new(),
primary_model: String::new(),
timeout_secs: 120,
credential_file: None,
cc_version: None,
account_uuid: None,
device_id: None,
extra_metadata: None,
});
self.selected_provider_name = Some(name);
}