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.
This commit is contained in:
Teknium 2026-07-16 00:49:04 -07:00
parent 156ea4ad89
commit 9ce0e67f27
7 changed files with 339 additions and 16 deletions

View file

@ -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,

View file

@ -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
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

View file

@ -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
)

View file

@ -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]

View file

@ -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=<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:
"""

View file

@ -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

View file

@ -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("") == ""