perf(orchestration): batch the per-session unread_count into one query

[mysma-9403]

Summary

• Collapse an N+1 in the orchestration session roster and attention queue: unread_count ran 2 queries per session and was called once per session in two hot RPC paths.
• Add a batched unread_counts(conn) -> HashMap<session_id, i64> (one GROUP BY with a LEFT JOIN on the read-cursor kv) and consume it in sessions_list and gather_unread_signals.
• Byte-identical results; add a parity test asserting the batched map equals the scalar unread_count across every cursor/visibility state.

Problem

store::unread_count(conn, session_id) does two round-trips — kv_get(read_cursor_key(..)) then a COUNT(*) — and is called once per session in:

• handle_sessions_list (schemas.rs) — the session roster, for session in rows { unread_count(..) }, and list_sessions has no LIMIT, so listing N sessions is 2N queries.
• gather_unread_signals (ops.rs) — the attention queue, same per-session loop.

This is especially visible in handle_sessions_list because the sibling message count in the very same loop was already batched into visible_message_counts / visible_message_counts_by_session (consumed via .get().copied().unwrap_or(0)); the unread count was simply left as the lone per-row DB hit.

Solution

Add a batched query mirroring visible_message_counts_by_session:

SELECT m.session_id, COUNT(*)
  FROM messages m
  LEFT JOIN kv ON kv.k = 'read:' || m.session_id
 WHERE m.timestamp > COALESCE(kv.v, '')
   AND (m.event_kind IS NULL OR m.event_kind NOT IN ('status','lifecycle','unknown','session_info'))
 GROUP BY m.session_id

• kv.k is PRIMARY KEY, so the LEFT JOIN matches at most one cursor row per session — no fan-out.
• COALESCE(kv.v, '') reproduces the scalar path's .unwrap_or_default() empty cursor (a never-read session counts all its visible messages).
• The event_kind predicate is copied verbatim from unread_count.
• Sessions with zero unread drop out of the GROUP BY; both call sites already default missing entries to 0, so behaviour is identical.

Each call site now computes the map once before its loop and reads map.get(&session_id).copied().unwrap_or(0). The scalar unread_count is retained (still used by existing store tests and as the parity oracle).

Submission Checklist

• Tests added or updated (happy path + at least one failure / edge case) — unread_counts_batches_and_matches_scalar_unread_count asserts the batched map equals the scalar unread_count for every session across five states: no cursor, mid-stream cursor (partial), fully read (zero), a hidden excluded-event_kind row, and a session with no messages.
• Diff coverage ≥ 80% — the new unread_counts query and both swapped call sites are exercised by the parity test plus existing sessions_list / attention coverage. cargo test -p openhuman --lib orchestration::store passes.
• Coverage matrix updated — N/A: internal query batching, no user-visible feature change.
• All affected feature IDs listed under ## Related — N/A.
• No new external network dependencies introduced.
• Manual smoke checklist updated — N/A: does not touch a release-cut surface.
• Linked issue closed via Closes #NNN — no existing issue; found by inspection.

Impact

• Desktop/CLI: openhuman.sessions_list and the attention queue now issue one unread query instead of 2N. No behaviour change (verified byte-identical by the parity test); no schema/migration.
• Extends the existing visible_message_counts batching pattern in the same loop.

Related

• Closes:
• Follow-up PR(s)/TODOs:

---

AI Authored PR Metadata

Linear Issue

• Key: N/A
• URL: N/A

Commit & Branch

• Branch: perf/orchestration-batch-unread-counts
• Commit SHA: 45c2fae86

Validation Run

• pnpm --filter openhuman-app format:check — N/A (no frontend change)
• pnpm typecheck — N/A (no frontend change)
• Focused tests: cargo test -p openhuman --lib orchestration::store
• Rust fmt/check (if changed): cargo fmt
• Tauri fmt/check (if changed): N/A (core-only change)

Behavior Changes

• Intended behavior change: none — pure query batching.
• User-visible effect: none (fewer DB round-trips on session listing / attention).

Parity Contract

• Legacy behavior preserved: unread_counts returns exactly what unread_count would per session; COALESCE(kv.v,'') = .unwrap_or_default(), event_kind predicate copied verbatim, zero-unread sessions default to 0 at both call sites.
• Guard/fallback/dispatch parity checks: parity test asserts map == scalar across all cursor/visibility states.

[coderabbitai]

📝 Walkthrough

Walkthrough

Unread message counts are now fetched in bulk through a new store helper. Signal gathering and session listing use map lookups with zero defaults, while tests cover visibility filtering, cursors, and empty results.

Changes

Unread count batching

Layer / File(s)	Summary
Batched unread count query src/openhuman/orchestration/store.rs	Adds unread_counts with cursor-aware visibility filtering and tests its results against scalar counts across multiple session states.
Orchestration integration src/openhuman/orchestration/ops.rs, src/openhuman/orchestration/schemas.rs	Updates unread signal gathering and session listing to fetch counts once and use per-session map lookups.

