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

turns: survive the surface disappearing

Browse source
This commit is contained in:
Fimeg 2026-08-11 15:01:18 -04:00
commit 3aa7019eb5
8 changed files with 426 additions and 168 deletions
Download patch file Download diff file Expand all files Collapse all files

179
src/api/handlers.rs
View file

@ -283,6 +283,19 @@ pub async fn stream_messages(
})?);
}
// A turn belongs to the conversation, not to the HTTP socket that
// happened to start it. Claim it before changing history so a raced POST
// cannot leave an extra user message behind.
server.sessions.begin_turn(&conversation_id).map_err(|e| {
(
StatusCode::CONFLICT,
Json(ErrorResponse {
error: "turn_already_active".to_string(),
message: e.to_string(),
}),
)
})?;
// 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.
@ -301,34 +314,30 @@ pub async fn stream_messages(
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(),
}),
)
})?;
if let Err(e) = server.sessions.add_message(&conversation_id, note) {
let _ = server.sessions.finish_turn(&conversation_id);
return Err((
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 conv_msg in conv_messages {
server
.sessions
.add_message(&conversation_id, conv_msg)
.map_err(|e| {
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse {
error: "message_store_failed".to_string(),
message: e.to_string(),
}),
)
})?;
if let Err(e) = server.sessions.add_message(&conversation_id, conv_msg) {
let _ = server.sessions.finish_turn(&conversation_id);
return Err((
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse {
error: "message_store_failed".to_string(),
message: e.to_string(),
}),
));
}
}
let (tx, rx) = mpsc::channel(100);
@ -387,21 +396,33 @@ async fn handle_conversation_stream(
// crosses the boundary via the exhaustive From impl — no silent skips;
// the full personification channel (subconscious, interstitials,
// atmosphere, strain, pressure) reaches every surface.
let mut client_connected = true;
while let Some(result) = be_rx.recv().await {
let be = match result {
Ok(be) => be,
Err(e) => {
eprintln!("Turn stream error ({conversation_id}): {e:#}");
let _ = tx
.send(StreamEvent::Error {
message: format!("{e:#}"),
})
.await;
let event = StreamEvent::Error {
message: format!("{e:#}"),
};
let _ = server
.sessions
.publish_turn_event(&conversation_id, event.clone());
if client_connected && tx.send(event).await.is_err() {
client_connected = false;
}
break;
}
};
if tx.send(StreamEvent::from(be)).await.is_err() {
break;
let event = StreamEvent::from(be);
let _ = server
.sessions
.publish_turn_event(&conversation_id, event.clone());
// Losing one surface must not stop draining the engine channel. If it
// did, the next engine send would fail and turn a display reload into
// a false user interrupt.
if client_connected && tx.send(event).await.is_err() {
client_connected = false;
}
}
// The channel closing means run_turn returned; a turn that died before
@ -409,15 +430,101 @@ async fn handle_conversation_stream(
// silently-ended stream (an empty reply reads as the agent going mute).
if let Ok(Err(e)) = turn.await {
eprintln!("Turn failed ({conversation_id}): {e:#}");
let _ = tx
.send(StreamEvent::Error {
message: format!("{e:#}"),
})
.await;
let event = StreamEvent::Error {
message: format!("{e:#}"),
};
let _ = server
.sessions
.publish_turn_event(&conversation_id, event.clone());
if client_connected {
let _ = tx.send(event).await;
}
}
let done = StreamEvent::Done;
let _ = server
.sessions
.publish_turn_event(&conversation_id, done.clone());
if client_connected {
let _ = tx.send(done).await;
}
server.turn_signals.remove(&conversation_id);
let _ = server.sessions.finish_turn(&conversation_id);
Ok(())
}
/// GET /v1/conversations/:id/events — replay and follow the turn currently in
/// flight. This is deliberately separate from POST /messages: reconnecting a
/// surface observes the existing turn and can never accidentally start one.
pub async fn stream_active_turn(
State(server): State<Arc<SouveraineServer>>,
Path(conversation_id): Path<String>,
) -> Result<
Sse<impl futures::Stream<Item = Result<axum::response::sse::Event, std::convert::Infallible>>>,
ApiError,
> {
let Some((replay, mut live)) = server
.sessions
.subscribe_active_turn(&conversation_id)
.map_err(|e| {
(
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: "conversation_not_found".to_string(),
message: e.to_string(),
}),
)
})?
else {
return Err((
StatusCode::CONFLICT,
Json(ErrorResponse {
error: "no_active_turn".to_string(),
message: format!("Conversation {conversation_id} has no active turn"),
}),
));
};
let (tx, rx) = mpsc::channel(256);
tokio::spawn(async move {
for event in replay {
let done = matches!(event, StreamEvent::Done);
if tx.send(event).await.is_err() {
return;
}
if done {
return;
}
}
loop {
match live.recv().await {
Ok(event) => {
let done = matches!(event, StreamEvent::Done);
if tx.send(event).await.is_err() || done {
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
let _ = tx
.send(StreamEvent::Error {
message: format!("active turn replay lagged by {n} events; reconnect"),
})
.await;
break;
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
});
let stream = ReceiverStream::new(rx).map(|event| {
let json = serde_json::to_string(&event).unwrap_or_else(|_| "{}".to_string());
Ok(axum::response::sse::Event::default()
.event(event.message_type())
.data(json))
});
Ok(Sse::new(stream))
}
/// 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.

4
src/api/mod.rs
View file

@ -55,6 +55,10 @@ pub fn create_routes(state: Arc<SouveraineServer>) -> Router {
"/v1/conversations/:id/messages",
get(handlers::get_conversation_messages).post(handlers::stream_messages),
)
.route(
"/v1/conversations/:id/events",
get(handlers::stream_active_turn),
)
.route(
"/v1/conversations/:id/fork",
post(handlers::fork_conversation),

148
src/server/session_manager.rs
View file

@ -35,6 +35,11 @@ pub struct Session {
pub last_n25: DateTime<Utc>,
pub context_pressure: f32,
pub event_sender: Sender<StreamEvent>,
/// Events belonging to the turn currently in flight. Unlike the broadcast
/// channel, this survives a surface disconnect so a reattached panel can
/// replay the turn from its beginning before following the live tail.
pub active_turn_events: Vec<StreamEvent>,
pub turn_active: bool,
}
impl SessionManager {
@ -68,6 +73,8 @@ impl SessionManager {
last_n25: Utc::now(),
context_pressure: 0.0,
event_sender: sender,
active_turn_events: Vec::new(),
turn_active: false,
};
self.sessions.insert(conversation_id.clone(), session);
@ -133,6 +140,8 @@ impl SessionManager {
last_n25: Utc::now(),
context_pressure: 0.0,
event_sender: sender,
active_turn_events: Vec::new(),
turn_active: false,
};
self.sessions.insert(conversation_id.clone(), session);
@ -218,6 +227,78 @@ impl SessionManager {
Ok(())
}
/// Claim the conversation for one turn and reset its replay journal.
pub fn begin_turn(&self, conversation_id: &str) -> anyhow::Result<()> {
let mut session = self
.sessions
.get_mut(conversation_id)
.ok_or_else(|| anyhow::anyhow!("Session not found: {}", conversation_id))?;
if session.turn_active {
anyhow::bail!("Conversation already has an active turn");
}
session.active_turn_events.clear();
session.turn_active = true;
Ok(())
}
/// Record an event for replay, then deliver it to every live follower.
/// Consecutive token frames are coalesced in the journal only; subscribers
/// still receive the original cadence in real time.
pub fn publish_turn_event(
&self,
conversation_id: &str,
event: StreamEvent,
) -> anyhow::Result<()> {
let mut session = self
.sessions
.get_mut(conversation_id)
.ok_or_else(|| anyhow::anyhow!("Session not found: {}", conversation_id))?;
if session.turn_active {
match (session.active_turn_events.last_mut(), &event) {
(
Some(StreamEvent::AssistantMessage { content: prior }),
StreamEvent::AssistantMessage { content },
) => prior.push_str(content),
(
Some(StreamEvent::SubconsciousToken { content: prior }),
StreamEvent::SubconsciousToken { content },
) => prior.push_str(content),
_ => session.active_turn_events.push(event.clone()),
}
}
let _ = session.event_sender.send(event);
Ok(())
}
/// Atomically take a replay snapshot and subscribe to everything that
/// follows it. Holding the session shard across both operations prevents
/// the usual snapshot/subscribe race.
pub fn subscribe_active_turn(
&self,
conversation_id: &str,
) -> anyhow::Result<Option<(Vec<StreamEvent>, broadcast::Receiver<StreamEvent>)>> {
let session = self
.sessions
.get(conversation_id)
.ok_or_else(|| anyhow::anyhow!("Session not found: {}", conversation_id))?;
if !session.turn_active {
return Ok(None);
}
let replay = session.active_turn_events.clone();
let receiver = session.event_sender.subscribe();
Ok(Some((replay, receiver)))
}
pub fn finish_turn(&self, conversation_id: &str) -> anyhow::Result<()> {
let mut session = self
.sessions
.get_mut(conversation_id)
.ok_or_else(|| anyhow::anyhow!("Session not found: {}", conversation_id))?;
session.turn_active = false;
session.active_turn_events.clear();
Ok(())
}
pub fn update_pressure(&self, conversation_id: &str, pressure: f32) -> anyhow::Result<()> {
let mut session = self
.sessions
@ -309,6 +390,8 @@ impl SessionManager {
last_n25: Utc::now(),
context_pressure: 0.0,
event_sender: sender,
active_turn_events: Vec::new(),
turn_active: false,
};
self.sessions.insert(forked_id.clone(), session);
@ -335,6 +418,71 @@ impl SessionManager {
mod tests {
use super::*;
#[tokio::test]
async fn active_turn_can_be_replayed_and_followed_after_disconnect() {
let sessions = SessionManager::new();
let conversation_id = sessions.create("agent-replay");
assert!(sessions
.subscribe_active_turn(&conversation_id)
.unwrap()
.is_none());
sessions.begin_turn(&conversation_id).unwrap();
assert!(sessions.begin_turn(&conversation_id).is_err());
sessions
.publish_turn_event(
&conversation_id,
StreamEvent::AssistantMessage {
content: "still ".into(),
},
)
.unwrap();
sessions
.publish_turn_event(
&conversation_id,
StreamEvent::AssistantMessage {
content: "working".into(),
},
)
.unwrap();
let (replay, mut live) = sessions
.subscribe_active_turn(&conversation_id)
.unwrap()
.unwrap();
assert!(matches!(
replay.as_slice(),
[StreamEvent::AssistantMessage { content }] if content == "still working"
));
sessions
.publish_turn_event(
&conversation_id,
StreamEvent::ToolCallMessage {
tool_call: crate::api::models::ToolCall {
id: "call-1".into(),
function: crate::api::models::ToolFunction {
name: "read".into(),
arguments: "{}".into(),
},
},
round: 1,
},
)
.unwrap();
assert!(matches!(
live.recv().await.unwrap(),
StreamEvent::ToolCallMessage { round: 1, .. }
));
sessions.finish_turn(&conversation_id).unwrap();
assert!(sessions
.subscribe_active_turn(&conversation_id)
.unwrap()
.is_none());
}
#[tokio::test]
async fn list_hydration_restores_persisted_agent_conversations() {
let temp = tempfile::tempdir().unwrap();

56
src/server/turn.rs
View file

@ -284,6 +284,10 @@ pub(crate) async fn run_turn(
let mut messages = initial_messages;
let mut tool_round = 0u32;
let mut final_content: String = String::new();
// The visible anatomy of this turn, persisted with the final assistant
// message so resume can reconstruct reasoning and tool cards instead of
// recovering only the last paragraph of prose.
let mut persisted_blocks: Vec<ContentBlock> = Vec::new();
// Replies the agent finished before a raced interjection reopened the
// turn. The turn commits one assistant message, so without carrying these
// the session record keeps only the last round's text.
@ -454,6 +458,9 @@ pub(crate) async fn run_turn(
// Emit reasoning trace if present
if let Some(reasoning) = &response.reasoning {
persisted_blocks.push(ContentBlock::Reasoning {
reasoning: reasoning.clone(),
});
dispatcher.emit_reasoning(reasoning);
let _ = tx
.send(Ok(BackendEvent::Reasoning(reasoning.clone())))
@ -615,6 +622,9 @@ pub(crate) async fn run_turn(
// tool calls must not produce an empty `⟡` gap line.
let narration = response.content.trim();
if !narration.is_empty() {
persisted_blocks.push(ContentBlock::Text {
text: narration.to_string(),
});
let cfg = server.app_config.read().await;
if cfg.tui.show_interstitial {
// Classify by length: a brief aside is a cenno, a full
@ -637,8 +647,25 @@ pub(crate) async fn run_turn(
// Execute each tool and stream results back — now with per-agent context
for tc in &response.tool_calls {
let input_str = tc.arguments.to_string();
persisted_blocks.push(ContentBlock::ToolUse {
id: tc.id.clone(),
name: tc.name.clone(),
input: input_str.clone(),
});
dispatcher.emit_tool_start(&tc.name, &tc.id);
dispatcher.emit_tool_call(&tc.name, &tc.id, &tc.arguments);
// A tool card must enter its running state before execution. The
// previous ordering emitted call and return back-to-back only
// after the tool had completed, so the surface could never show
// what the agent was doing mid-call.
let _ = tx
.send(Ok(BackendEvent::ToolCall {
id: tc.id.clone(),
name: tc.name.clone(),
arguments: input_str.clone(),
round: tool_round,
}))
.await;
let result =
crate::core::tools::execute_tool_with_context(&tc.name, &input_str, &tool_ctx)
.await;
@ -649,6 +676,12 @@ pub(crate) async fn run_turn(
} else {
result.output
};
persisted_blocks.push(ContentBlock::ToolResult {
tool_use_id: tc.id.clone(),
tool_name: tc.name.clone(),
output: output.clone(),
is_error: result.is_error,
});
// Accumulate for checkpoint (capture output before moving).
let snippet = output.chars().take(120).collect::<String>();
@ -659,16 +692,7 @@ pub(crate) async fn run_turn(
result_snippet: snippet,
});
// Emit structured ToolCall + ToolResult events for the TUI to render
// as cards (chat.rs subscribes). The old Token-text path is kept off.
let _ = tx
.send(Ok(BackendEvent::ToolCall {
id: tc.id.clone(),
name: tc.name.clone(),
arguments: input_str.clone(),
round: tool_round,
}))
.await;
// Complete the running card with its result.
let _ = tx
.send(Ok(BackendEvent::ToolResult {
id: tc.id.clone(),
@ -890,14 +914,14 @@ pub(crate) async fn run_turn(
// there's nothing real to commit and a bare empty assistant message
// confuses the next turn's history.
if !(halted_by_subconscious && committed_content.is_empty()) {
if !committed_content.is_empty() {
persisted_blocks.push(ContentBlock::Text {
text: committed_content.clone(),
});
}
server.sessions.add_message(
&conversation_id,
ConversationMessage::assistant_with_usage(
vec![ContentBlock::Text {
text: committed_content.clone(),
}],
Some(turn_usage),
),
ConversationMessage::assistant_with_usage(persisted_blocks, Some(turn_usage)),
)?;
}

1
surfaces/quickshell/ii-base/modules/ii/sidebarLeft/aiChat/ToolCallBlock.qml
View file

@ -1,5 +1,6 @@
pragma ComponentBehavior: Bound
import qs.services
import qs.modules.common
import qs.modules.common.widgets
import QtQuick

101
surfaces/quickshell/patches/0005-bar-mount-the-agent-island.patch
View file

@ -1,101 +0,0 @@
From a5314f17f9c9356d5e57212bd077702a4a0ba0dc Mon Sep 17 00:00:00 2001
From: Souveraine <souveraine@wiuf.net>
Date: Tue, 11 Aug 2026 14:22:01 -0400
Subject: [PATCH] bar: mount the agent island
Registers AgentSessions as a service singleton, adds its config block, mounts
Island in the Souveraine-owned UtilButtons (so it lands on both device bars
without forking BarContent twice), and adds the five new files to the deploy
manifest.
Activating change: reloads the shell.
---
surfaces/quickshell/deploy.sh | 5 +++++
surfaces/quickshell/modules/common/Config.qml | 11 ++++++++++-
surfaces/quickshell/modules/ii/bar/UtilButtons.qml | 14 ++++++++++++++
surfaces/quickshell/services/qmldir | 1 +
4 files changed, 30 insertions(+), 1 deletion(-)
diff --git a/surfaces/quickshell/deploy.sh b/surfaces/quickshell/deploy.sh
index 3b1baa1..a111b99 100755
--- a/surfaces/quickshell/deploy.sh
+++ b/surfaces/quickshell/deploy.sh
@@ -81,6 +81,11 @@ services/WallpaperAssets.qml souveraine/services/WallpaperAssets.qml
services/WallpaperDownload.qml souveraine/services/WallpaperDownload.qml
scripts/wallpaper/download_wallhaven.sh souveraine/scripts/wallpaper/download_wallhaven.sh
services/ConflictKiller.qml souveraine/services/ConflictKiller.qml
+services/AgentSessions.qml souveraine/services/AgentSessions.qml
+scripts/agent/agent-sessions.sh souveraine/scripts/agent/agent-sessions.sh
+modules/souveraine/island/Island.qml souveraine/modules/souveraine/island/Island.qml
+modules/souveraine/island/IslandExpansion.qml souveraine/modules/souveraine/island/IslandExpansion.qml
+modules/souveraine/island/qmldir souveraine/modules/souveraine/island/qmldir
services/SessionEvents.qml souveraine/services/SessionEvents.qml
services/SessiondBridge.qml souveraine/services/SessiondBridge.qml
services/SessiondPolicy.qml souveraine/services/SessiondPolicy.qml
diff --git a/surfaces/quickshell/modules/common/Config.qml b/surfaces/quickshell/modules/common/Config.qml
index bbb7ee4..31e1dbb 100644
--- a/surfaces/quickshell/modules/common/Config.qml
+++ b/surfaces/quickshell/modules/common/Config.qml
@@ -220,7 +220,16 @@ Singleton {
property string text: ""
}
}
- property JsonObject weather: JsonObject {
+ // TASK-69/70: one collector for every agent session
+ // (Souveraine, Claude Code, Codex). Replaces per-provider
+ // fetches; claudeUsage above stays for the subscription gauge
+ // until stage 3 folds its OAuth poll in here.
+ property JsonObject agentSessions: JsonObject {
+ property bool enable: true // Show the agent island in the bar
+ property int refreshInterval: 60 // seconds
+ property int windowMinutes: 1440 // How far back counts as "a session"
+ }
+ property JsonObject weather: JsonObject {
property bool enable: false
property string placementStrategy: "free" // "free", "leastBusy", "mostBusy"
property real x: 400
diff --git a/surfaces/quickshell/modules/ii/bar/UtilButtons.qml b/surfaces/quickshell/modules/ii/bar/UtilButtons.qml
index 7ba0095..d06ba12 100644
--- a/surfaces/quickshell/modules/ii/bar/UtilButtons.qml
+++ b/surfaces/quickshell/modules/ii/bar/UtilButtons.qml
@@ -2,6 +2,7 @@ import qs
import qs.services
import qs.modules.common
import qs.modules.common.widgets
+import qs.modules.souveraine.island
import QtQuick
import QtQuick.Layouts
import Quickshell
@@ -20,6 +21,19 @@ Item {
spacing: 4
anchors.centerIn: parent
+ // Agent island (TASK-70). Mounted here rather than in BarContent
+ // deliberately: UtilButtons is already a Souveraine-owned override,
+ // so the island lands on BOTH device bars without forking the 13.8K
+ // BarContent twice (ii-base and ii-phone) and carrying that
+ // divergence against the pin. The island narrows itself - dot on the
+ // phone bar, pill on the desktop - so one mount serves both.
+ Loader {
+ Layout.alignment: Qt.AlignVCenter
+ active: Config.options.bar.agentSessions?.enable ?? true
+ visible: active
+ sourceComponent: Island {}
+ }
+
Loader {
active: Config.options.bar.utilButtons.showScreenSnip
visible: Config.options.bar.utilButtons.showScreenSnip
diff --git a/surfaces/quickshell/services/qmldir b/surfaces/quickshell/services/qmldir
index 0e63101..1882c7d 100644
--- a/surfaces/quickshell/services/qmldir
+++ b/surfaces/quickshell/services/qmldir
@@ -1,4 +1,5 @@
BooruResponseData 1.0 BooruResponseData.qml
+singleton AgentSessions 1.0 AgentSessions.qml
singleton Ai 1.0 Ai.qml
singleton AppSearch 1.0 AppSearch.qml
singleton Audio 1.0 Audio.qml
--
2.55.0

57
surfaces/quickshell/services/Ai.qml
View file

@ -133,6 +133,16 @@ Singleton {
}
function onStreamEvent(event) {
// A reattached stream has no local message object: the old one
// died with the shell. Create it lazily on the first event that
// actually belongs in the assistant bubble; idle resume produces
// no empty card.
if (!root.streamingMessage && [
"assistant_message", "reasoning_message", "tool_call_message",
"tool_return_message", "interstitial"
].indexOf(event.message_type) >= 0) {
root._startStreaming();
}
root.handleStreamEvent(event);
}
@ -594,23 +604,40 @@ Singleton {
root.tokenCount.total = -1;
messages.forEach(message => {
const segments = (message.blocks ?? []).map(block => {
const segments = [];
(message.blocks ?? []).forEach(block => {
switch (block.type) {
case "text": return { type: "text", content: block.text ?? "" };
case "reasoning": return { type: "think", content: block.reasoning ?? "" };
case "tool_use": return {
type: "tool", id: block.id ?? "", name: block.name ?? "tool",
arguments: block.input ?? "", round: 0, status: "done", output: "", failed: false,
};
case "tool_result": return {
type: "tool", id: block.tool_use_id ?? "", name: block.tool_name ?? "tool",
arguments: "", round: 0, status: block.is_error ? "error" : "done",
output: block.output ?? "", failed: block.is_error ?? false,
};
case "image": return { type: "text", content: "[image]" };
default: return null;
case "text":
segments.push({ type: "text", content: block.text ?? "" });
break;
case "reasoning":
segments.push({ type: "think", content: block.reasoning ?? "" });
break;
case "tool_use":
segments.push({
type: "tool", id: block.id ?? "", name: block.name ?? "tool",
arguments: block.input ?? "", round: 0, status: "running", output: "", failed: false,
});
break;
case "tool_result": {
const resultId = String(block.tool_use_id ?? "");
const index = segments.findIndex(segment =>
segment.type === "tool" && segment.id === resultId);
const result = {
type: "tool", id: resultId, name: block.tool_name ?? "tool",
arguments: index >= 0 ? segments[index].arguments : "", round: 0,
status: block.is_error ? "error" : "done",
output: block.output ?? "", failed: block.is_error ?? false,
};
if (index >= 0) segments[index] = result;
else segments.push(result);
break;
}
}).filter(Boolean);
case "image":
segments.push({ type: "text", content: "[image]" });
break;
}
});
const content = segments.map(segment => {
if (segment.type === "tool") return `${segment.name}(${segment.arguments})\n${segment.output}`;
return segment.content;

48
surfaces/quickshell/services/Souveraine.qml
View file

@ -272,6 +272,11 @@ Singleton {
const messages = JSON.parse(text);
root.conversationId = loadConversation.requestedConversationId;
root.conversationResumed(root.currentAgentId, root.conversationId, messages);
// If the shell reloaded in the middle of a turn, the POST
// socket that started it is gone but the turn belongs to
// the server and is still running. Replay its journal and
// follow the live tail without posting another message.
Qt.callLater(root._reattachActiveTurn);
} catch (e) {
console.log("[Souveraine] Could not parse conversation transcript:", e);
}
@ -309,6 +314,49 @@ Singleton {
loadConversation.running = true;
}
Process {
id: reattachRequester
property bool receivedEvent: false
stdout: SplitParser {
onRead: data => {
if (data.length === 0 || !data.startsWith("data:")) return;
let event;
try {
event = JSON.parse(data.slice(5).trim());
} catch (e) {
console.log("[Souveraine] Unparseable replay SSE line:", data);
return;
}
reattachRequester.receivedEvent = true;
root.streamEvent(event);
}
}
onExited: exitCode => {
if (root.turnStartedAt > 0)
root.turnElapsedMs = Date.now() - root.turnStartedAt;
root.turnActive = false;
// A 409 means there was no turn to recover. It is an ordinary
// resume, not a failed response and must not finish a blank card.
if (reattachRequester.receivedEvent)
root.streamClosed(exitCode);
}
}
function _reattachActiveTurn() {
if (root.conversationId.length === 0 || requester.running || reattachRequester.running)
return;
reattachRequester.receivedEvent = false;
reattachRequester.command = ["bash", "-c",
root._tokenReadLine(root.currentAgentId)
+ `curl --no-buffer -sf "${root.serverBase}/v1/conversations/${root.conversationId}/events"`
+ ` -H "Authorization: Bearer $TOKEN"`
];
root.turnStartedAt = Date.now();
root.turnElapsedMs = 0;
root.turnActive = true;
reattachRequester.running = true;
}
// ── Ambient sensorium ────────────────────────────────────────────────
// What the desktop feels like at the moment of speaking. Cheap,
// synchronous reads here; the cursor needs a hyprctl round-trip and is