From 9ce0e67f27eb9574b59746346e24c53bd63d180a Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 16 Jul 2026 00:49:04 -0700 Subject: [PATCH] feat(portal): ambient conversation context entangles aux/MoA/delegate calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends the conversation= 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 | 8 +- agent/portal_tags.py | 68 +++++++++- agent/title_generator.py | 15 +++ hermes_state.py | 14 ++ run_agent.py | 62 +++++++-- tests/agent/test_portal_tags.py | 131 +++++++++++++++++++ tests/hermes_state/test_conversation_root.py | 57 ++++++++ 7 files changed, 339 insertions(+), 16 deletions(-) create mode 100644 tests/hermes_state/test_conversation_root.py diff --git a/agent/moa_loop.py b/agent/moa_loop.py index 43691fd7f997..d6a0bebf101d 100644 --- a/agent/moa_loop.py +++ b/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, diff --git a/agent/portal_tags.py b/agent/portal_tags.py index 50a42ee2778c..ecfecdc6c57a 100644 --- a/agent/portal_tags.py +++ b/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=`` 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 - always-on base tags) omit it and get the canonical two-tag set. + conversation. When it is omitted, the ambient conversation context + (``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: - tags.append(conversation_tag(session_id)) + # Ambient context first: the agent loop publishes the lineage ROOT 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 diff --git a/agent/title_generator.py b/agent/title_generator.py index 5534b34710d5..840b56aaf09b 100644 --- a/agent/title_generator.py +++ b/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 ) diff --git a/hermes_state.py b/hermes_state.py index 34923ebc4f23..e5a517c50534 100644 --- a/hermes_state.py +++ b/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] diff --git a/run_agent.py b/run_agent.py index bcacec3909b6..7cb3839d504e 100644 --- a/run_agent.py +++ b/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( - self, - user_message, - system_message, - conversation_history, - task_id, - stream_callback, - persist_user_message, - persist_user_timestamp=persist_user_timestamp, - moa_config=moa_config, + from agent.portal_tags import ( + reset_conversation_context, + set_conversation_context, ) + # 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=`` 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: """ diff --git a/tests/agent/test_portal_tags.py b/tests/agent/test_portal_tags.py index 69b96c443037..418d3b59f147 100644 --- a/tests/agent/test_portal_tags.py +++ b/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 diff --git a/tests/hermes_state/test_conversation_root.py b/tests/hermes_state/test_conversation_root.py new file mode 100644 index 000000000000..58132072c67d --- /dev/null +++ b/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("") == ""