feat(portal): ambient conversation context entangles aux/MoA/delegate calls

Extends the conversation=<id> Portal tag (salvaged from PR #65183 by
@J-SUPHA) from main-loop-only to every LLM call in a conversation:

- agent/portal_tags.py: ContextVar-based conversation context.
  nous_portal_tags() falls back to the ambient id when no explicit
  session_id is passed, so every aux tag site (auxiliary_client,
  chat_completion_helpers summary path, web_tools) inherits the tag
  with zero per-call-site plumbing. Ambient id wins over explicit
  per-segment ids since it carries the lineage root.
- hermes_state.py: SessionDB.get_conversation_root() — public wrapper
  over the lineage walk; returns the ROOT session id, so one
  user-facing conversation keeps a single conversation= value across
  context-compression rotation, and delegate subagent trees tag as
  their parent conversation.
- run_agent.py: run_conversation() publishes the root id for the turn
  and resets it in finally. _conversation_root_id() resolves via
  _parent_session_id for subagents.
- agent/moa_loop.py: MoA reference fan-out workers now run under
  propagate_context_to_thread so advisor slots attribute to the acting
  conversation (also fixes approval-callback propagation on that path).
- agent/title_generator.py: bare title thread republishes the context
  from its session id (spawned after turn reset).

Tests: ContextVar semantics, cross-context isolation, thread-hop
propagation, lineage-root resolution incl. cycle guard.

agent/moa_loop.py

