feat(orchestration): surface tool_result outcome + per-session message_count

app/src-tauri/src/telegram_scanner/mod.rs

@@ -694,7 +694,10 @@ mod tests {
         assert_eq!(done.load(Ordering::SeqCst), 20, "every item must run");
         let peak = max_seen.load(Ordering::SeqCst);
         assert!(peak >= 2, "work should actually overlap (peak {peak})");
-        assert!(peak <= 3, "concurrency must stay bounded to 3 (peak {peak})");
+        assert!(
+            peak <= 3,
+            "concurrency must stay bounded to 3 (peak {peak})"
+        );
         assert_eq!(
             inflight.load(Ordering::SeqCst),
             0,

src/openhuman/orchestration/ingest.rs

@@ -104,6 +104,10 @@ struct ClassifiedMessage {
     tool_name: Option<String>,
     /// v2 `call_id` correlating `tool_result` → `tool_call`.
     call_id: Option<String>,
+    /// v2 `tool_result` outcome (`None` for rows that are not `tool_result`).
+    ok: Option<bool>,
+    is_error: Option<bool>,
+    exit_code: Option<i64>,
     /// v2 `status.state` (or a derived state for approval/lifecycle/error) written
     /// onto the session row so `derive_status` reads a real run-state.
     status_state: Option<String>,
@@ -187,6 +191,9 @@ fn classify_message(plaintext: String, fallback_timestamp: &str) -> ClassifiedMe
         event_kind: None,
         tool_name: None,
         call_id: None,
+        ok: None,
+        is_error: None,
+        exit_code: None,
         status_state: None,
         status_detail: None,
         active_call_id: None,
@@ -231,6 +238,9 @@ fn classify_v1(env: SessionEnvelopeV1, fallback_timestamp: &str) -> ClassifiedMe
         event_kind: None,
         tool_name: None,
         call_id: None,
+        ok: None,
+        is_error: None,
+        exit_code: None,
         status_state: None,
         status_detail: None,
         active_call_id: None,
@@ -303,6 +313,10 @@ fn classify_v2(env: SessionEnvelopeV2, fallback_timestamp: &str) -> ClassifiedMe
         HarnessEventKind::ToolResult(p) => {
             b.body = p.output;
             b.call_id = non_empty(p.call_id);
+            // Carry the outcome so the renderer can distinguish a failed run.
+            b.ok = p.ok;
+            b.is_error = Some(p.is_error);
+            b.exit_code = p.exit_code;
         }
         HarnessEventKind::ApprovalRequest(p) => {
             b.body = p.display;
@@ -363,6 +377,9 @@ fn classify_v2(env: SessionEnvelopeV2, fallback_timestamp: &str) -> ClassifiedMe
         event_kind: Some(b.kind_override.map(str::to_string).unwrap_or(kind_str)),
         tool_name: b.tool_name,
         call_id: b.call_id,
+        ok: b.ok,
+        is_error: b.is_error,
+        exit_code: b.exit_code,
         status_state: b.status_state,
         status_detail: b.status_detail,
         active_call_id: b.active_call_id,
@@ -384,6 +401,9 @@ struct V2Body {
     default_role: &'static str,
     tool_name: Option<String>,
     call_id: Option<String>,
+    ok: Option<bool>,
+    is_error: Option<bool>,
+    exit_code: Option<i64>,
     status_state: Option<String>,
     status_detail: Option<String>,
     active_call_id: Option<String>,
@@ -409,6 +429,9 @@ impl Default for V2Body {
             default_role: "agent",
             tool_name: None,
             call_id: None,
+            ok: None,
+            is_error: None,
+            exit_code: None,
             status_state: None,
             status_detail: None,
             active_call_id: None,
@@ -517,6 +540,9 @@ fn persist_message(
                     event_kind: classified.event_kind.clone(),
                     tool_name: classified.tool_name.clone(),
                     call_id: classified.call_id.clone(),
+                    ok: classified.ok,
+                    is_error: classified.is_error,
+                    exit_code: classified.exit_code,
                 },
             )?;
             // An `error` event also records the short cause on the status surface
@@ -873,6 +899,44 @@ mod tests {
         assert_eq!(tr.event_kind.as_deref(), Some("tool_result"));
         assert_eq!(tr.body, "file.rs");
         assert_eq!(tr.call_id.as_deref(), Some("c1"));
+        // A successful result carries ok=true / is_error=false, exit_code absent.
+        assert_eq!(tr.ok, Some(true));
+        assert_eq!(tr.is_error, Some(false));
+        assert_eq!(tr.exit_code, None);
+
+        // A FAILED tool_result carries the outcome so the renderer can flag it red.
+        let tr_err = classify_message(
+            v2_env(
+                "tool_result",
+                r#"{ "call_id": "c1", "ok": false, "is_error": true, "exit_code": 1, "output": "boom" }"#,
+                "w2",
+                "agent",
+            ),
+            "2026-07-05T09:00:00Z",
+        );
+        assert_eq!(tr_err.ok, Some(false));
+        assert_eq!(tr_err.is_error, Some(true));
+        assert_eq!(tr_err.exit_code, Some(1));
+
+        // Older/partial payloads may omit `ok`; keep it unknown instead of
+        // defaulting to failure.
+        let tr_unknown = classify_message(
+            v2_env(
+                "tool_result",
+                r#"{ "call_id": "c1", "is_error": false, "output": "done" }"#,
+                "w2",
+                "agent",
+            ),
+            "2026-07-05T09:00:00Z",
+        );
+        assert_eq!(tr_unknown.ok, None);
+        assert_eq!(tr_unknown.is_error, Some(false));
+        assert_eq!(tr_unknown.exit_code, None);
+
+        // Non-tool_result rows leave the outcome fields unset.
+        assert_eq!(tc.ok, None);
+        assert_eq!(tc.is_error, None);
+        assert_eq!(am.ok, None);
     }
 
     #[test]

src/openhuman/orchestration/schemas.rs

@@ -205,6 +205,9 @@ struct SessionSummary {
     chat_kind: String,
     last_message_at: String,
     unread: i64,
+    /// Total persisted messages in the session (all kinds), for the roster's
+    /// per-session count. `0` for the pinned windows / a freshly created session.
+    message_count: i64,
     active: bool,
     pinned: bool,
 }
@@ -333,6 +336,7 @@ fn summarize(
     unread: i64,
     pinned: bool,
     current_task: Option<String>,
+    message_count: i64,
 ) -> SessionSummary {
     let chat_kind = chat_kind_for_session(&session.session_id);
     let active = pinned || is_active(&session.last_message_at);
@@ -342,6 +346,7 @@ fn summarize(
         chat_kind: chat_kind.as_str().to_string(),
         active,
         unread,
+        message_count,
         pinned,
         harness_type,
         status,
@@ -362,6 +367,8 @@ fn handle_sessions_list(_params: Map<String, Value>) -> ControllerFuture {
         let config = load_config("sessions_list").await?;
         let sessions = store::with_connection(&config.workspace_dir, |conn| {
             let rows = store::list_sessions(conn)?;
+            let visible_counts = store::visible_message_counts(conn)?;
+            let pinned_visible_counts = store::visible_message_counts_by_session(conn)?;
             let mut out: Vec<SessionSummary> = Vec::with_capacity(rows.len() + 2);
             let mut have_master = false;
             let mut have_subconscious = false;
@@ -394,7 +401,24 @@ fn handle_sessions_list(_params: Map<String, Value>) -> ControllerFuture {
                         .map(|body| task_preview(&body)),
                     }
                 };
-                out.push(summarize(session, unread, pinned, current_task));
+                let message_count = if pinned {
+                    pinned_visible_counts
+                        .get(&session.session_id)
+                        .copied()
+                        .unwrap_or(0)
+                } else {
+                    visible_counts
+                        .get(&(session.agent_id.clone(), session.session_id.clone()))
+                        .copied()
+                        .unwrap_or(0)
+                };
+                out.push(summarize(
+                    session,
+                    unread,
+                    pinned,
+                    current_task,
+                    message_count,
+                ));
             }
             // Ensure the pinned windows always exist even before any traffic.
             if !have_master {
@@ -442,7 +466,7 @@ fn handle_sessions_create(params: Map<String, Value>) -> ControllerFuture {
         })
         .map_err(|e| format!("sessions_create: {e}"))?;
         super::bus::notify_orchestration_message(&agent_id, &session_id, "session");
-        to_json(serde_json::json!({ "session": summarize(session, 0, false, None) }))
+        to_json(serde_json::json!({ "session": summarize(session, 0, false, None, 0) }))
     })
 }
 
@@ -459,6 +483,7 @@ fn pinned_placeholder(session_id: &str) -> SessionSummary {
         chat_kind: chat_kind_for_session(session_id).as_str().to_string(),
         last_message_at: String::new(),
         unread: 0,
+        message_count: 0,
         active: true,
         pinned: true,
     }
@@ -1170,9 +1195,10 @@ mod tests {
             ..Default::default()
         };
         // current_task is threaded from the handler (status.detail preferred there).
-        let summary = summarize(session, 0, false, Some("approve rm -rf".to_string()));
+        let summary = summarize(session, 0, false, Some("approve rm -rf".to_string()), 7);
         assert_eq!(summary.status, "waiting-approval");
         assert_eq!(summary.current_task.as_deref(), Some("approve rm -rf"));
+        assert_eq!(summary.message_count, 7);
     }
 
     #[test]
@@ -1200,10 +1226,11 @@ mod tests {
             last_message_at: "2020-01-01T00:00:00Z".to_string(),
             ..Default::default()
         };
-        let summary = summarize(session, 2, false, Some("drafting cards".to_string()));
+        let summary = summarize(session, 2, false, Some("drafting cards".to_string()), 12);
         assert_eq!(summary.harness_type.as_deref(), Some("claude"));
         assert_eq!(summary.status, "stopped");
         assert_eq!(summary.current_task.as_deref(), Some("drafting cards"));
+        assert_eq!(summary.message_count, 12);
         assert!(!summary.active);
 
         // A pinned window is always active → idle, and carries no harness/task.