Estimated code review effort: 2 (Simple) | ~10 minutes

Suggested labels: rust-core

Suggested reviewers: sanil-23

Poem

A rabbit counts messages in one swift sweep,
No session-by-session hops to keep.
Cursors guide what remains unread,
Hidden rows stay safely tucked in bed.
Bulk counts bloom—then off I leap!

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly summarizes the main change: batching per-session unread_count queries into one query for orchestration.

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[senamakel]

Automated technical review: not approved.

Unmet gate -- CI check failure:

One check fails: Rust Feature-Gate Smoke (gates off) fails due to a pre-existing allowlist drift -- a new gated test file (src/openhuman/composio/action_tool.rs) was added to the tree but the CI lane's EXPECTED allowlist was not updated to include it. This is unrelated to this PR (which touches only src/openhuman/orchestration/ files), but per the approval standard all reported checks must be passing.

N+1 claim: Verified correct. The new store::unread_counts function issues a single GROUP BY query with LEFT JOIN kv ON kv.k = 'read:' || m.session_id to batch the cursor lookup and count, replacing 2N queries (cursor lookup + COUNT per session) with 1. The COALESCE(kv.v, '') reproduces the scalar path's .unwrap_or_default(), and the event_kind predicate is copied verbatim. Both call sites (handle_sessions_list and gather_unread_signals) pre-compute the map and use .get().copied().unwrap_or(0). The parity test unread_counts_batches_and_matches_scalar_unread_count asserts byte-identical results across all cursor/visibility states (no cursor, partial, fully read, hidden event_kind, empty session). The scalar unread_count is retained as the parity oracle.

Next step: The Feature-Gate Smoke allowlist needs updating to include openhuman/composio/action_tool.rs (a change to .github/workflows/ci-lite.yml or the relevant script). Once CI is all green, this PR is ready for approval.

[greptile-apps]

Greptile Summary

This PR collapses an N+1 query pattern in two hot RPC paths (handle_sessions_list and gather_unread_signals) by replacing per-session unread_count calls (2 queries each) with a single batched unread_counts function that groups by session_id in one LEFT JOIN query.

• store.rs: Adds unread_counts(conn) -> HashMap<String, i64>, a single SELECT … GROUP BY that LEFT JOINs the kv read-cursor table to reproduce the per-session scalar unread_count semantics, plus a parity test across five cursor/visibility states.
• schemas.rs / ops.rs: Both loop sites pre-compute the map once before iterating sessions and resolve per-session counts via .get().copied().unwrap_or(0), identical to the pattern already used for visible_message_counts_by_session.

Confidence Score: 5/5

Safe to merge — pure query batching with no behaviour change; the SQL logic, kv-key format, COALESCE semantics, and event_kind predicate all match the scalar path exactly, and a dedicated parity test validates this across every cursor/visibility state.

The batched SQL reproduces the scalar unread_count faithfully: the LEFT JOIN kv ON kv.k = 'read:' || m.session_id key format matches read_cursor_key, COALESCE(kv.v, '') matches .unwrap_or_default(), the event_kind NOT IN predicate is verbatim, and zero-unread sessions falling out of the GROUP BY are handled by .unwrap_or(0) at both call sites. The parity test covers all five described states. Both call sites follow the existing visible_message_counts_by_session pattern. No schema changes, no user-visible behaviour change.

No files require special attention.

Important Files Changed

Filename	Overview
src/openhuman/orchestration/store.rs	Adds unread_counts batched query (correct SQL, kv-key format matches scalar, COALESCE semantics verified) and a thorough parity test covering all five cursor/visibility states.
src/openhuman/orchestration/schemas.rs	Replaces per-session unread_count call in the session-roster loop with a pre-computed map lookup; consistent with the existing visible_message_counts_by_session pattern in the same loop.
src/openhuman/orchestration/ops.rs	Replaces per-session unread_count call in gather_unread_signals with a pre-computed map lookup; minor note that master/subconscious entries are included in the map but are never consumed (skipped in the loop), which is harmless.

Sequence Diagram

sequenceDiagram
    participant Caller
    participant store

    Note over Caller,store: BEFORE (N sessions → 2N queries)
    Caller->>store: list_sessions()
    loop for each session (N times)
        store->>store: kv_get(read:session_id)
        store->>store: "COUNT(*) WHERE session_id=?"
        store-->>Caller: unread count
    end

    Note over Caller,store: AFTER (N sessions → 1 query)
    Caller->>store: unread_counts()
    store->>store: "SELECT m.session_id, COUNT(*) FROM messages LEFT JOIN kv GROUP BY session_id"
    store-->>Caller: "HashMap<session_id, i64>"
    Caller->>store: list_sessions()
    loop for each session (N times)
        Caller->>Caller: map.get(session_id).copied().unwrap_or(0)
    end

Loading

_{Reviews (1): Last reviewed commit: "perf(orchestration): batch the per-sessi..." | Re-trigger Greptile}