@@ -1,125 +1,288 @@
1+//! The Crux `App`: model, events, effects, update logic and view.
3+//! Streaming works like this: after a successful connect the shell opens the
4+//! `/event` SSE stream and forwards every decoded event as
5+//! [`Event::ServerEvent`]. The core parses those and mutates the chat
6+//! incrementally — creating the assistant message on `message.updated`, growing
7+//! its text on `message.part.updated`, and clearing the "generating" flag on
8+//! `session.idle`. User messages are shown optimistically the moment they're
9+//! sent, so the stream only ever *adds* assistant content.
1use crux_core::{
12use crux_core::{
3 macros::effect,
13 macros::effect,
4− render::{RenderOperation, render},
14+ render::{render, RenderOperation},
17+use crux_http::command::{Http, RequestBuilder};
6use crux_http::protocol::HttpRequest;
18use crux_http::protocol::HttpRequest;
8use serde::{Deserialize, Serialize};
19use serde::{Deserialize, Serialize};
21+use crate::protocol::{self, ServerEvent, WireMessageEnvelope, WirePart};
23+/// Result of an HTTP effect handed back to the core.
24+pub type HttpResult = Result<crux_http::Response<String>, crux_http::HttpError>;
10−// ─── Events ────────────────────────────────────────────────────────────────
26+// ─── Events ──────────────────────────────────────────────────────────────────
12−#[derive(Facet, Serialize, Deserialize, Clone, Debug)]
14pub enum Event {
28pub enum Event {
15− // Lifecycle
29+ // Lifecycle / connect
17 ServerUrlChanged(String),
31 ServerUrlChanged(String),
32+ UsernameChanged(String),
33+ PasswordChanged(String),
20− // Session list
36+ // Sessions
21 LoadSessions,
37 LoadSessions,
22− // events local to the core
25− SessionsLoaded(#[facet(opaque)] HttpResult<String>),
26 SelectSession(String),
38 SelectSession(String),
27 CreateSession,
39 CreateSession,
28− SessionCreated(#[facet(opaque)] HttpResult<String>),
40+ DeleteSession(String),
31 LoadMessages(String),
43 LoadMessages(String),
32− MessagesLoaded(#[facet(opaque)] HttpResult<String>),
44+ DraftChanged(String),
33 SendMessage(String),
45 SendMessage(String),
34− MessageSent(#[facet(opaque)] HttpResult<String>),
46+ CancelGeneration,
36− // SSE event received
48+ // A raw event line from the shell's `/event` SSE subscription.
37− EventReceived(String),
49+ ServerEvent(String),
39− // Navigation
51+ // Navigation / misc
40− NavigateToChat(String),
41 NavigateToSessions,
52 NavigateToSessions,
42−
53+ NavigateToConnect,
44 DismissError,
54 DismissError,
56+ // ── Internal: results of HTTP effects (never sent by the shell) ──────────
57+ HealthProbed(HttpResult),
58+ SessionsLoaded(HttpResult),
59+ SessionCreated(HttpResult),
60+ MessagesLoaded(String, HttpResult),
61+ PromptSent(HttpResult),
62+ SessionDeleted(HttpResult),
47−// ─── Effects ───────────────────────────────────────────────────────────────
66+// ─── Effects ─────────────────────────────────────────────────────────────────
49−#[effect(facet_typegen)]
68+#[effect]
50#[derive(Debug)]
69#[derive(Debug)]
51pub enum Effect {
70pub enum Effect {
52 Render(RenderOperation),
71 Render(RenderOperation),
53 Http(HttpRequest),
72 Http(HttpRequest),
56−// ─── HTTP result wrapper (opaque, not type-generated) ──────────────────────
58−pub type HttpResult<T> = Result<crux_http::Response<T>, crux_http::HttpError>;
60−// ─── Model ─────────────────────────────────────────────────────────────────
75+// ─── Model ───────────────────────────────────────────────────────────────────
62#[derive(Default)]
77#[derive(Default)]
63pub struct Model {
78pub struct Model {
64 server_url: String,
79 server_url: String,
82+ /// The server answered 401 to an unauthenticated probe — show credentials.
84+ /// Credentials were accepted; attach the auth header to every request.
65 connected: bool,
86 connected: bool,
66 loading: bool,
87 loading: bool,
67 error: Option<String>,
88 error: Option<String>,
68− sessions: Vec<SessionSummary>,
91+ sessions: Vec<SessionView>,
69 current_session_id: Option<String>,
92 current_session_id: Option<String>,
93+ current_session_title: String,
70− messages: Vec<MessageView>,
95+ messages: Vec<MsgState>,
71− draft_message: String,
96+ draft: String,
99+ /// Monotonic counter for optimistic (client-side) user-message ids.
104+ fn message_mut(&mut self, id: &str) -> Option<&mut MsgState> {
105+ self.messages.iter_mut().find(|m| m.id == id)
108+ fn has_credentials(&self) -> bool {
109+ !self.username.is_empty() || !self.password.is_empty()
112+ fn is_current(&self, session_id: &str) -> bool {
113+ self.current_session_id.as_deref() == Some(session_id)
117+/// Internal per-message state. Parts are tracked by id so streaming updates can
118+/// upsert them; the [`MessageView`] handed to the shell is derived from these.
123+ status: MessageStatus,
125+ text_parts: Vec<(String, String)>,
126+ reasoning_parts: Vec<(String, String)>,
127+ tool_parts: Vec<(String, ToolView)>,
131+ fn new(id: String, role: String, time: u64) -> Self {
136+ status: MessageStatus::Sent,
138+ text_parts: Vec::new(),
139+ reasoning_parts: Vec::new(),
140+ tool_parts: Vec::new(),
144+ fn upsert_part(parts: &mut Vec<(String, String)>, id: &str, text: String) {
145+ if let Some(slot) = parts.iter_mut().find(|(pid, _)| pid == id) {
148+ parts.push((id.to_string(), text));
152+ fn upsert_tool(&mut self, id: &str, view: ToolView) {
153+ if let Some(slot) = self.tool_parts.iter_mut().find(|(pid, _)| pid == id) {
156+ self.tool_parts.push((id.to_string(), view));
160+ fn remove_part(&mut self, part_id: &str) {
161+ self.text_parts.retain(|(pid, _)| pid != part_id);
162+ self.reasoning_parts.retain(|(pid, _)| pid != part_id);
163+ self.tool_parts.retain(|(pid, _)| pid != part_id);
166+ fn to_view(&self) -> MessageView {
167+ let join = |parts: &[(String, String)]| {
168+ parts.iter().map(|(_, t)| t.as_str()).collect::<String>()
170+ let reasoning = join(&self.reasoning_parts);
172+ id: self.id.clone(),
173+ role: self.role.clone(),
174+ text: join(&self.text_parts),
175+ reasoning: (!reasoning.is_empty()).then_some(reasoning),
176+ tools: self.tool_parts.iter().map(|(_, t)| t.clone()).collect(),
178+ status: self.status.clone(),
179+ streaming: self.streaming,
184+// ─── View model ──────────────────────────────────────────────────────────────
186+#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
74−#[derive(Clone, Debug, Default)]
194+#[derive(Serialize, Deserialize, Clone, Debug, Default)]
75−pub struct SessionSummary {
195+pub struct SessionView {
76 pub id: String,
196 pub id: String,
77 pub title: String,
197 pub title: String,
200+#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
201+pub enum MessageStatus {
208+#[derive(Serialize, Deserialize, Clone, Debug, Default)]
209+pub struct ToolView {
212+ pub title: Option<String>,
80−#[derive(Facet, Serialize, Deserialize, Clone, Debug, Default)]
215+#[derive(Serialize, Deserialize, Clone, Debug, Default)]
81pub struct MessageView {
216pub struct MessageView {
82 pub id: String,
217 pub id: String,
83 pub role: String,
218 pub role: String,
84 pub text: String,
219 pub text: String,
220+ pub reasoning: Option<String>,
221+ pub tools: Vec<ToolView>,
85 pub time: u64,
222 pub time: u64,
223+ pub status: MessageStatus,
224+ pub streaming: bool,
88−// ─── ViewModel ─────────────────────────────────────────────────────────────
90−#[derive(Facet, Serialize, Deserialize, Clone, Debug, Default)]
227+#[derive(Serialize, Deserialize, Clone, Debug, Default)]
91pub struct ViewModel {
228pub struct ViewModel {
92 pub screen: Screen,
229 pub screen: Screen,
93 pub server_url: String,
230 pub server_url: String,
231+ pub username: String,
232+ pub password: String,
233+ pub auth_required: bool,
94 pub connected: bool,
234 pub connected: bool,
95 pub loading: bool,
235 pub loading: bool,
96 pub error: Option<String>,
236 pub error: Option<String>,
97 pub sessions: Vec<SessionView>,
237 pub sessions: Vec<SessionView>,
98 pub current_session_id: Option<String>,
238 pub current_session_id: Option<String>,
239+ pub current_session_title: String,
99 pub messages: Vec<MessageView>,
240 pub messages: Vec<MessageView>,
100− pub draft_message: String,
241+ pub draft: String,
242+ pub generating: bool,
103−#[derive(Facet, Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
112−#[derive(Facet, Serialize, Deserialize, Clone, Debug, Default)]
113−pub struct SessionView {
118−// ─── App ───────────────────────────────────────────────────────────────────
245+// ─── App ─────────────────────────────────────────────────────────────────────
120#[derive(Default)]
247#[derive(Default)]
121pub struct OpencodeApp;
248pub struct OpencodeApp;
251+ fn get(&self, model: &Model, path: &str) -> RequestBuilder<Effect, Event, String> {
252+ let url = format!("{}{path}", model.server_url);
253+ self.with_auth(Http::get(&url).expect_string(), model)
256+ fn post_json(&self, model: &Model, path: &str, body: &serde_json::Value) -> RequestBuilder<Effect, Event, String> {
257+ let url = format!("{}{path}", model.server_url);
258+ let bytes = serde_json::to_vec(body).unwrap_or_default();
259+ let req = Http::post(&url)
261+ .header("Content-Type", "application/json")
263+ self.with_auth(req, model)
266+ fn delete(&self, model: &Model, path: &str) -> RequestBuilder<Effect, Event, String> {
267+ let url = format!("{}{path}", model.server_url);
268+ self.with_auth(Http::delete(&url).expect_string(), model)
271+ /// Attach an RFC 7617 Basic-Auth header when the model carries credentials.
272+ fn with_auth<E>(&self, req: RequestBuilder<Effect, Event, E>, model: &Model) -> RequestBuilder<Effect, Event, E>
276+ if model.has_credentials() {
277+ let creds = format!("{}:{}", model.username, model.password);
278+ let encoded = base64::engine::general_purpose::STANDARD.encode(creds.as_bytes());
279+ req.header("Authorization", format!("Basic {encoded}"))
123impl App for OpencodeApp {
286impl App for OpencodeApp {
124 type Event = Event;
287 type Event = Event;
125 type Model = Model;
288 type Model = Model;
@@ -128,150 +291,246 @@ impl App for OpencodeApp {
129 fn update(&self, event: Event, model: &mut Model) -> Command<Effect, Event> {
292 fn update(&self, event: Event, model: &mut Model) -> Command<Effect, Event> {
130 match event {
293 match event {
294+ // ── Lifecycle / connect ─────────────────────────────────────────
131− Event::Start => {
295+ Event::Start => render(),
132− model.server_url = "http://localhost:4096".to_string();
297+ // Store the field verbatim while typing; normalizing here would fight
298+ // the user's edits (e.g. eat the "://" as they type it). The URL is
299+ // normalized once, on Connect.
300+ // Clear any prior error as soon as the user edits a field, so a stale
301+ // message (e.g. "Invalid credentials") doesn't linger while they correct it.
136 Event::ServerUrlChanged(url) => {
302 Event::ServerUrlChanged(url) => {
137 model.server_url = url;
303 model.server_url = url;
304+ if model.error.is_some() {
309+ Event::UsernameChanged(u) => {
311+ if model.error.is_some() {
316+ Event::PasswordChanged(p) => {
318+ if model.error.is_some() {
141 Event::Connect => {
324 Event::Connect => {
325+ model.server_url = normalize_url(&model.server_url);
326+ if model.server_url.is_empty() {
327+ model.error = Some("Enter a server URL".to_string());
142 model.loading = true;
330 model.loading = true;
143 model.error = None;
331 model.error = None;
144− let url = format!("{}/global/health", model.server_url);
332+ model.auth_required = false;
145− Http::get(&url).expect_string().build().then_send(|_| Event::LoadSessions)
333+ render().and(self.get(model, "/global/health").build().then_send(Event::HealthProbed))
336+ Event::HealthProbed(result) => self.on_health_probed(model, result),
338+ // ── Sessions ────────────────────────────────────────────────────
148 Event::LoadSessions => {
339 Event::LoadSessions => {
149 model.loading = true;
340 model.loading = true;
150− let url = format!("{}/session", model.server_url);
151− Http::get(&url).expect_string().build().then_send(Event::SessionsLoaded)
341+ render().and(self.get(model, "/session").build().then_send(Event::SessionsLoaded))
154 Event::SessionsLoaded(result) => {
344 Event::SessionsLoaded(result) => {
155 model.loading = false;
345 model.loading = false;
156− match result {
346+ match ok_body(result) {
157− Ok(mut response) => {
347+ Ok(body) => {
158− let body = response.take_body().unwrap_or_default();
159− model.sessions = parse_sessions(&body);
348+ model.sessions = protocol::parse_sessions(&body)
160− model.connected = true;
349+ .into_iter()
161− render()
350+ .map(|s| SessionView {
352+ title: display_title(&s.title),
163− Err(e) => {
354+ .collect();
164− model.error = Some(format!("Failed to load sessions: {e}"));
355+ // Impose a stable alphabetical order so the list doesn't shuffle
165− render()
356+ // with whatever order the server happens to return sessions in.
357+ model.sessions.sort_by_key(|s| s.title.to_lowercase());
359+ Err(e) => model.error = Some(e),
170 Event::SelectSession(id) => {
364 Event::SelectSession(id) => {
171− model.current_session_id = Some(id.clone());
365+ model.current_session_title = model
172− Command::event(Event::NavigateToChat(id))
366+ .sessions
174−
368+ .find(|s| s.id == id)
175− Event::NavigateToChat(id) => {
369+ .map(|s| s.title.clone())
370+ .unwrap_or_default();
176 model.current_session_id = Some(id.clone());
371 model.current_session_id = Some(id.clone());
177 model.messages.clear();
372 model.messages.clear();
373+ model.screen = Screen::Chat;
374+ model.loading = true;
178 render().and(Command::event(Event::LoadMessages(id)))
376 render().and(Command::event(Event::LoadMessages(id)))
181− Event::NavigateToSessions => {
182− model.current_session_id = None;
183− model.messages.clear();
187 Event::CreateSession => {
379 Event::CreateSession => {
188 model.loading = true;
380 model.loading = true;
189− let url = format!("{}/session", model.server_url);
381+ render().and(
382+ self.post_json(model, "/session", &serde_json::json!({}))
190− Http::post(&url).expect_string().build().then_send(Event::SessionCreated)
384+ .then_send(Event::SessionCreated),
193 Event::SessionCreated(result) => {
388 Event::SessionCreated(result) => {
194 model.loading = false;
389 model.loading = false;
195− match result {
390+ match ok_body(result) {
196− Ok(mut response) => {
391+ Ok(body) => {
197− let body = response.take_body().unwrap_or_default();
392+ if let Some(s) = protocol::parse_session(&body) {
198− let id = parse_session_id(&body);
393+ let view = SessionView {
395+ title: display_title(&s.title),
397+ if !model.sessions.iter().any(|e| e.id == view.id) {
398+ model.sessions.insert(0, view);
199− Command::event(Event::NavigateToChat(id))
400+ return render().and(Command::event(Event::SelectSession(s.id)));
202− model.error = Some(format!("Failed to create session: {e}"));
402+ model.error = Some("Malformed session response".to_string());
404+ Err(e) => model.error = Some(e),
409+ Event::DeleteSession(id) => {
410+ model.sessions.retain(|s| s.id != id);
411+ let mut cmd = self.delete(model, &format!("/session/{id}")).build().then_send(Event::SessionDeleted);
412+ if model.is_current(&id) {
413+ model.current_session_id = None;
414+ model.messages.clear();
415+ model.screen = Screen::Sessions;
417+ cmd = render().and(cmd);
421+ Event::SessionDeleted(result) => {
422+ if let Err(e) = ok_body(result) {
423+ model.error = Some(e);
428+ // ── Chat ────────────────────────────────────────────────────────
208− Event::LoadMessages(session_id) => {
429+ Event::LoadMessages(id) => {
209− model.loading = true;
430+ render().and(
210− let url = format!("{}/session/{}/message", model.server_url, session_id);
431+ self.get(model, &format!("/session/{id}/message"))
211− Http::get(&url).expect_string().build().then_send(Event::MessagesLoaded)
433+ .then_send(move |result| Event::MessagesLoaded(id.clone(), result)),
214− Event::MessagesLoaded(result) => {
437+ Event::MessagesLoaded(session_id, result) => {
438+ if !model.is_current(&session_id) {
215 model.loading = false;
441 model.loading = false;
216− match result {
442+ match ok_body(result) {
217− Ok(mut response) => {
443+ Ok(body) => {
218− let body = response.take_body().unwrap_or_default();
219− model.messages = parse_messages(&body);
444+ model.messages = protocol::parse_messages(&body).iter().map(message_from_wire).collect();
220− render()
445+ model.generating = model.messages.iter().any(|m| m.streaming);
223− model.error = Some(format!("Failed to load messages: {e}"));
447+ Err(e) => model.error = Some(e),
452+ Event::DraftChanged(text) => {
229 Event::SendMessage(text) => {
457 Event::SendMessage(text) => {
458+ let text = text.trim().to_string();
230− let session_id = match &model.current_session_id {
459+ let Some(session_id) = model.current_session_id.clone() else {
231− Some(id) => id.clone(),
232− None => return render(),
460+ return render();
234− model.draft_message.clear();
462+ if text.is_empty() {
235− let url = format!(
463+ return render();
236− "{}/session/{}/prompt_async",
464+ }
237− model.server_url, session_id
465+ model.local_seq += 1;
238− );
466+ let mut user = MsgState::new(format!("local-{}", model.local_seq), "user".to_string(), 0);
239− let body = format!(
467+ user.status = MessageStatus::Pending;
468+ MsgState::upsert_part(&mut user.text_parts, "local", text.clone());
469+ model.messages.push(user);
470+ model.draft.clear();
471+ model.generating = true;
240− r#"{{"sessionID":"{}","parts":[{{"type":"text","text":"{}"}}]}}"#,
474+ let body = serde_json::json!({ "parts": [{ "type": "text", "text": text }] });
241− session_id, text
475+ render().and(
242− );
476+ self.post_json(model, &format!("/session/{session_id}/prompt_async"), &body)
244− .header("Content-Type", "application/json")
245− .body_bytes(body.into_bytes())
247− .build()
477+ .build()
248− .then_send(Event::MessageSent)
478+ .then_send(Event::PromptSent),
251− Event::MessageSent(result) => {
482+ Event::PromptSent(result) => {
252− match result {
483+ match ok_body(result) {
253− Ok(_) => {
484+ Ok(_) => set_last_pending(model, MessageStatus::Sent),
254− // Reload messages to get the response
255− match &model.current_session_id {
256− Some(id) => Command::event(Event::LoadMessages(id.clone())),
260 Err(e) => {
485 Err(e) => {
486+ set_last_pending(model, MessageStatus::Failed);
487+ model.generating = false;
261− model.error = Some(format!("Failed to send message: {e}"));
488+ model.error = Some(e);
267− Event::EventReceived(_data) => {
494+ Event::CancelGeneration => {
268− // SSE event — reload messages to show updates
495+ model.generating = false;
269− match &model.current_session_id {
496+ match model.current_session_id.clone() {
270− Some(id) => Command::event(Event::LoadMessages(id.clone())),
497+ Some(id) => render().and(
498+ self.post_json(model, &format!("/session/{id}/abort"), &serde_json::json!({}))
500+ .then_send(Event::Aborted),
271 None => render(),
502 None => render(),
506+ Event::Aborted(_) => render(),
508+ // ── SSE ─────────────────────────────────────────────────────────
509+ Event::ServerEvent(json) => {
510+ if let Some(event) = protocol::parse_event(&json) {
511+ apply_server_event(model, event);
516+ // ── Navigation / misc ───────────────────────────────────────────
517+ Event::NavigateToSessions => {
518+ model.screen = Screen::Sessions;
519+ model.current_session_id = None;
520+ model.messages.clear();
521+ model.generating = false;
522+ render().and(self.get(model, "/session").build().then_send(Event::SessionsLoaded))
525+ Event::NavigateToConnect => {
526+ model.screen = Screen::Connect;
527+ model.connected = false;
528+ model.sessions.clear();
529+ model.messages.clear();
530+ model.current_session_id = None;
275 Event::DismissError => {
534 Event::DismissError => {
276 model.error = None;
535 model.error = None;
@@ -281,214 +540,512 @@ impl App for OpencodeApp {
282 fn view(&self, model: &Model) -> ViewModel {
541 fn view(&self, model: &Model) -> ViewModel {
283 ViewModel {
542 ViewModel {
284− screen: if model.current_session_id.is_some() {
543+ screen: model.screen.clone(),
286− } else if model.connected {
291 server_url: model.server_url.clone(),
544 server_url: model.server_url.clone(),
545+ username: model.username.clone(),
546+ password: model.password.clone(),
547+ auth_required: model.auth_required,
292 connected: model.connected,
548 connected: model.connected,
293 loading: model.loading,
549 loading: model.loading,
294 error: model.error.clone(),
550 error: model.error.clone(),
295− sessions: model
551+ sessions: model.sessions.clone(),
298− .map(|s| SessionView {
300− title: s.title.clone(),
303 current_session_id: model.current_session_id.clone(),
552 current_session_id: model.current_session_id.clone(),
553+ current_session_title: model.current_session_title.clone(),
304− messages: model.messages.clone(),
554+ messages: model.messages.iter().map(MsgState::to_view).collect(),
305− draft_message: model.draft_message.clone(),
555+ draft: model.draft.clone(),
556+ generating: model.generating,
310−// ─── JSON parsing helpers (naive — the core stays pure) ────────────────────
561+impl OpencodeApp {
311−
562+ fn on_health_probed(&self, model: &mut Model, result: HttpResult) -> Command<Effect, Event> {
312−fn parse_sessions(json: &str) -> Vec<SessionSummary> {
563+ let has_creds = model.has_credentials();
313− let mut sessions = Vec::new();
316− let bytes = json.as_bytes();
319− while i < bytes.len() {
320− match bytes[i] {
564+ match result {
321− b'{' if depth == 0 => {
565+ Ok(response) => {
322− depth = 1;
566+ let status = response.status().as_u16();
323− start = i;
567+ if status == 401 {
324− }
568+ model.loading = false;
325− b'{' => depth += 1,
569+ if has_creds {
326− b'}' => {
570+ model.error = Some("Invalid credentials".to_string());
327− depth -= 1;
571+ } else {
328− if depth == 0 {
572+ model.auth_required = true;
329− let obj = &json[start..=i];
573+ }
330− let id = extract_string(obj, "id").unwrap_or_default();
574+ return render();
331− let title = extract_string(obj, "title").unwrap_or_default();
575+ }
332− if !id.is_empty() {
576+ if (200..300).contains(&status) {
333− sessions.push(SessionSummary { id, title });
577+ if has_creds {
578+ model.authed = true;
580+ model.connected = true;
581+ model.auth_required = false;
583+ model.screen = Screen::Sessions;
584+ model.loading = true;
585+ return render().and(self.get(model, "/session").build().then_send(Event::SessionsLoaded));
587+ model.loading = false;
588+ model.error = Some(format!("Server returned status {status}"));
592+ model.loading = false;
593+ model.error = Some(format!("Connection failed: {e}"));
344−fn parse_session_id(json: &str) -> String {
600+// ─── SSE application ─────────────────────────────────────────────────────────
345− extract_string(json, "id").unwrap_or_default()
601+
602+fn apply_server_event(model: &mut Model, event: ServerEvent) {
604+ // Only assistant messages are created from the stream — the user's own
605+ // message is already shown optimistically, so ignoring the server's copy
606+ // avoids a duplicate bubble.
607+ ServerEvent::MessageUpdated { properties } => {
608+ let info = properties.info;
609+ if info.role != "assistant" || !model.is_current(&info.session_id) {
612+ let streaming = info.time.completed.is_none() && info.error.is_none();
613+ if let Some(msg) = model.message_mut(&info.id) {
614+ msg.streaming = streaming;
616+ msg.time = info.time.created;
619+ let mut msg = MsgState::new(info.id.clone(), "assistant".to_string(), info.time.created);
620+ msg.streaming = streaming;
621+ model.messages.push(msg);
623+ if let Some(err) = info.error {
624+ model.error = Some(err.message());
625+ model.generating = false;
348−fn parse_messages(json: &str) -> Vec<MessageView> {
629+ // Apply only to a message we already track (an assistant message from the
349− let mut messages = Vec::new();
630+ // event above). Parts for unknown ids are ignored — see module docs.
350− let mut depth = 0;
631+ ServerEvent::MessagePartUpdated { properties } => {
351− let mut start = 0;
632+ let part = properties.part;
352− let bytes = json.as_bytes();
633+ let delta = properties.delta;
353−
634+ let msg_id = part.message_id().to_string();
354− let mut i = 0;
635+ let Some(msg) = model.message_mut(&msg_id) else {
355− while i < bytes.len() {
636+ return;
356− match bytes[i] {
638+ match part {
357− b'{' if depth == 0 => {
639+ WirePart::Text { id, text, synthetic, .. } => {
358− depth = 1;
640+ if synthetic {
359− start = i;
641+ return;
365− let obj = &json[start..=i];
366− let id = extract_string(obj, "id").unwrap_or_default();
367− let role = extract_string(obj, "role").unwrap_or_default();
368− let time = extract_number(obj, "created").unwrap_or(0);
369− if !id.is_empty() && !role.is_empty() {
370− messages.push(MessageView {
373− text: String::new(),
643+ let value = pick_text(text, &delta, &msg.text_parts, &id);
644+ MsgState::upsert_part(&mut msg.text_parts, &id, value);
646+ WirePart::Reasoning { id, text, .. } => {
647+ let value = pick_text(text, &delta, &msg.reasoning_parts, &id);
648+ MsgState::upsert_part(&mut msg.reasoning_parts, &id, value);
650+ WirePart::Tool { id, tool, state, .. } => {
651+ let sv = protocol::tool_state_view(&state);
652+ msg.upsert_tool(&id, ToolView { name: tool, status: sv.status, title: sv.title });
654+ WirePart::Other => {}
658+ ServerEvent::MessagePartRemoved { properties } => {
659+ if let Some(msg) = model.message_mut(&properties.message_id) {
660+ msg.remove_part(&properties.part_id);
664+ ServerEvent::MessageRemoved { properties } => {
665+ model.messages.retain(|m| m.id != properties.message_id);
668+ ServerEvent::SessionIdle { properties } => {
669+ if model.is_current(&properties.session_id) {
670+ model.generating = false;
671+ for m in &mut model.messages {
672+ m.streaming = false;
677+ ServerEvent::SessionError { properties } => {
678+ let scoped = properties.session_id.as_deref().map_or(true, |s| model.is_current(s));
680+ if let Some(err) = properties.error {
681+ model.error = Some(err.message());
683+ model.generating = false;
687+ ServerEvent::SessionCreated { properties } | ServerEvent::SessionUpdated { properties } => {
688+ let s = properties.info;
689+ let view = SessionView { id: s.id.clone(), title: display_title(&s.title) };
690+ if let Some(existing) = model.sessions.iter_mut().find(|e| e.id == view.id) {
691+ existing.title = view.title.clone();
693+ model.sessions.insert(0, view);
695+ if model.is_current(&s.id) {
696+ model.current_session_title = display_title(&s.title);
700+ ServerEvent::SessionDeleted { properties } => {
701+ let id = properties.session_id.or(properties.info.map(|i| i.id));
702+ if let Some(id) = id {
703+ model.sessions.retain(|s| s.id != id);
704+ if model.is_current(&id) {
705+ model.current_session_id = None;
706+ model.messages.clear();
707+ model.screen = Screen::Sessions;
712+ ServerEvent::Other => {}
716+// ─── Helpers ─────────────────────────────────────────────────────────────────
718+/// `part.text` is the full current text when present; otherwise fall back to
719+/// accumulating the incremental `delta` onto whatever we already have.
720+fn pick_text(text: String, delta: &Option<String>, existing: &[(String, String)], part_id: &str) -> String {
721+ if !text.is_empty() {
724+ let prior = existing.iter().find(|(pid, _)| pid == part_id).map(|(_, t)| t.clone()).unwrap_or_default();
726+ Some(d) => format!("{prior}{d}"),
731+fn message_from_wire(env: &WireMessageEnvelope) -> MsgState {
732+ let info = &env.info;
733+ let mut msg = MsgState::new(info.id.clone(), info.role.clone(), info.time.created);
734+ msg.streaming = info.role == "assistant" && info.time.completed.is_none() && info.error.is_none();
735+ for part in &env.parts {
737+ WirePart::Text { id, text, synthetic, .. } => {
739+ MsgState::upsert_part(&mut msg.text_parts, id, text.clone());
379− _ => {}
742+ WirePart::Reasoning { id, text, .. } => {
743+ MsgState::upsert_part(&mut msg.reasoning_parts, id, text.clone());
745+ WirePart::Tool { id, tool, state, .. } => {
746+ let sv = protocol::tool_state_view(state);
747+ msg.upsert_tool(id, ToolView { name: tool.clone(), status: sv.status, title: sv.title });
749+ WirePart::Other => {}
755+fn set_last_pending(model: &mut Model, status: MessageStatus) {
756+ if let Some(msg) = model.messages.iter_mut().rev().find(|m| m.status == MessageStatus::Pending) {
757+ msg.status = status;
386−fn extract_string(json: &str, key: &str) -> Option<String> {
761+/// Trim whitespace and any trailing `/` so paths concatenate cleanly, and
387− let pattern = format!("\"{key}\"");
762+/// default to an `http://` scheme when the user didn't type one (e.g.
388− let idx = json.find(&pattern)?;
763+/// "192.168.1.10:4096") — otherwise the shell's URL parser rejects it.
389− let rest = &json[idx + pattern.len()..];
764+fn normalize_url(url: &str) -> String {
390− let colon = rest.find(':')?;
765+ let trimmed = url.trim().trim_end_matches('/');
391− let rest = &rest[colon + 1..];
766+ if trimmed.is_empty() {
392− let quote = rest.find('"')?;
767+ String::new()
393− let rest = &rest[quote + 1..];
768+ } else if trimmed.contains("://") {
394− let end = rest.find('"')?;
769+ trimmed.to_string()
395− Some(rest[..end].to_string())
770+ } else {
771+ format!("http://{trimmed}")
398−fn extract_number(json: &str, key: &str) -> Option<u64> {
775+fn display_title(title: &str) -> String {
399− let pattern = format!("\"{key}\"");
776+ let t = title.trim();
400− let idx = json.find(&pattern)?;
777+ if t.is_empty() {
401− let rest = &json[idx + pattern.len()..];
778+ "Untitled".to_string()
402− let colon = rest.find(':')?;
779+ } else {
403− let rest = rest[colon + 1..].trim_start();
780+ t.to_string()
404− let end = rest
781+ }
405− .find(|c: char| !c.is_ascii_digit())
406− .unwrap_or(rest.len());
407− rest[..end].parse().ok()
410−// ─── HTTP type alias for the command builder ───────────────────────────────
784+/// Reduce an [`HttpResult`] to `Ok(body)` for 2xx, or `Err(message)` otherwise.
785+fn ok_body(result: HttpResult) -> Result<String, String> {
787+ Ok(mut response) => {
788+ let status = response.status().as_u16();
789+ if (200..300).contains(&status) {
790+ Ok(response.take_body().unwrap_or_default())
792+ Err(format!("Server returned status {status}"))
795+ Err(e) => Err(format!("Request failed: {e}")),
412−use crux_http::command::Http;
799+// ─── Tests ───────────────────────────────────────────────────────────────────
414#[cfg(test)]
801#[cfg(test)]
415−mod test {
802+mod tests {
416 use super::*;
803 use super::*;
805+ fn app() -> OpencodeApp {
806+ OpencodeApp::default()
809+ /// Put the model into a connected Chat view for the given session.
810+ fn in_chat(session_id: &str) -> Model {
811+ let mut model = Model::default();
812+ model.server_url = "http://localhost:4096".to_string();
813+ model.connected = true;
814+ model.screen = Screen::Chat;
815+ model.current_session_id = Some(session_id.to_string());
819+ fn feed(app: &OpencodeApp, model: &mut Model, json: serde_json::Value) {
820+ let _ = app.update(Event::ServerEvent(json.to_string()), model);
419− fn start_sets_default_server_url() {
824+ fn server_url_is_kept_raw_while_typing() {
420− let app = OpencodeApp;
825+ let app = app();
421 let mut model = Model::default();
826 let mut model = Model::default();
827+ // Mid-scheme input must survive verbatim — normalizing per keystroke
828+ // would eat the "://" the user is typing and make the field unusable.
422− app.update(Event::Start, &mut model).expect_only_render();
829+ let _ = app.update(Event::ServerUrlChanged("http:/".into()), &mut model);
423− assert_eq!(model.server_url, "http://localhost:4096");
830+ assert_eq!(app.view(&model).server_url, "http:/");
427− fn server_url_change_updates_model() {
834+ fn connect_normalizes_scheme_less_url() {
428− let app = OpencodeApp;
835+ let app = app();
429 let mut model = Model::default();
836 let mut model = Model::default();
430− app.update(Event::Start, &mut model);
837+ // A scheme-less host (what a user often types) must gain http:// on
431− app.update(
838+ // connect, or the shell's URL parser throws and the attempt crashes.
432− Event::ServerUrlChanged("http://10.0.0.1:4096".to_string()),
839+ let _ = app.update(Event::ServerUrlChanged("192.168.1.10:4096".into()), &mut model);
433− &mut model,
840+ assert_eq!(app.view(&model).server_url, "192.168.1.10:4096");
434− )
841+ let _ = app.update(Event::Connect, &mut model);
435− .expect_only_render();
436− assert_eq!(model.server_url, "http://10.0.0.1:4096");
842+ assert_eq!(app.view(&model).server_url, "http://192.168.1.10:4096");
440− fn connect_emits_http_effect() {
846+ fn connect_trims_and_keeps_explicit_scheme() {
441− let app = OpencodeApp;
847+ let app = app();
442 let mut model = Model::default();
848 let mut model = Model::default();
443− app.update(Event::Start, &mut model);
849+ let _ = app.update(Event::ServerUrlChanged(" https://host:4096/ ".into()), &mut model);
444− let mut cmd = app.update(Event::Connect, &mut model);
850+ let _ = app.update(Event::Connect, &mut model);
445− assert!(cmd.effects().any(|e| matches!(e, Effect::Http(_))));
851+ assert_eq!(app.view(&model).server_url, "https://host:4096");
446− assert!(model.loading);
450− fn parse_sessions_extracts_id_and_title() {
855+ fn connect_requires_a_url() {
451− let json = r#"[{"id":"abc123","title":"My Session"},{"id":"def456","title":"Test"}]"#;
856+ let app = app();
452− let sessions = parse_sessions(json);
857+ let mut model = Model::default();
453− assert_eq!(sessions.len(), 2);
858+ let _ = app.update(Event::Connect, &mut model);
454− assert_eq!(sessions[0].id, "abc123");
859+ assert_eq!(app.view(&model).error.as_deref(), Some("Enter a server URL"));
455− assert_eq!(sessions[0].title, "My Session");
456− assert_eq!(sessions[1].id, "def456");
460− fn parse_session_id_extracts_id() {
863+ fn sending_adds_optimistic_user_message_and_sets_generating() {
461− let json = r#"{"id":"xyz789","title":"New"}"#;
864+ let app = app();
462− let id = parse_session_id(json);
865+ let mut model = in_chat("ses_1");
463− assert_eq!(id, "xyz789");
866+ let _ = app.update(Event::DraftChanged("hello there".into()), &mut model);
867+ let _ = app.update(Event::SendMessage("hello there".into()), &mut model);
869+ let view = app.view(&model);
870+ assert!(view.generating);
871+ assert!(view.draft.is_empty());
872+ assert_eq!(view.messages.len(), 1);
873+ assert_eq!(view.messages[0].role, "user");
874+ assert_eq!(view.messages[0].text, "hello there");
875+ assert_eq!(view.messages[0].status, MessageStatus::Pending);
467− fn extract_string_finds_nested_keys() {
879+ fn empty_send_is_ignored() {
468− let json = r#"{"a":"b","id":"target","c":"d"}"#;
880+ let app = app();
469− assert_eq!(extract_string(json, "id"), Some("target".to_string()));
881+ let mut model = in_chat("ses_1");
470− assert_eq!(extract_string(json, "missing"), None);
882+ let _ = app.update(Event::SendMessage(" ".into()), &mut model);
883+ assert!(app.view(&model).messages.is_empty());
884+ assert!(!app.view(&model).generating);
474− fn extract_number_parses_values() {
888+ fn assistant_reply_streams_in_from_sse() {
475− let json = r#"{"created":1234567890}"#;
889+ let app = app();
476− assert_eq!(extract_number(json, "created"), Some(1234567890));
890+ let mut model = in_chat("ses_1");
892+ // Assistant message shell arrives first.
893+ feed(&app, &mut model, serde_json::json!({
894+ "type": "message.updated",
895+ "properties": { "info": {
896+ "id": "msg_a", "role": "assistant", "sessionID": "ses_1",
897+ "time": { "created": 100 }
900+ assert_eq!(app.view(&model).messages.len(), 1);
901+ assert!(app.view(&model).messages[0].streaming);
903+ // Two text deltas for the same part accumulate into full text.
904+ feed(&app, &mut model, serde_json::json!({
905+ "type": "message.part.updated",
906+ "properties": { "part": {
907+ "id": "prt_1", "messageID": "msg_a", "sessionID": "ses_1",
908+ "type": "text", "text": "Hello"
911+ feed(&app, &mut model, serde_json::json!({
912+ "type": "message.part.updated",
913+ "properties": { "part": {
914+ "id": "prt_1", "messageID": "msg_a", "sessionID": "ses_1",
915+ "type": "text", "text": "Hello, world"
919+ let view = app.view(&model);
920+ assert_eq!(view.messages[0].role, "assistant");
921+ assert_eq!(view.messages[0].text, "Hello, world");
480− fn navigate_to_sessions_clears_state() {
925+ fn text_delta_accumulates_when_full_text_absent() {
481− let app = OpencodeApp;
926+ let app = app();
482− let mut model = Model::default();
927+ let mut model = in_chat("ses_1");
483− model.current_session_id = Some("test".to_string());
928+ feed(&app, &mut model, serde_json::json!({
484− model.messages.push(MessageView {
929+ "type": "message.updated",
485− id: "m1".to_string(),
930+ "properties": { "info": { "id": "msg_a", "role": "assistant", "sessionID": "ses_1", "time": {"created": 1}}}
486− role: "user".to_string(),
931+ }));
487− text: "hello".to_string(),
932+ for delta in ["Wor", "ld"] {
488− time: 0,
933+ feed(&app, &mut model, serde_json::json!({
934+ "type": "message.part.updated",
936+ "part": { "id": "p", "messageID": "msg_a", "sessionID": "ses_1", "type": "text", "text": "" },
941+ assert_eq!(app.view(&model).messages[0].text, "World");
945+ fn user_message_from_sse_is_not_duplicated() {
947+ let mut model = in_chat("ses_1");
948+ let _ = app.update(Event::SendMessage("hi".into()), &mut model);
949+ // Server echoes the user message on the stream — must be ignored.
950+ feed(&app, &mut model, serde_json::json!({
951+ "type": "message.updated",
952+ "properties": { "info": { "id": "msg_user", "role": "user", "sessionID": "ses_1", "time": {"created": 5}}}
954+ let users = app.view(&model).messages.iter().filter(|m| m.role == "user").count();
955+ assert_eq!(users, 1);
959+ fn session_idle_clears_generating() {
961+ let mut model = in_chat("ses_1");
962+ let _ = app.update(Event::SendMessage("hi".into()), &mut model);
963+ assert!(app.view(&model).generating);
964+ feed(&app, &mut model, serde_json::json!({
965+ "type": "session.idle", "properties": { "sessionID": "ses_1" }
967+ assert!(!app.view(&model).generating);
971+ fn idle_for_other_session_is_ignored() {
973+ let mut model = in_chat("ses_1");
974+ let _ = app.update(Event::SendMessage("hi".into()), &mut model);
975+ feed(&app, &mut model, serde_json::json!({
976+ "type": "session.idle", "properties": { "sessionID": "ses_other" }
978+ assert!(app.view(&model).generating);
982+ fn reasoning_and_tool_parts_are_surfaced() {
984+ let mut model = in_chat("ses_1");
985+ feed(&app, &mut model, serde_json::json!({
986+ "type": "message.updated",
987+ "properties": { "info": { "id": "m", "role": "assistant", "sessionID": "ses_1", "time": {"created": 1}}}
989+ feed(&app, &mut model, serde_json::json!({
990+ "type": "message.part.updated",
991+ "properties": { "part": { "id": "r", "messageID": "m", "type": "reasoning", "text": "thinking..." }}
993+ feed(&app, &mut model, serde_json::json!({
994+ "type": "message.part.updated",
995+ "properties": { "part": {
996+ "id": "t", "messageID": "m", "type": "tool", "tool": "bash",
997+ "state": { "status": "completed", "title": "ls -la" }
1000+ let msg = &app.view(&model).messages[0];
1001+ assert_eq!(msg.reasoning.as_deref(), Some("thinking..."));
1002+ assert_eq!(msg.tools.len(), 1);
1003+ assert_eq!(msg.tools[0].name, "bash");
1004+ assert_eq!(msg.tools[0].status, "completed");
1005+ assert_eq!(msg.tools[0].title.as_deref(), Some("ls -la"));
1009+ fn session_created_event_updates_list() {
1011+ let mut model = in_chat("ses_1");
1012+ feed(&app, &mut model, serde_json::json!({
1013+ "type": "session.created",
1014+ "properties": { "info": { "id": "ses_new", "title": "Fresh" }}
1016+ assert!(app.view(&model).sessions.iter().any(|s| s.id == "ses_new" && s.title == "Fresh"));
1020+ fn unknown_event_types_are_ignored() {
1022+ let mut model = in_chat("ses_1");
1023+ feed(&app, &mut model, serde_json::json!({
1024+ "type": "pty.created", "properties": { "whatever": true }
490− app.update(Event::NavigateToSessions, &mut model);
1026+ feed(&app, &mut model, serde_json::json!({ "type": "totally.new.event" }));
491− assert!(model.current_session_id.is_none());
1027+ // No panic, no state change.
492− assert!(model.messages.is_empty());
1028+ assert!(app.view(&model).messages.is_empty());
1032+ fn messages_loaded_parses_info_and_parts() {
1034+ let mut model = in_chat("ses_1");
1035+ let body = serde_json::json!([
1036+ { "info": { "id": "m1", "role": "user", "sessionID": "ses_1", "time": {"created": 1} },
1037+ "parts": [{ "id": "p1", "type": "text", "text": "Question?" }] },
1038+ { "info": { "id": "m2", "role": "assistant", "sessionID": "ses_1", "time": {"created": 2, "completed": 3} },
1039+ "parts": [{ "id": "p2", "type": "text", "text": "Answer." }] }
1041+ let result: HttpResult = Ok(crux_http::testing::ResponseBuilder::ok().body(body).build());
1042+ let _ = app.update(Event::MessagesLoaded("ses_1".into(), result), &mut model);
1044+ let view = app.view(&model);
1045+ assert_eq!(view.messages.len(), 2);
1046+ assert_eq!(view.messages[0].text, "Question?");
1047+ assert_eq!(view.messages[1].text, "Answer.");
1048+ assert!(!view.messages[1].streaming);
1049+ assert!(!view.generating);