@@ -404,6 +404,12 @@ def _run_references_parallel(
     results: list[tuple[str, str, Any] | None] = [None] * len(reference_models)
     futures = {}
     workers = min(_MAX_REFERENCE_WORKERS, len(reference_models))
+    # Reference slots run on bare executor threads, which start with an empty
+    # contextvars.Context — propagate the parent turn's context (approval
+    # callbacks + the Nous Portal conversation tag) into each worker so
+    # advisor calls attribute to the same conversation as the acting turn.
+    from tools.thread_context import propagate_context_to_thread
+
     with ThreadPoolExecutor(max_workers=workers) as executor:
         for idx, slot in enumerate(reference_models):
             if slot.get("provider") == "moa":
@@ -415,7 +421,7 @@ def _run_references_parallel(
                 continue
             futures[
                 executor.submit(
-                    _run_reference,
+                    propagate_context_to_thread(_run_reference),
                     slot,
                     ref_messages,
                     temperature=temperature,

agent/portal_tags.py

@@ -31,7 +31,55 @@ version can change at runtime (editable installs, hot-reload tooling), and
 
 from __future__ import annotations
 
-from typing import List
+from contextvars import ContextVar
+from typing import List, Optional
+
+# ── Ambient conversation context ─────────────────────────────────────────────
+#
+# The main agent loop knows its ``session_id``; the dozens of auxiliary call
+# sites (compression, title generation, vision, web_extract, session_search,
+# MoA reference/aggregator slots, curator, kanban helpers, ...) do not — they
+# funnel through ``agent.auxiliary_client.call_llm`` which has no session
+# handle. Rather than threading a ``session_id`` parameter through every one
+# of those call sites (and every future one), the agent loop publishes the
+# active conversation id here and ``nous_portal_tags()`` picks it up as a
+# fallback whenever no explicit ``session_id`` is passed.
+#
+# ContextVar (not a module global) so concurrent agents in one process —
+# gateway sessions, delegate_task subagents, batch runners — never see each
+# other's conversation id. Worker threads spawned via
+# ``tools.thread_context.propagate_context_to_thread`` (background review,
+# MoA fan-out, tool executor) inherit it through the copied Context; bare
+# threads (title generator) capture it explicitly at spawn time.
+_conversation_id: ContextVar[Optional[str]] = ContextVar(
+    "nous_portal_conversation_id", default=None
+)
+
+
+def set_conversation_context(conversation_id: Optional[str]):
+    """Publish the active conversation id for ambient Portal tagging.
+
+    Called by the agent loop at turn entry with the conversation's stable
+    id (the session-lineage ROOT id, so the tag survives context-compression
+    session rotation). Pass ``None`` to clear. Returns the ContextVar token
+    so callers can ``reset_conversation_context(token)`` on turn exit.
+    """
+    return _conversation_id.set(conversation_id or None)
+
+
+def reset_conversation_context(token) -> None:
+    """Restore the previous conversation context (pair with ``set_...``)."""
+    try:
+        _conversation_id.reset(token)
+    except Exception:
+        # Token from another Context (e.g. reset on a different thread) —
+        # fall back to clearing rather than raising in cleanup paths.
+        _conversation_id.set(None)
+
+
+def get_conversation_context() -> Optional[str]:
+    """Return the ambient conversation id, or ``None`` when unset."""
+    return _conversation_id.get()
 
 
 def _hermes_version() -> str:
@@ -77,10 +125,20 @@ def nous_portal_tags(session_id: str | None = None) -> List[str]:
 
     When ``session_id`` is provided, a ``conversation=<session_id>`` tag is
     appended so Portal usage can be attributed to a specific Hermes
-    conversation. Callers without a session id (e.g. the auxiliary client's
+    conversation. When it is omitted, the ambient conversation context
-    always-on base tags) omit it and get the canonical two-tag set.
+    (``set_conversation_context``, published by the agent loop at turn
+    entry) is used instead — this is how auxiliary calls (compression,
+    titles, vision, MoA slots, ...) inherit the conversation tag without
+    per-call-site plumbing. Callers outside any conversation (e.g. the
+    auxiliary client's import-time base tags) get the canonical two-tag set.
     """
     tags = ["product=hermes-agent", hermes_client_tag()]
-    if session_id:
+    # Ambient context first: the agent loop publishes the lineage ROOT id
-        tags.append(conversation_tag(session_id))
+    # (stable across context-compression rotation and delegate subagent
+    # trees), which is the better conversation key than a per-segment
+    # session_id passed explicitly. The explicit argument remains as a
+    # fallback for callers running outside any agent turn.
+    effective = get_conversation_context() or session_id
+    if effective:
+        tags.append(conversation_tag(effective))
     return tags

agent/title_generator.py

@@ -145,6 +145,21 @@ def auto_title_session(
     except Exception:
         return
 
+    # This runs on a bare daemon thread spawned AFTER the turn's ambient
+    # conversation context was reset, so publish it here from the session id
+    # we already hold — the title-generation LLM call then carries the same
+    # ``conversation=`` Portal tag as the turn it titles. Root-of-lineage for
+    # consistency with the agent loop (a no-op on first exchange, where
+    # titling happens, but correct if this ever runs on a continuation).
+    from agent.portal_tags import set_conversation_context
+
+    conversation_id = session_id
+    try:
+        conversation_id = session_db.get_conversation_root(session_id) or session_id
+    except Exception:
+        pass
+    set_conversation_context(conversation_id)
+
     title = generate_title(
         user_message, assistant_response, failure_callback=failure_callback, main_runtime=main_runtime
     )

hermes_state.py

@@ -4544,6 +4544,20 @@ class SessionDB:
         messages = _strip_background_review_harness(messages)
         return messages
 
+    def get_conversation_root(self, session_id: str) -> str:
+        """Return the ROOT id of *session_id*'s lineage chain.
+
+        The root is the stable "conversation id": context compression
+        rotates ``session_id`` to a new segment linked via
+        ``parent_session_id``, and delegate subagents hang off their
+        parent the same way. Walking to the root gives every segment of
+        one user-facing conversation (and its delegation tree) a single
+        identifier — used for Nous Portal ``conversation=`` usage tagging.
+        Returns *session_id* unchanged when it has no recorded parent.
+        """
+        chain = self._session_lineage_root_to_tip(session_id)
+        return (chain[0] if chain and chain[0] else session_id)
+
     def _session_lineage_root_to_tip(self, session_id: str) -> List[str]:
         if not session_id:
             return [session_id]

run_agent.py

@@ -5876,6 +5876,35 @@ class AIAgent:
         from agent.chat_completion_helpers import handle_max_iterations
         return handle_max_iterations(self, messages, api_call_count)
 
+    def _conversation_root_id(self) -> Optional[str]:
+        """Resolve the stable conversation id for Portal usage attribution.
+
+        Returns the session-lineage ROOT id rather than the current segment
+        id, so one user-facing conversation keeps a single ``conversation=``
+        tag across context-compression rotation (`/new` starts a genuinely
+        new lineage). Delegate subagents resolve through their
+        ``_parent_session_id`` so an entire delegation tree tags as the
+        parent conversation.
+
+        Best-effort: falls back to the raw session id when the session DB
+        is unavailable or the lineage walk fails.
+        """
+        sid = getattr(self, "session_id", None)
+        if not sid:
+            return None
+        # Subagents may not have a DB row yet on their first turn; walking
+        # from the parent id still lands on the right root.
+        start = getattr(self, "_parent_session_id", None) or sid
+        db = getattr(self, "_session_db", None)
+        if db is not None:
+            try:
+                root = db.get_conversation_root(start)
+                if root:
+                    return root
+            except Exception:
+                logger.debug("Conversation root lineage walk failed", exc_info=True)
+        return start
+
     def run_conversation(
         self,
         user_message: Any,
@@ -5889,17 +5918,30 @@ class AIAgent:
     ) -> Dict[str, Any]:
         """Forwarder — see ``agent.conversation_loop.run_conversation``."""
         from agent.conversation_loop import run_conversation
-        return run_conversation(
+        from agent.portal_tags import (
-            self,
+            reset_conversation_context,
-            user_message,
+            set_conversation_context,
-            system_message,
-            conversation_history,
-            task_id,
-            stream_callback,
-            persist_user_message,
-            persist_user_timestamp=persist_user_timestamp,
-            moa_config=moa_config,
         )
+        # Publish the conversation id for ambient Nous Portal tagging. Every
+        # LLM call made inside this turn — main loop, compression, vision,
+        # web_extract, session_search, MoA slots, background-review forks
+        # (which copy this Context into their thread) — inherits the
+        # ``conversation=<root>`` tag with zero per-call-site plumbing.
+        token = set_conversation_context(self._conversation_root_id())
+        try:
+            return run_conversation(
+                self,
+                user_message,
+                system_message,
+                conversation_history,
+                task_id,
+                stream_callback,
+                persist_user_message,
+                persist_user_timestamp=persist_user_timestamp,
+                moa_config=moa_config,
+            )
+        finally:
+            reset_conversation_context(token)
 
     def chat(self, message: str, stream_callback: Optional[callable] = None) -> str:
         """

tests/agent/test_portal_tags.py

@@ -69,6 +69,137 @@ def test_nous_portal_tags_omits_conversation_without_session_id():
         assert not any(t.startswith("conversation=") for t in tags)
 
 
+# ── Ambient conversation context (ContextVar) ────────────────────────────────
+
+
+def test_ambient_context_tags_calls_without_explicit_session_id():
+    """set_conversation_context makes bare nous_portal_tags() carry the tag.
+
+    This is the mechanism auxiliary calls (compression, titles, vision, MoA
+    slots) rely on — they call nous_portal_tags() with no argument.
+    """
+    from agent.portal_tags import (
+        conversation_tag,
+        nous_portal_tags,
+        reset_conversation_context,
+        set_conversation_context,
+    )
+
+    token = set_conversation_context("root-sess-1")
+    try:
+        tags = nous_portal_tags()
+        assert conversation_tag("root-sess-1") in tags
+        assert len(tags) == 3
+    finally:
+        reset_conversation_context(token)
+    # After reset the ambient tag is gone.
+    assert not any(t.startswith("conversation=") for t in nous_portal_tags())
+
+
+def test_ambient_context_wins_over_explicit_session_id():
+    """The lineage-root ambient id outranks a per-segment explicit id."""
+    from agent.portal_tags import (
+        conversation_tag,
+        nous_portal_tags,
+        reset_conversation_context,
+        set_conversation_context,
+    )
+
+    token = set_conversation_context("lineage-root")
+    try:
+        tags = nous_portal_tags(session_id="segment-2")
+        assert conversation_tag("lineage-root") in tags
+        assert conversation_tag("segment-2") not in tags
+    finally:
+        reset_conversation_context(token)
+
+
+def test_ambient_context_set_none_clears():
+    """set_conversation_context(None) publishes no tag (and coerces '')."""
+    from agent.portal_tags import (
+        get_conversation_context,
+        nous_portal_tags,
+        reset_conversation_context,
+        set_conversation_context,
+    )
+
+    for empty in (None, ""):
+        token = set_conversation_context(empty)
+        try:
+            assert get_conversation_context() is None
+            assert len(nous_portal_tags()) == 2
+        finally:
+            reset_conversation_context(token)
+
+
+def test_ambient_context_isolated_between_contexts():
+    """Two copied Contexts (≈ two concurrent agents) don't leak into each other."""
+    import contextvars
+
+    from agent.portal_tags import (
+        conversation_tag,
+        nous_portal_tags,
+        set_conversation_context,
+    )
+
+    def _in_conversation(cid):
+        set_conversation_context(cid)
+        return nous_portal_tags()
+
+    tags_a = contextvars.copy_context().run(_in_conversation, "agent-a")
+    tags_b = contextvars.copy_context().run(_in_conversation, "agent-b")
+    assert conversation_tag("agent-a") in tags_a
+    assert conversation_tag("agent-b") in tags_b
+    assert conversation_tag("agent-b") not in tags_a
+    # The outer (test) context stays clean.
+    assert not any(t.startswith("conversation=") for t in nous_portal_tags())
+
+
+def test_ambient_context_propagates_via_thread_context_helper():
+    """propagate_context_to_thread carries the tag onto executor workers (MoA path)."""
+    from concurrent.futures import ThreadPoolExecutor
+
+    from agent.portal_tags import (
+        conversation_tag,
+        nous_portal_tags,
+        reset_conversation_context,
+        set_conversation_context,
+    )
+    from tools.thread_context import propagate_context_to_thread
+
+    token = set_conversation_context("moa-root")
+    try:
+        with ThreadPoolExecutor(max_workers=1) as ex:
+            plain = ex.submit(nous_portal_tags).result()
+            propagated = ex.submit(
+                propagate_context_to_thread(nous_portal_tags)
+            ).result()
+    finally:
+        reset_conversation_context(token)
+
+    # Bare submit loses the ContextVar; the propagation wrapper keeps it.
+    assert not any(t.startswith("conversation=") for t in plain)
+    assert conversation_tag("moa-root") in propagated
+
+
+def test_reset_with_foreign_token_clears_instead_of_raising():
+    """reset_conversation_context on another Context's token must not raise."""
+    import contextvars
+
+    from agent.portal_tags import (
+        get_conversation_context,
+        reset_conversation_context,
+        set_conversation_context,
+    )
+
+    foreign_token = contextvars.copy_context().run(
+        lambda: set_conversation_context("elsewhere")
+    )
+    set_conversation_context("here")
+    reset_conversation_context(foreign_token)  # must not raise
+    assert get_conversation_context() is None
+
+
 def test_auxiliary_client_nous_extra_body_uses_helper():
     """auxiliary_client.NOUS_EXTRA_BODY must match the canonical helper output."""
     from agent.auxiliary_client import NOUS_EXTRA_BODY

tests/hermes_state/test_conversation_root.py

@@ -0,0 +1,57 @@
+"""Tests for SessionDB.get_conversation_root — stable conversation id resolution.
+
+The conversation root is the Nous Portal ``conversation=`` tag value: one
+stable id per user-facing conversation, surviving context-compression
+session rotation and covering delegate subagent trees.
+"""
+import pytest
+
+from hermes_state import SessionDB
+
+
+@pytest.fixture
+def db(tmp_path):
+    return SessionDB(tmp_path / "state.db")
+
+
+def test_root_of_standalone_session_is_itself(db):
+    db.create_session("solo", source="cli")
+    assert db.get_conversation_root("solo") == "solo"
+
+
+def test_root_of_unknown_session_is_itself(db):
+    # No DB row at all (e.g. subagent's first turn before create_session).
+    assert db.get_conversation_root("ghost") == "ghost"
+
+
+def test_root_follows_compression_rotation_chain(db):
+    # root -> seg2 -> seg3 (two compression rotations)
+    db.create_session("root", source="cli")
+    db.create_session("seg2", source="cli", parent_session_id="root")
+    db.create_session("seg3", source="cli", parent_session_id="seg2")
+    assert db.get_conversation_root("seg3") == "root"
+    assert db.get_conversation_root("seg2") == "root"
+    assert db.get_conversation_root("root") == "root"
+
+
+def test_root_covers_delegate_child_sessions(db):
+    db.create_session("parent", source="cli")
+    db.create_session("child", source="delegate", parent_session_id="parent")
+    assert db.get_conversation_root("child") == "parent"
+
+
+def test_root_handles_parent_cycle_without_hanging(db):
+    # Defensive: a corrupted parent chain with a cycle must terminate.
+    db.create_session("a", source="cli")
+    db.create_session("b", source="cli", parent_session_id="a")
+    with db._lock:
+        db._conn.execute(
+            "UPDATE sessions SET parent_session_id = ? WHERE id = ?", ("b", "a")
+        )
+        db._conn.commit()
+    root = db.get_conversation_root("b")
+    assert root in ("a", "b")  # terminated, returned a chain member
+
+
+def test_root_empty_session_id_passthrough(db):
+    assert db.get_conversation_root("") == ""