From aedc907e92e030b68a7866a2927bcdd3c4218167 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 8 Jul 2026 19:47:45 +0530 Subject: [PATCH] feat(orchestration): receive session_info harness events into the v2 mirror MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the session_info HarnessEventKind to OpenHuman's hand-rolled v2 harness mirror so a wrapped agent's session intro/announce (spec §2b) is decoded, classified, and folded into the durable session record. - types.rs: SessionInfo variant (first, adjacently tagged) + SessionInfoPayload struct with wire-byte-identical snake_case field names to the TS emitter (spec §2a); enrich OrchestrationSession with title/model/handle/repo/branch/ capabilities. - ingest.rs: classify_v2 SessionInfo arm — advances_seq=false (seq=0), title doubles as the row body, model falls back to event.model, payload folds onto the session record. - store.rs: additive nullable session columns (capabilities as a JSON array, empty→NULL so COALESCE preserves a prior list); upsert COALESCEs the intro metadata so ordinary events never wipe it and a resumed=true re-intro refreshes rather than duplicates; add session_info to the no-bubble filter so the intro never renders as an in-thread chat bubble or counts as unread. Receiver behavior (spec §4): upsert keyed on wrapper_session_id, lazy-create on the first event of any kind (session_info is enrichment, not a prerequisite), idempotent on event.id, resumed=true updates in place. Tests: session_info decode + wire-field-name round-trip, model/event.model fallback, lazy-create + idempotent + resumed-refresh upsert, capabilities JSON codec (null-safe), and no-bubble hiding from thread/unread. --- src/openhuman/orchestration/ingest.rs | 202 ++++++++++++++++++++++++++ src/openhuman/orchestration/store.rs | 147 +++++++++++++++++-- src/openhuman/orchestration/types.rs | 142 ++++++++++++++++++ 3 files changed, 478 insertions(+), 13 deletions(-) diff --git a/src/openhuman/orchestration/ingest.rs b/src/openhuman/orchestration/ingest.rs index 2d096ae4fd..bdc98f9b11 100644 --- a/src/openhuman/orchestration/ingest.rs +++ b/src/openhuman/orchestration/ingest.rs @@ -103,6 +103,19 @@ struct ClassifiedMessage { /// (e.g. `running_tool` → `idle`). Content events leave this false so the /// COALESCE upsert preserves the last status instead of wiping it. authoritative_status: bool, + // ── session_info enrichment (only a `session_info` event sets these) ────── + /// `session_info.title` → the session row's `title`. + title: Option, + /// `session_info.model` (or `event.model` fallback) → the session row's `model`. + model: Option, + /// `session_info.handle` → the session row's `handle`. + handle: Option, + /// `session_info.repo` → the session row's `repo`. + repo: Option, + /// `session_info.branch` → the session row's `branch`. + branch: Option, + /// `session_info.capabilities` → the session row's `capabilities`. + capabilities: Vec, } /// True for streams that carry ciphertext DM envelopes worth ingesting. @@ -161,6 +174,13 @@ fn classify_message(plaintext: String, fallback_timestamp: &str) -> ClassifiedMe active_call_id: None, advances_seq: true, authoritative_status: false, + // session_info enrichment — only a v2 `session_info` event populates these. + title: None, + model: None, + handle: None, + repo: None, + branch: None, + capabilities: Vec::new(), } } @@ -198,6 +218,13 @@ fn classify_v1(env: SessionEnvelopeV1, fallback_timestamp: &str) -> ClassifiedMe active_call_id: None, advances_seq: true, authoritative_status: false, + // session_info enrichment — only a v2 `session_info` event populates these. + title: None, + model: None, + handle: None, + repo: None, + branch: None, + capabilities: Vec::new(), } } @@ -219,10 +246,27 @@ fn classify_v2(env: SessionEnvelopeV2, fallback_timestamp: &str) -> ClassifiedMe let kind_str = env.event.kind.clone(); let wire_role = env.event.role.clone(); let seq = env.event.seq; + // `session_info.model` is optional; fall back to the frame's `event.model`. + let event_model = env.event.model.clone(); // Per-kind body + tool/status fields + wake disposition. let mut b = V2Body::default(); match env.event.decoded() { + HarnessEventKind::SessionInfo(p) => { + // Session intro/announce: enrichment, not a chat bubble. Persist a + // seq=0 row (id-dedupe + ack) that does NOT advance the wake ordinal — + // same disposition as status/lifecycle — and fold the payload into the + // session record fields. `title` doubles as the row body so the + // session_info message carries a human-readable trace of the intro. + b.body = p.title.clone().unwrap_or_default(); + b.advances_seq = false; + b.title = p.title; + b.model = p.model.or(event_model); + b.handle = p.handle; + b.repo = p.repo; + b.branch = p.branch; + b.capabilities = p.capabilities; + } HarnessEventKind::UserPrompt(p) => { b.body = p.text; b.default_role = "owner"; @@ -306,6 +350,12 @@ fn classify_v2(env: SessionEnvelopeV2, fallback_timestamp: &str) -> ClassifiedMe active_call_id: b.active_call_id, advances_seq: b.advances_seq, authoritative_status: b.authoritative_status, + title: b.title, + model: b.model, + handle: b.handle, + repo: b.repo, + branch: b.branch, + capabilities: b.capabilities, } } @@ -325,6 +375,13 @@ struct V2Body { /// `decoded()` folded to `Unknown`: stored as the literal `"unknown"` so the /// store readers keep it out of the thread instead of leaking the raw kind. kind_override: Option<&'static str>, + // ── session_info enrichment (only the `SessionInfo` arm sets these) ─────── + title: Option, + model: Option, + handle: Option, + repo: Option, + branch: Option, + capabilities: Vec, } impl Default for V2Body { @@ -340,6 +397,12 @@ impl Default for V2Body { advances_seq: true, authoritative_status: false, kind_override: None, + title: None, + model: None, + handle: None, + repo: None, + branch: None, + capabilities: Vec::new(), } } } @@ -401,6 +464,12 @@ fn persist_message( status_state: classified.status_state.clone(), current_detail: classified.status_detail.clone(), active_call_id: classified.active_call_id.clone(), + title: classified.title.clone(), + model: classified.model.clone(), + handle: classified.handle.clone(), + repo: classified.repo.clone(), + branch: classified.branch.clone(), + capabilities: classified.capabilities.clone(), }, )?; // An authoritative `status` snapshot OWNS the run-state columns and may @@ -850,6 +919,139 @@ mod tests { assert!(uk.body.contains("flux")); } + #[test] + fn classifies_v2_session_info_as_session_enrichment() { + // A full session_info: enrichment fields map onto the session record, the + // title doubles as the row body, and it does NOT advance the wake ordinal + // (same disposition as status/lifecycle). + let si = classify_message( + v2_env( + "session_info", + r#"{ "agent_address": "A1", "handle": "@alice", "title": "myrepo · feat/x", + "repo": "org/myrepo", "branch": "feat/x", "model": "claude-opus-4-8", + "capabilities": ["agent_message", "tool_call"], "resumed": false, + "started_at": "2026-07-08T00:00:00Z" }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:00Z", + ); + assert_eq!(si.chat_kind, ChatKind::Session); + assert_eq!(si.session_id, "w2"); + assert_eq!(si.event_kind.as_deref(), Some("session_info")); + assert!( + !si.advances_seq, + "session_info must not advance the wake ordinal" + ); + assert!(!si.authoritative_status); + assert_eq!(si.body, "myrepo · feat/x", "title doubles as the row body"); + assert_eq!(si.title.as_deref(), Some("myrepo · feat/x")); + assert_eq!(si.model.as_deref(), Some("claude-opus-4-8")); + assert_eq!(si.handle.as_deref(), Some("@alice")); + assert_eq!(si.repo.as_deref(), Some("org/myrepo")); + assert_eq!(si.branch.as_deref(), Some("feat/x")); + assert_eq!(si.capabilities, vec!["agent_message", "tool_call"]); + } + + #[test] + fn session_info_model_falls_back_to_event_model() { + // The payload omits `model`; the frame's `event.model` fills it (spec §2b: + // "may also ride on event.model"). + let wire = format!( + r#"{{ + "envelope_version": "tinyplace.harness.session.v2", + "version": 2, + "scope": {{ "type": "folder", "key": "my-repo", "cwd": "/w", + "wrapper_session_id": "w2", "harness_session_id": "h2" }}, + "harness": {{ "provider": "claude", "command": "claude", "argv": [] }}, + "event": {{ "id": "e1", "seq": 0, "ts": "2026-07-05T01:00:00Z", + "model": "opus-from-frame", "role": "agent", + "kind": "session_info", + "payload": {{ "agent_address": "A1" }} }}, + "source": {{ "path": "p", "record_type": "assistant" }} + }}"# + ); + let si = classify_message(wire, "2026-07-05T09:00:00Z"); + assert_eq!(si.model.as_deref(), Some("opus-from-frame")); + assert!(si.title.is_none()); + assert!(si.capabilities.is_empty()); + } + + #[test] + fn persisting_session_info_enriches_session_lazily_and_idempotently() { + let tmp = tempfile::tempdir().unwrap(); + // session_info is the FIRST event for this session — it must lazy-create + // the record (enrichment, not a prerequisite), populate the metadata, and + // stamp last_seq = 0 (no spurious wake). + let si = classify_message( + v2_env( + "session_info", + r#"{ "agent_address": "A1", "handle": "@alice", "title": "Intro", + "repo": "org/myrepo", "branch": "feat/x", "model": "opus", + "capabilities": ["agent_message"], "resumed": false, + "started_at": "2026-07-08T00:00:00Z" }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:00Z", + ); + assert!(persist_message(tmp.path(), "si1", "@peer", &si, "now").unwrap()); + // Re-persisting the SAME event id is idempotent (dedup on the relay id). + assert!(!persist_message(tmp.path(), "si1", "@peer", &si, "now").unwrap()); + + store::with_connection(tmp.path(), |c| { + let s = store::load_session(c, "@peer", "w2")?.expect("session lazy-created"); + assert_eq!(s.title.as_deref(), Some("Intro")); + assert_eq!(s.model.as_deref(), Some("opus")); + assert_eq!(s.handle.as_deref(), Some("@alice")); + assert_eq!(s.repo.as_deref(), Some("org/myrepo")); + assert_eq!(s.branch.as_deref(), Some("feat/x")); + assert_eq!(s.capabilities, vec!["agent_message".to_string()]); + assert_eq!(s.last_seq, 0, "session_info must not advance last_seq"); + assert_eq!(store::count_messages(c, "@peer", "w2")?, 1); + Ok(()) + }) + .unwrap(); + + // A reconnect re-intro (resumed=true, NEW event id, refreshed title) + // UPDATES the record rather than duplicating it, and a content event in + // between never wipes the intro metadata (COALESCE). + let content = classify_message( + v2_env("agent_message", r#"{ "text": "working" }"#, "w2", "agent"), + "2026-07-05T09:01:00Z", + ); + assert!(persist_message(tmp.path(), "m1", "@peer", &content, "now").unwrap()); + let resumed = classify_message( + v2_env( + "session_info", + r#"{ "agent_address": "A1", "title": "Intro v2", "resumed": true, + "started_at": "2026-07-08T01:00:00Z" }"#, + "w2", + "agent", + ), + "2026-07-05T09:02:00Z", + ); + assert!(persist_message(tmp.path(), "si2", "@peer", &resumed, "now").unwrap()); + + store::with_connection(tmp.path(), |c| { + let s = store::load_session(c, "@peer", "w2")?.expect("session"); + assert_eq!( + s.title.as_deref(), + Some("Intro v2"), + "resumed intro refreshes title" + ); + // Untouched intro fields survive the content event + the thin re-intro. + assert_eq!(s.model.as_deref(), Some("opus")); + assert_eq!(s.capabilities, vec!["agent_message".to_string()]); + // Only the content event advanced the ordinal (seq 1); both session_info + // rows persist at seq 0. + assert_eq!(s.last_seq, 1); + assert_eq!(store::count_messages(c, "@peer", "w2")?, 3); + Ok(()) + }) + .unwrap(); + } + #[test] fn v1_and_v2_envelopes_coexist_in_one_session_model() { // During rollout a wrapped harness may emit v1 then v2 under the SAME diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index 741b2d46f2..b9ed3e8287 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -18,6 +18,9 @@ const SCHEMA_DDL: &str = " -- `status_state`/`current_detail`/`active_call_id` carry the v2 harness -- run-state (`status.state`/`status.detail`/`status.active_call_id`). Nullable -- and additive: a v1/legacy store gets them via `migrate` (existing rows NULL). + -- `title`/`model`/`handle`/`repo`/`branch`/`capabilities` carry the v2 + -- `session_info` enrichment (`capabilities` is a JSON array of kind strings). + -- Also nullable/additive — `migrate` backfills them on an older store. CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT NOT NULL, agent_id TEXT NOT NULL, @@ -30,6 +33,12 @@ const SCHEMA_DDL: &str = " status_state TEXT, current_detail TEXT, active_call_id TEXT, + title TEXT, + model TEXT, + handle TEXT, + repo TEXT, + branch TEXT, + capabilities TEXT, PRIMARY KEY (agent_id, session_id) ); @@ -228,6 +237,17 @@ 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 `session_info` enrichment columns (spec §4). Same per-column, + // freshness-independent guard as the run-state block above: a fresh DB has + // them from SCHEMA_DDL (no-op); a pre-session_info store gains them here with + // existing rows defaulting NULL. `capabilities` holds a JSON array of kinds. + add_column_if_missing(conn, "sessions", "title", "TEXT")?; + add_column_if_missing(conn, "sessions", "model", "TEXT")?; + add_column_if_missing(conn, "sessions", "handle", "TEXT")?; + add_column_if_missing(conn, "sessions", "repo", "TEXT")?; + add_column_if_missing(conn, "sessions", "branch", "TEXT")?; + add_column_if_missing(conn, "sessions", "capabilities", "TEXT")?; Ok(()) } @@ -243,13 +263,20 @@ pub fn message_exists(conn: &Connection, id: &str) -> Result { .is_some()) } -/// Insert or update the session row (keyed by agent + session). +/// Insert or update the session row (keyed by agent + session). The +/// `session_info` enrichment columns COALESCE like the run-state ones, so an +/// ordinary event (which carries none) never wipes a prior intro's metadata, and +/// a later `session_info` (`resumed=true`) refreshes rather than duplicates. +/// `capabilities` is stored as a JSON array; an empty list encodes to NULL so it +/// COALESCEs to "no change" instead of clobbering a prior non-empty list. pub fn upsert_session(conn: &Connection, s: &OrchestrationSession) -> Result<()> { + let capabilities = encode_capabilities(&s.capabilities); conn.execute( "INSERT INTO sessions (session_id, agent_id, source, label, workspace, last_seq, created_at, last_message_at, - status_state, current_detail, active_call_id) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) + status_state, current_detail, active_call_id, + title, model, handle, repo, branch, capabilities) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17) ON CONFLICT(agent_id, session_id) DO UPDATE SET last_seq = MAX(sessions.last_seq, excluded.last_seq), last_message_at = excluded.last_message_at, @@ -259,7 +286,15 @@ pub fn upsert_session(conn: &Connection, s: &OrchestrationSession) -> Result<()> -- never wipes the last status; a fresh `status` event overwrites them. status_state = COALESCE(excluded.status_state, sessions.status_state), current_detail = COALESCE(excluded.current_detail, sessions.current_detail), - active_call_id = COALESCE(excluded.active_call_id, sessions.active_call_id)", + active_call_id = COALESCE(excluded.active_call_id, sessions.active_call_id), + -- session_info enrichment: COALESCE so non-session_info events preserve + -- the last intro's metadata, and a `resumed=true` re-intro refreshes it. + title = COALESCE(excluded.title, sessions.title), + model = COALESCE(excluded.model, sessions.model), + handle = COALESCE(excluded.handle, sessions.handle), + repo = COALESCE(excluded.repo, sessions.repo), + branch = COALESCE(excluded.branch, sessions.branch), + capabilities = COALESCE(excluded.capabilities, sessions.capabilities)", params![ s.session_id, s.agent_id, @@ -272,11 +307,37 @@ pub fn upsert_session(conn: &Connection, s: &OrchestrationSession) -> Result<()> s.status_state, s.current_detail, s.active_call_id, + s.title, + s.model, + s.handle, + s.repo, + s.branch, + capabilities, ], )?; Ok(()) } +/// Encode `session_info.capabilities` for the `sessions.capabilities` TEXT column: +/// a JSON array, or `None` for an empty list so the COALESCE upsert treats it as +/// "no update" (a content/status event carries no capabilities and must not wipe +/// a prior intro's list). +fn encode_capabilities(capabilities: &[String]) -> Option { + if capabilities.is_empty() { + return None; + } + // A `Vec` always serialises, but fall back to NULL rather than + // failing the whole upsert on the impossible error path. + serde_json::to_string(capabilities).ok() +} + +/// Decode the `sessions.capabilities` JSON array back into a `Vec`. A +/// NULL/absent or malformed value reads as an empty list (never an error). +fn decode_capabilities(raw: Option) -> Vec { + raw.and_then(|s| serde_json::from_str(&s).ok()) + .unwrap_or_default() +} + /// Overwrite a session's v2 run-state columns from an authoritative `status` /// snapshot. `upsert_session` COALESCEs these so a content event (which carries /// no run-state) never wipes the last status; a `status` event, by contrast, OWNS @@ -386,7 +447,7 @@ pub fn latest_message_preview( "SELECT body FROM messages WHERE agent_id = ?1 AND session_id = ?2 AND (event_kind IS NULL - OR event_kind NOT IN ('status', 'lifecycle', 'unknown')) + OR event_kind NOT IN ('status', 'lifecycle', 'unknown', 'session_info')) ORDER BY timestamp DESC, seq DESC LIMIT 1", params![agent_id, session_id], |row| row.get(0), @@ -398,7 +459,8 @@ pub fn latest_message_preview( pub fn list_sessions(conn: &Connection) -> Result> { let mut stmt = conn.prepare( "SELECT session_id, agent_id, source, label, workspace, last_seq, created_at, last_message_at, - status_state, current_detail, active_call_id + status_state, current_detail, active_call_id, + title, model, handle, repo, branch, capabilities FROM sessions ORDER BY last_message_at DESC", )?; let rows = stmt @@ -423,7 +485,7 @@ pub fn list_messages_by_session( event_kind, tool_name, call_id FROM messages WHERE session_id = ?1 AND timestamp < ?2 AND (event_kind IS NULL - OR event_kind NOT IN ('status', 'lifecycle', 'unknown')) + OR event_kind NOT IN ('status', 'lifecycle', 'unknown', 'session_info')) ORDER BY timestamp DESC, seq DESC LIMIT ?3", )?; let rows = stmt @@ -437,7 +499,7 @@ pub fn list_messages_by_session( event_kind, tool_name, call_id FROM messages WHERE session_id = ?1 AND (event_kind IS NULL - OR event_kind NOT IN ('status', 'lifecycle', 'unknown')) + OR event_kind NOT IN ('status', 'lifecycle', 'unknown', 'session_info')) ORDER BY timestamp DESC, seq DESC LIMIT ?2", )?; let rows = stmt @@ -484,6 +546,12 @@ fn map_session_row(row: &rusqlite::Row<'_>) -> rusqlite::Result Result { Ok(conn.query_row( "SELECT COUNT(*) FROM messages WHERE session_id = ?1 AND timestamp > ?2 AND (event_kind IS NULL - OR event_kind NOT IN ('status', 'lifecycle', 'unknown'))", + OR event_kind NOT IN ('status', 'lifecycle', 'unknown', 'session_info'))", params![session_id, cursor], |row| row.get(0), )?) @@ -598,7 +666,8 @@ pub fn load_session( ) -> Result> { conn.query_row( "SELECT session_id, agent_id, source, label, workspace, last_seq, created_at, last_message_at, - status_state, current_detail, active_call_id + status_state, current_detail, active_call_id, + title, model, handle, repo, branch, capabilities FROM sessions WHERE agent_id = ?1 AND session_id = ?2", params![agent_id, session_id], map_session_row, @@ -1016,6 +1085,57 @@ mod tests { .unwrap(); } + #[test] + fn capabilities_codec_round_trips_and_is_null_safe() { + // Non-empty → JSON array; decodes back identically. + let caps = vec!["agent_message".to_string(), "tool_call".to_string()]; + let encoded = encode_capabilities(&caps).expect("non-empty encodes"); + assert_eq!(encoded, r#"["agent_message","tool_call"]"#); + assert_eq!(decode_capabilities(Some(encoded)), caps); + + // Empty → NULL (so the COALESCE upsert treats it as "no update"). + assert_eq!(encode_capabilities(&[]), None); + + // NULL / malformed decode to an empty list, never an error. + assert!(decode_capabilities(None).is_empty()); + assert!(decode_capabilities(Some("not json".into())).is_empty()); + } + + #[test] + fn session_info_enrichment_persists_and_coalesces_across_upserts() { + let tmp = tempfile::tempdir().unwrap(); + with_connection(tmp.path(), |conn| { + // First upsert carries the full intro. + let intro = OrchestrationSession { + title: Some("Intro".into()), + model: Some("opus".into()), + handle: Some("@alice".into()), + repo: Some("org/myrepo".into()), + branch: Some("feat/x".into()), + capabilities: vec!["agent_message".into()], + ..session("@a", "h1", 0) + }; + upsert_session(conn, &intro)?; + let loaded = load_session(conn, "@a", "h1")?.expect("session"); + assert_eq!(loaded.title.as_deref(), Some("Intro")); + assert_eq!(loaded.capabilities, vec!["agent_message".to_string()]); + + // A subsequent upsert with NO enrichment (e.g. a content event) must + // COALESCE — the intro metadata survives. + upsert_session(conn, &session("@a", "h1", 1))?; + let after = load_session(conn, "@a", "h1")?.expect("session"); + assert_eq!( + after.title.as_deref(), + Some("Intro"), + "title survives a bare upsert" + ); + assert_eq!(after.model.as_deref(), Some("opus")); + assert_eq!(after.capabilities, vec!["agent_message".to_string()]); + Ok(()) + }) + .unwrap(); + } + #[test] fn status_lifecycle_unknown_rows_are_hidden_from_thread_and_unread() { let tmp = tempfile::tempdir().unwrap(); @@ -1023,8 +1143,8 @@ mod tests { upsert_session(conn, &session("@a", "h1", 1))?; // A v1 row (no event_kind) and typed content rows stay visible; - // status/lifecycle/unknown are persisted (for relay dedup) but must - // not surface in the thread or the unread count. + // status/lifecycle/unknown/session_info are persisted (for relay dedup) + // but must not surface in the thread or the unread count. let mut plain = msg("v1", "@a", "h1", 1); plain.timestamp = "2026-07-02T00:00:01Z".into(); insert_message(conn, &plain)?; @@ -1038,6 +1158,7 @@ mod tests { ("st", "status", 3), ("lc", "lifecycle", 4), ("uk", "unknown", 5), + ("si", "session_info", 6), ] { let mut hidden = msg(id, "@a", "h1", seq); hidden.event_kind = Some(kind.into()); @@ -1049,7 +1170,7 @@ mod tests { let ids: Vec<&str> = thread.iter().map(|m| m.id.as_str()).collect(); assert_eq!(ids, vec!["v1", "call"], "only v1 + typed content rows"); - // Unread (cursor at 0) counts the two visible rows, not the 3 hidden. + // Unread (cursor at 0) counts the two visible rows, not the 4 hidden. assert_eq!(unread_count(conn, "h1")?, 2); // Roster preview skips the hidden rows → newest visible is the call. diff --git a/src/openhuman/orchestration/types.rs b/src/openhuman/orchestration/types.rs index bfd49d99d6..64c7a60911 100644 --- a/src/openhuman/orchestration/types.rs +++ b/src/openhuman/orchestration/types.rs @@ -175,6 +175,10 @@ pub const SESSION_ENVELOPE_VERSION_V2: &str = "tinyplace.harness.session.v2"; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "kind", content = "payload", rename_all = "snake_case")] pub enum HarnessEventKind { + /// `session_info` — the session intro/announce, emitted once a wrapped agent + /// session is initialised (spec §2). Enrichment, not a prerequisite: a session + /// is lazy-created on the first event of any kind regardless. + SessionInfo(SessionInfoPayload), UserPrompt(UserPromptPayload), AgentMessage(TextPayload), AgentThinking(TextPayload), @@ -189,6 +193,46 @@ pub enum HarnessEventKind { Unknown(UnknownPayload), } +/// `session_info` payload — the session intro/announce (spec §2b). Thin: the +/// provider / cwd / session-ids ride on the envelope frame, so this carries only +/// what the frame lacks (identity confirmation, capabilities, UI metadata). Field +/// names are byte-identical to the TS `SessionInfoPayload` emitter (spec §2a). +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct SessionInfoPayload { + /// base58 wallet identity — confirms `envelope.from`. + #[serde(default)] + pub agent_address: String, + /// `@handle`, if registered. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub handle: Option, + /// Human-friendly session title for the UI header. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub title: Option, + /// git remote/slug, if in a repo. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub repo: Option, + /// git branch. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub branch: Option, + /// Active model (may also ride on `event.model`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub model: Option, + /// plugin/CLI version — compat gating. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub harness_version: Option, + /// Advertised event kinds (subset of the wire `kind` strings) this session + /// will emit — feature-gates the UI. + #[serde(default)] + pub capabilities: Vec, + /// `false` = fresh spawn, `true` = reconnect/resume (drives idempotent upsert + /// rather than a duplicate record). + #[serde(default)] + pub resumed: bool, + /// ISO-8601 session start. + #[serde(default)] + pub started_at: String, +} + /// `user_prompt` payload. #[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] pub struct UserPromptPayload { @@ -433,6 +477,28 @@ pub struct OrchestrationSession { /// v2 `status.active_call_id` — the in-flight tool call while running a tool. #[serde(default, skip_serializing_if = "Option::is_none")] pub active_call_id: Option, + // ── session_info enrichment (spec §4) ──────────────────────────────────── + // Populated from a `session_info` event; `None`/empty until one arrives (a + // session is lazy-created on the first event of any kind regardless). + /// `session_info.title` — human-friendly session title for the UI header. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub title: Option, + /// `session_info.model` — the active model advertised at session start. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub model: Option, + /// `session_info.handle` — the agent's `@handle`, if registered. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub handle: Option, + /// `session_info.repo` — git remote/slug the session runs in. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub repo: Option, + /// `session_info.branch` — git branch. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub branch: Option, + /// `session_info.capabilities` — advertised event kinds, for feature-gating + /// the UI. Empty for v1/legacy sessions and until a `session_info` arrives. + #[serde(default)] + pub capabilities: Vec, } /// A single persisted message. `body` is DECRYPTED plaintext and therefore @@ -663,6 +729,82 @@ mod tests { } } + #[test] + fn v2_decodes_session_info_and_round_trips_wire_field_names() { + let wire = v2_wire( + "session_info", + r#"{ "agent_address": "ELQUJvq27tYx", "handle": "@alice", + "title": "myrepo · feat/x", "repo": "org/myrepo", "branch": "feat/x", + "model": "claude-opus-4-8", "harness_version": "1.4.2", + "capabilities": ["agent_message", "tool_call"], + "resumed": false, "started_at": "2026-07-08T00:00:00Z" }"#, + ); + let env = SessionEnvelopeV2::parse(&wire).expect("valid v2"); + let decoded = env.event.decoded(); + let expected = SessionInfoPayload { + agent_address: "ELQUJvq27tYx".into(), + handle: Some("@alice".into()), + title: Some("myrepo · feat/x".into()), + repo: Some("org/myrepo".into()), + branch: Some("feat/x".into()), + model: Some("claude-opus-4-8".into()), + harness_version: Some("1.4.2".into()), + capabilities: vec!["agent_message".into(), "tool_call".into()], + resumed: false, + started_at: "2026-07-08T00:00:00Z".into(), + }; + assert_eq!(decoded, HarnessEventKind::SessionInfo(expected.clone())); + + // Re-encode the adjacently-tagged kind and decode again: a full round-trip + // proves the `kind`/`payload` framing survives serialize → deserialize. + let reencoded = serde_json::to_string(&HarnessEventKind::SessionInfo(expected.clone())) + .expect("encode kind"); + let back: HarnessEventKind = serde_json::from_str(&reencoded).expect("decode kind"); + assert_eq!(back, HarnessEventKind::SessionInfo(expected.clone())); + + // The payload's wire field names must be byte-identical to the TS emitter + // (spec §2a): snake_case keys, optionals present only when set. + let payload_json = serde_json::to_value(&expected).unwrap(); + for key in [ + "agent_address", + "handle", + "title", + "repo", + "branch", + "model", + "harness_version", + "capabilities", + "resumed", + "started_at", + ] { + assert!(payload_json.get(key).is_some(), "missing wire key `{key}`"); + } + + // Optionals are omitted when absent (`skip_serializing_if`), and defaults + // fill missing wire fields so a thin `session_info` never fails to decode. + let minimal = + SessionEnvelopeV2::parse(&v2_wire("session_info", r#"{ "agent_address": "X" }"#)) + .expect("valid v2"); + match minimal.event.decoded() { + HarnessEventKind::SessionInfo(p) => { + assert_eq!(p.agent_address, "X"); + assert_eq!(p.handle, None); + assert!(p.capabilities.is_empty()); + assert!(!p.resumed); + } + other => panic!("expected session_info, got {other:?}"), + } + let minimal_json = serde_json::to_value(SessionInfoPayload { + agent_address: "X".into(), + ..Default::default() + }) + .unwrap(); + assert!( + minimal_json.get("handle").is_none(), + "unset optionals must be skipped on the wire" + ); + } + #[test] fn v2_unrecognised_kind_folds_to_unknown_not_a_parse_error() { // A future kind the receiver doesn't model must not fail the envelope