feat(orchestration): receive session_info harness events into the v2 mirror (B′2) (#4710)

src/openhuman/orchestration/ingest.rs

@@ -121,6 +121,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<String>,
+    /// `session_info.model` (or `event.model` fallback) → the session row's `model`.
+    model: Option<String>,
+    /// `session_info.handle` → the session row's `handle`.
+    handle: Option<String>,
+    /// `session_info.repo` → the session row's `repo`.
+    repo: Option<String>,
+    /// `session_info.branch` → the session row's `branch`.
+    branch: Option<String>,
+    /// `session_info.capabilities` → the session row's `capabilities`.
+    capabilities: Vec<String>,
 }
 
 /// True for streams that carry ciphertext DM envelopes worth ingesting.
@@ -179,6 +192,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(),
     }
 }
 
@@ -216,6 +236,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(),
     }
 }
 
@@ -237,10 +264,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";
@@ -324,6 +368,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,
     }
 }
 
@@ -343,6 +393,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<String>,
+    model: Option<String>,
+    handle: Option<String>,
+    repo: Option<String>,
+    branch: Option<String>,
+    capabilities: Vec<String>,
 }
 
 impl Default for V2Body {
@@ -358,6 +415,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(),
         }
     }
 }
@@ -419,6 +482,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
@@ -879,6 +948,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

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<bool> {
         .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<String> {
+    if capabilities.is_empty() {
+        return None;
+    }
+    // A `Vec<String>` 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<String>`. A
+/// NULL/absent or malformed value reads as an empty list (never an error).
+fn decode_capabilities(raw: Option<String>) -> Vec<String> {
+    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<Vec<OrchestrationSession>> {
     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<OrchestrationSes
         status_state: row.get(8)?,
         current_detail: row.get(9)?,
         active_call_id: row.get(10)?,
+        title: row.get(11)?,
+        model: row.get(12)?,
+        handle: row.get(13)?,
+        repo: row.get(14)?,
+        branch: row.get(15)?,
+        capabilities: decode_capabilities(row.get(16)?),
     })
 }
 
@@ -493,7 +561,7 @@ pub fn unread_count(conn: &Connection, session_id: &str) -> Result<i64> {
     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<Option<OrchestrationSession>> {
     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.

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<String>,
+    /// Human-friendly session title for the UI header.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub title: Option<String>,
+    /// git remote/slug, if in a repo.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub repo: Option<String>,
+    /// git branch.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub branch: Option<String>,
+    /// Active model (may also ride on `event.model`).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub model: Option<String>,
+    /// plugin/CLI version — compat gating.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub harness_version: Option<String>,
+    /// Advertised event kinds (subset of the wire `kind` strings) this session
+    /// will emit — feature-gates the UI.
+    #[serde(default)]
+    pub capabilities: Vec<String>,
+    /// `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<String>,
+    // ── 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<String>,
+    /// `session_info.model` — the active model advertised at session start.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub model: Option<String>,
+    /// `session_info.handle` — the agent's `@handle`, if registered.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub handle: Option<String>,
+    /// `session_info.repo` — git remote/slug the session runs in.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub repo: Option<String>,
+    /// `session_info.branch` — git branch.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub branch: Option<String>,
+    /// `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<String>,
 }
 
 /// 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