From c1b0c150f23da28af37745f022043b3b4da39648 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 8 Jul 2026 21:13:25 +0530 Subject: [PATCH 1/6] feat(orchestration): surface tool_result outcome + per-session message_count OrchestrationMessage now carries ok/is_error/exit_code decoded from the v2 tool_result payload, persisted via additive, migration-guarded messages columns. A failed tool run is therefore distinguishable from a successful one on read instead of both collapsing to plain output. SessionSummary gains message_count (total persisted messages per session) via the existing store::count_messages, for the roster's per-session count. Both unblock the orchestration UI work (failed-result styling + session counts) without touching any other surface. --- src/openhuman/orchestration/ingest.rs | 49 ++++++++++++++++++++++ src/openhuman/orchestration/schemas.rs | 24 +++++++++-- src/openhuman/orchestration/store.rs | 57 +++++++++++++++++++++++--- src/openhuman/orchestration/types.rs | 11 +++++ 4 files changed, 131 insertions(+), 10 deletions(-) diff --git a/src/openhuman/orchestration/ingest.rs b/src/openhuman/orchestration/ingest.rs index 2d096ae4fd..a4c1470dbc 100644 --- a/src/openhuman/orchestration/ingest.rs +++ b/src/openhuman/orchestration/ingest.rs @@ -86,6 +86,10 @@ struct ClassifiedMessage { tool_name: Option, /// v2 `call_id` correlating `tool_result` → `tool_call`. call_id: Option, + /// v2 `tool_result` outcome (all `None` off a `tool_result` row). + ok: Option, + is_error: Option, + exit_code: Option, /// 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, @@ -156,6 +160,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, @@ -193,6 +200,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, @@ -241,6 +251,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 = Some(p.ok); + b.is_error = Some(p.is_error); + b.exit_code = p.exit_code; } HarnessEventKind::ApprovalRequest(p) => { b.body = p.display; @@ -301,6 +315,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, @@ -316,6 +333,9 @@ struct V2Body { default_role: &'static str, tool_name: Option, call_id: Option, + ok: Option, + is_error: Option, + exit_code: Option, status_state: Option, status_detail: Option, active_call_id: Option, @@ -334,6 +354,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, @@ -430,6 +453,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 @@ -775,6 +801,29 @@ 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)); + + // 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] diff --git a/src/openhuman/orchestration/schemas.rs b/src/openhuman/orchestration/schemas.rs index 5fab2b74e6..f52e733fe9 100644 --- a/src/openhuman/orchestration/schemas.rs +++ b/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, + 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, @@ -394,7 +399,15 @@ fn handle_sessions_list(_params: Map) -> ControllerFuture { .map(|body| task_preview(&body)), } }; - out.push(summarize(session, unread, pinned, current_task)); + let message_count = + store::count_messages(conn, &session.agent_id, &session.session_id)?; + 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 +455,7 @@ fn handle_sessions_create(params: Map) -> 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 +472,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 +1184,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 +1215,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. diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index 741b2d46f2..1c7bfe5995 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -47,7 +47,10 @@ const SCHEMA_DDL: &str = " seq INTEGER NOT NULL DEFAULT 0, event_kind TEXT, tool_name TEXT, - call_id TEXT + call_id TEXT, + ok INTEGER, + is_error INTEGER, + exit_code INTEGER ); CREATE INDEX IF NOT EXISTS idx_messages_session @@ -228,6 +231,10 @@ fn migrate(conn: &Connection) -> Result<()> { add_column_if_missing(conn, "messages", "event_kind", "TEXT")?; add_column_if_missing(conn, "messages", "tool_name", "TEXT")?; add_column_if_missing(conn, "messages", "call_id", "TEXT")?; + // v2 tool_result outcome — additive, existing rows default NULL. + add_column_if_missing(conn, "messages", "ok", "INTEGER")?; + add_column_if_missing(conn, "messages", "is_error", "INTEGER")?; + add_column_if_missing(conn, "messages", "exit_code", "INTEGER")?; Ok(()) } @@ -312,8 +319,8 @@ pub fn insert_message(conn: &Connection, m: &OrchestrationMessage) -> Result Result 0) @@ -420,7 +430,7 @@ pub fn list_messages_by_session( Some(before) => { let mut stmt = conn.prepare( "SELECT id, agent_id, session_id, chat_kind, role, body, timestamp, seq, - event_kind, tool_name, call_id + event_kind, tool_name, call_id, ok, is_error, exit_code FROM messages WHERE session_id = ?1 AND timestamp < ?2 AND (event_kind IS NULL OR event_kind NOT IN ('status', 'lifecycle', 'unknown')) @@ -434,7 +444,7 @@ pub fn list_messages_by_session( None => { let mut stmt = conn.prepare( "SELECT id, agent_id, session_id, chat_kind, role, body, timestamp, seq, - event_kind, tool_name, call_id + event_kind, tool_name, call_id, ok, is_error, exit_code FROM messages WHERE session_id = ?1 AND (event_kind IS NULL OR event_kind NOT IN ('status', 'lifecycle', 'unknown')) @@ -466,6 +476,9 @@ fn map_message_row(row: &rusqlite::Row<'_>) -> rusqlite::Result Result> { let mut stmt = conn.prepare( "SELECT id, agent_id, session_id, chat_kind, role, body, timestamp, seq, - event_kind, tool_name, call_id + event_kind, tool_name, call_id, ok, is_error, exit_code FROM messages WHERE agent_id = ?1 AND session_id = ?2 ORDER BY timestamp DESC, seq DESC LIMIT ?3", )?; @@ -991,6 +1004,38 @@ mod tests { .unwrap(); } + #[test] + fn persists_and_reads_back_tool_result_outcome() { + let tmp = tempfile::tempdir().unwrap(); + with_connection(tmp.path(), |conn| { + upsert_session(conn, &session("@a", "h1", 1))?; + let failed = OrchestrationMessage { + event_kind: Some("tool_result".into()), + tool_name: Some("Bash".into()), + call_id: Some("c1".into()), + ok: Some(false), + is_error: Some(true), + exit_code: Some(1), + ..msg("m1", "@a", "h1", 1) + }; + assert!(insert_message(conn, &failed)?); + let back = list_recent_messages(conn, "@a", "h1", 10)?; + assert_eq!(back.len(), 1); + assert_eq!(back[0].ok, Some(false)); + assert_eq!(back[0].is_error, Some(true)); + assert_eq!(back[0].exit_code, Some(1)); + // A plain message leaves the outcome columns NULL → None on read. + assert!(insert_message(conn, &msg("m2", "@a", "h1", 2))?); + let plain = list_recent_messages(conn, "@a", "h1", 10)?; + let m2 = plain.iter().find(|m| m.id == "m2").unwrap(); + assert_eq!(m2.ok, None); + assert_eq!(m2.is_error, None); + assert_eq!(m2.exit_code, None); + Ok(()) + }) + .unwrap(); + } + #[test] fn latest_message_preview_returns_newest_or_none() { let tmp = tempfile::tempdir().unwrap(); diff --git a/src/openhuman/orchestration/types.rs b/src/openhuman/orchestration/types.rs index bfd49d99d6..03be26f163 100644 --- a/src/openhuman/orchestration/types.rs +++ b/src/openhuman/orchestration/types.rs @@ -458,6 +458,17 @@ pub struct OrchestrationMessage { /// v2 correlation id linking a `tool_result` back to its `tool_call`. #[serde(default, skip_serializing_if = "Option::is_none")] pub call_id: Option, + /// v2 `tool_result.ok` — whether the tool call succeeded. `None` on every row + /// that is not a `tool_result` (so the renderer can distinguish a failed run + /// from a successful one instead of both reading as plain output). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ok: Option, + /// v2 `tool_result.is_error` — the harness flagged the result as an error. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub is_error: Option, + /// v2 `tool_result.exit_code` — process exit code when the tool was a command. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub exit_code: Option, } #[cfg(test)] From 33743fafe2eb928a8b27d16b38c3d8f88ae6a50b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 8 Jul 2026 16:42:18 -0700 Subject: [PATCH 2/6] test(orchestration): cover outcome column migration --- src/openhuman/orchestration/ingest.rs | 2 +- src/openhuman/orchestration/store.rs | 32 ++++++++++++++++++++++++++- 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/src/openhuman/orchestration/ingest.rs b/src/openhuman/orchestration/ingest.rs index 45e5e47d1b..4dc905abca 100644 --- a/src/openhuman/orchestration/ingest.rs +++ b/src/openhuman/orchestration/ingest.rs @@ -104,7 +104,7 @@ struct ClassifiedMessage { tool_name: Option, /// v2 `call_id` correlating `tool_result` → `tool_call`. call_id: Option, - /// v2 `tool_result` outcome (all `None` off a `tool_result` row). + /// v2 `tool_result` outcome (`None` for rows that are not `tool_result`). ok: Option, is_error: Option, exit_code: Option, diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index 9bd1e3170b..d10d4aa59b 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -1549,7 +1549,7 @@ mod tests { } #[test] - fn migrates_pre_v2_schema_by_adding_status_and_event_columns() { + fn migrates_pre_v2_schema_by_adding_session_and_message_columns() { // A store created before the harness-session-v2 receiver: the sessions and // messages tables lack the new run-state / event columns. Opening through // `with_connection` must add them additively (existing rows read NULL) and @@ -1595,9 +1595,18 @@ mod tests { ("sessions", "status_state"), ("sessions", "current_detail"), ("sessions", "active_call_id"), + ("sessions", "title"), + ("sessions", "model"), + ("sessions", "handle"), + ("sessions", "repo"), + ("sessions", "branch"), + ("sessions", "capabilities"), ("messages", "event_kind"), ("messages", "tool_name"), ("messages", "call_id"), + ("messages", "ok"), + ("messages", "is_error"), + ("messages", "exit_code"), ] { assert!( column_exists(conn, table, column)?, @@ -1614,6 +1623,9 @@ mod tests { assert_eq!(msgs.len(), 1); assert_eq!(msgs[0].body, "legacy body"); assert_eq!(msgs[0].event_kind, None); + assert_eq!(msgs[0].ok, None); + assert_eq!(msgs[0].is_error, None); + assert_eq!(msgs[0].exit_code, None); // And the upgraded schema accepts writes that populate the new fields. upsert_session( @@ -1629,6 +1641,24 @@ mod tests { assert_eq!(updated.status_state.as_deref(), Some("running")); assert_eq!(updated.current_detail.as_deref(), Some("compiling")); assert_eq!(updated.active_call_id.as_deref(), Some("call-1")); + let tool_result = OrchestrationMessage { + event_kind: Some("tool_result".into()), + tool_name: Some("Bash".into()), + call_id: Some("call-1".into()), + ok: Some(false), + is_error: Some(true), + exit_code: Some(1), + ..msg("m-new", "@a", "h-old", 2) + }; + assert!(insert_message(conn, &tool_result)?); + let upgraded_messages = list_recent_messages(conn, "@a", "h-old", 10)?; + let saved = upgraded_messages + .iter() + .find(|m| m.id == "m-new") + .expect("upgraded schema stores outcome fields"); + assert_eq!(saved.ok, Some(false)); + assert_eq!(saved.is_error, Some(true)); + assert_eq!(saved.exit_code, Some(1)); Ok(()) }) .unwrap(); From 6d90afaf8aa61dace9633919a938cfaf51793353 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 8 Jul 2026 17:13:12 -0700 Subject: [PATCH 3/6] fix(orchestration): preserve visible message outcome semantics --- src/openhuman/orchestration/ingest.rs | 17 ++++++++++++++++- src/openhuman/orchestration/schemas.rs | 2 +- src/openhuman/orchestration/store.rs | 22 ++++++++++++++++++++-- src/openhuman/orchestration/types.rs | 16 +++++++++++++--- 4 files changed, 50 insertions(+), 7 deletions(-) diff --git a/src/openhuman/orchestration/ingest.rs b/src/openhuman/orchestration/ingest.rs index 4dc905abca..8ca35630c9 100644 --- a/src/openhuman/orchestration/ingest.rs +++ b/src/openhuman/orchestration/ingest.rs @@ -314,7 +314,7 @@ fn classify_v2(env: SessionEnvelopeV2, fallback_timestamp: &str) -> ClassifiedMe 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 = Some(p.ok); + b.ok = p.ok; b.is_error = Some(p.is_error); b.exit_code = p.exit_code; } @@ -918,6 +918,21 @@ mod tests { 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); diff --git a/src/openhuman/orchestration/schemas.rs b/src/openhuman/orchestration/schemas.rs index f52e733fe9..613a2f6a94 100644 --- a/src/openhuman/orchestration/schemas.rs +++ b/src/openhuman/orchestration/schemas.rs @@ -400,7 +400,7 @@ fn handle_sessions_list(_params: Map) -> ControllerFuture { } }; let message_count = - store::count_messages(conn, &session.agent_id, &session.session_id)?; + store::count_visible_messages(conn, &session.agent_id, &session.session_id)?; out.push(summarize( session, unread, diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index d10d4aa59b..63d3c5c1da 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -43,8 +43,9 @@ const SCHEMA_DDL: &str = " ); -- `event_kind`/`tool_name`/`call_id` carry the v2 per-message event shape - -- (`event.kind` + tool identity/correlation). Nullable and additive; v1 and - -- pinned master/subconscious rows leave them NULL. + -- (`event.kind` + tool identity/correlation). `ok`/`is_error`/`exit_code` + -- carry the `tool_result` outcome. Nullable and additive; v1 and pinned + -- master/subconscious rows leave them NULL. CREATE TABLE IF NOT EXISTS messages ( id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, @@ -411,6 +412,18 @@ pub fn count_messages(conn: &Connection, agent_id: &str, session_id: &str) -> Re )?) } +/// Count transcript-visible messages for a session, using the same visibility +/// predicate as message reads, unread counts, and roster previews. +pub fn count_visible_messages(conn: &Connection, agent_id: &str, session_id: &str) -> Result { + Ok(conn.query_row( + "SELECT COUNT(*) FROM messages WHERE agent_id = ?1 AND session_id = ?2 + AND (event_kind IS NULL + OR event_kind NOT IN ('status', 'lifecycle', 'unknown', 'session_info'))", + params![agent_id, session_id], + |row| row.get(0), + )?) +} + /// The next monotonic per-session ingest ordinal: `MAX(seq) + 1` over the /// session's messages (`1` for the first message). Stamped at persist time so /// the wake idempotence cursor rides a strictly-increasing value instead of the @@ -1217,6 +1230,11 @@ mod tests { // Unread (cursor at 0) counts the two visible rows, not the 4 hidden. assert_eq!(unread_count(conn, "h1")?, 2); + // UI session summaries use the same visibility predicate as unread and + // transcript reads, while the raw observability count still includes + // all persisted relay-dedupe rows. + assert_eq!(count_visible_messages(conn, "@a", "h1")?, 2); + assert_eq!(count_messages(conn, "@a", "h1")?, 6); // Roster preview skips the hidden rows → newest visible is the call. assert_eq!( diff --git a/src/openhuman/orchestration/types.rs b/src/openhuman/orchestration/types.rs index 71b598fe88..8243605652 100644 --- a/src/openhuman/orchestration/types.rs +++ b/src/openhuman/orchestration/types.rs @@ -272,8 +272,8 @@ pub struct ToolCallPayload { pub struct ToolResultPayload { #[serde(default)] pub call_id: String, - #[serde(default)] - pub ok: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ok: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub exit_code: Option, #[serde(default)] @@ -676,7 +676,7 @@ mod tests { tr.event.decoded(), ToolResult(ToolResultPayload { call_id: "c1".into(), - ok: true, + ok: Some(true), exit_code: Some(0), is_error: false, output: "done".into(), @@ -684,6 +684,16 @@ mod tests { }) ); + let tr_unknown = SessionEnvelopeV2::parse(&v2_wire( + "tool_result", + r#"{ "call_id": "c1", "is_error": false, "output": "done" }"#, + )) + .unwrap(); + match tr_unknown.event.decoded() { + ToolResult(p) => assert_eq!(p.ok, None), + other => panic!("expected tool_result, got {other:?}"), + } + let ar = SessionEnvelopeV2::parse(&v2_wire( "approval_request", r#"{ "call_id": "c9", "tool_name": "rm", "display": "rm -rf x", "reason": "destructive" }"#, From f558fb54760c84ad46662f42b68ab42df61fb05f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 8 Jul 2026 17:18:39 -0700 Subject: [PATCH 4/6] chore(pr-fix): apply formatting --- app/src-tauri/src/telegram_scanner/mod.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/app/src-tauri/src/telegram_scanner/mod.rs b/app/src-tauri/src/telegram_scanner/mod.rs index e55e21be0e..92a1ad9e38 100644 --- a/app/src-tauri/src/telegram_scanner/mod.rs +++ b/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, From d935f35060eb24ff206f717049198afef993d212 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 8 Jul 2026 18:53:44 -0700 Subject: [PATCH 5/6] fix(orchestration): scope pinned message counts --- src/openhuman/orchestration/schemas.rs | 7 +++++-- src/openhuman/orchestration/store.rs | 18 ++++++++++++++++++ 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/src/openhuman/orchestration/schemas.rs b/src/openhuman/orchestration/schemas.rs index 613a2f6a94..249e0c872c 100644 --- a/src/openhuman/orchestration/schemas.rs +++ b/src/openhuman/orchestration/schemas.rs @@ -399,8 +399,11 @@ fn handle_sessions_list(_params: Map) -> ControllerFuture { .map(|body| task_preview(&body)), } }; - let message_count = - store::count_visible_messages(conn, &session.agent_id, &session.session_id)?; + let message_count = if pinned { + store::count_visible_messages_by_session(conn, &session.session_id)? + } else { + store::count_visible_messages(conn, &session.agent_id, &session.session_id)? + }; out.push(summarize( session, unread, diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index 63d3c5c1da..f60f64c4ee 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -424,6 +424,18 @@ pub fn count_visible_messages(conn: &Connection, agent_id: &str, session_id: &st )?) } +/// Count transcript-visible messages for a pinned chat, whose transcript is +/// scoped only by `session_id` and can include rows from multiple peers. +pub fn count_visible_messages_by_session(conn: &Connection, session_id: &str) -> Result { + Ok(conn.query_row( + "SELECT COUNT(*) FROM messages WHERE session_id = ?1 + AND (event_kind IS NULL + OR event_kind NOT IN ('status', 'lifecycle', 'unknown', 'session_info'))", + params![session_id], + |row| row.get(0), + )?) +} + /// The next monotonic per-session ingest ordinal: `MAX(seq) + 1` over the /// session's messages (`1` for the first message). Stamped at persist time so /// the wake idempotence cursor rides a strictly-increasing value instead of the @@ -1236,6 +1248,12 @@ mod tests { assert_eq!(count_visible_messages(conn, "@a", "h1")?, 2); assert_eq!(count_messages(conn, "@a", "h1")?, 6); + let mut other_peer = msg("other-peer", "@b", "h1", 7); + other_peer.timestamp = "2026-07-02T00:00:07Z".into(); + insert_message(conn, &other_peer)?; + assert_eq!(count_visible_messages(conn, "@a", "h1")?, 2); + assert_eq!(count_visible_messages_by_session(conn, "h1")?, 3); + // Roster preview skips the hidden rows → newest visible is the call. assert_eq!( latest_message_preview(conn, "@a", "h1")?.as_deref(), From cf122dc23fbeade1944bd39b0b4aa7c905762775 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 8 Jul 2026 18:58:22 -0700 Subject: [PATCH 6/6] fix(orchestration): aggregate visible message counts --- src/openhuman/orchestration/schemas.rs | 12 +++++-- src/openhuman/orchestration/store.rs | 49 ++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 2 deletions(-) diff --git a/src/openhuman/orchestration/schemas.rs b/src/openhuman/orchestration/schemas.rs index 249e0c872c..1d3d936bba 100644 --- a/src/openhuman/orchestration/schemas.rs +++ b/src/openhuman/orchestration/schemas.rs @@ -367,6 +367,8 @@ fn handle_sessions_list(_params: Map) -> 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 = Vec::with_capacity(rows.len() + 2); let mut have_master = false; let mut have_subconscious = false; @@ -400,9 +402,15 @@ fn handle_sessions_list(_params: Map) -> ControllerFuture { } }; let message_count = if pinned { - store::count_visible_messages_by_session(conn, &session.session_id)? + pinned_visible_counts + .get(&session.session_id) + .copied() + .unwrap_or(0) } else { - store::count_visible_messages(conn, &session.agent_id, &session.session_id)? + visible_counts + .get(&(session.agent_id.clone(), session.session_id.clone())) + .copied() + .unwrap_or(0) }; out.push(summarize( session, diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index f60f64c4ee..63b7788722 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -5,6 +5,7 @@ //! `is_workspace_internal_path`). Follows the subconscious/cron `with_connection` //! pattern. +use std::collections::HashMap; use std::path::Path; use anyhow::{Context, Result}; @@ -436,6 +437,43 @@ pub fn count_visible_messages_by_session(conn: &Connection, session_id: &str) -> )?) } +/// Transcript-visible message counts keyed by `(agent_id, session_id)`. +pub fn visible_message_counts(conn: &Connection) -> Result> { + let mut stmt = conn.prepare( + "SELECT agent_id, session_id, COUNT(*) + FROM messages + WHERE event_kind IS NULL + OR event_kind NOT IN ('status', 'lifecycle', 'unknown', 'session_info') + GROUP BY agent_id, session_id", + )?; + let rows = stmt + .query_map([], |row| { + Ok(( + (row.get::<_, String>(0)?, row.get::<_, String>(1)?), + row.get::<_, i64>(2)?, + )) + })? + .collect::, _>>()?; + Ok(rows) +} + +/// Transcript-visible message counts keyed by `session_id` for pinned chats. +pub fn visible_message_counts_by_session(conn: &Connection) -> Result> { + let mut stmt = conn.prepare( + "SELECT session_id, COUNT(*) + FROM messages + WHERE event_kind IS NULL + OR event_kind NOT IN ('status', 'lifecycle', 'unknown', 'session_info') + GROUP BY session_id", + )?; + let rows = stmt + .query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)) + })? + .collect::, _>>()?; + Ok(rows) +} + /// The next monotonic per-session ingest ordinal: `MAX(seq) + 1` over the /// session's messages (`1` for the first message). Stamped at persist time so /// the wake idempotence cursor rides a strictly-increasing value instead of the @@ -1253,6 +1291,17 @@ mod tests { insert_message(conn, &other_peer)?; assert_eq!(count_visible_messages(conn, "@a", "h1")?, 2); assert_eq!(count_visible_messages_by_session(conn, "h1")?, 3); + let by_agent_session = visible_message_counts(conn)?; + assert_eq!( + by_agent_session.get(&("@a".to_string(), "h1".to_string())), + Some(&2) + ); + assert_eq!( + by_agent_session.get(&("@b".to_string(), "h1".to_string())), + Some(&1) + ); + let by_session = visible_message_counts_by_session(conn)?; + assert_eq!(by_session.get("h1"), Some(&3)); // Roster preview skips the hidden rows → newest visible is the call. assert_eq!(