admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
This commit is contained in:
sanil-23
2026-07-08 19:14:38 -07:00
committed by GitHub
co-authored by Steven Enamakel
parent 55c33b990f
commit 9175603a9c
4 changed files with 288 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openhuman/orchestration/ingest.rs
+64
View File
@@ -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
+31 -4
View File
@@ -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.
src/openhuman/orchestration/store.rs
+169 -9
View File
@@ -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};
@@ -43,8 +44,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,
@@ -56,7 +58,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
@@ -237,6 +242,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")?;
// v2 `session_info` enrichment columns (spec §4). Same per-column,
// freshness-independent guard as the run-state block above: a fresh DB has
@@ -373,8 +382,8 @@ pub fn insert_message(conn: &Connection, m: &OrchestrationMessage) -> Result<boo
let changed = conn.execute(
"INSERT OR IGNORE INTO messages
(id, agent_id, session_id, chat_kind, role, body, timestamp, seq,
event_kind, tool_name, call_id)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
event_kind, tool_name, call_id, ok, is_error, exit_code)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
params![
m.id,
m.agent_id,
@@ -387,6 +396,9 @@ pub fn insert_message(conn: &Connection, m: &OrchestrationMessage) -> Result<boo
m.event_kind,
m.tool_name,
m.call_id,
m.ok,
m.is_error,
m.exit_code,
],
)?;
Ok(changed > 0)
@@ -401,6 +413,67 @@ 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<i64> {
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),
)?)
}
/// 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<i64> {
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),
)?)
}
/// Transcript-visible message counts keyed by `(agent_id, session_id)`.
pub fn visible_message_counts(conn: &Connection) -> Result<HashMap<(String, String), i64>> {
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::<std::result::Result<HashMap<_, _>, _>>()?;
Ok(rows)
}
/// Transcript-visible message counts keyed by `session_id` for pinned chats.
pub fn visible_message_counts_by_session(conn: &Connection) -> Result<HashMap<String, i64>> {
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::<std::result::Result<HashMap<_, _>, _>>()?;
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
@@ -482,7 +555,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', 'session_info'))
@@ -496,7 +569,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', 'session_info'))
@@ -528,6 +601,9 @@ fn map_message_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<OrchestrationMes
event_kind: row.get(8)?,
tool_name: row.get(9)?,
call_id: row.get(10)?,
ok: row.get(11)?,
is_error: row.get(12)?,
exit_code: row.get(13)?,
})
}
@@ -686,7 +762,7 @@ pub fn list_recent_messages(
) -> Result<Vec<OrchestrationMessage>> {
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",
)?;
@@ -1060,6 +1136,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();
@@ -1172,6 +1280,28 @@ 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);
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);
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!(
@@ -1504,7 +1634,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
@@ -1550,9 +1680,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)?,
@@ -1569,6 +1708,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(
@@ -1584,6 +1726,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();
src/openhuman/orchestration/types.rs
+24 -3
View File
@@ -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<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub exit_code: Option<i64>,
#[serde(default)]
@@ -524,6 +524,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<String>,
/// 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<bool>,
/// 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<bool>,
/// 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<i64>,
}
#[cfg(test)]
@@ -665,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(),
@@ -673,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" }"#,