mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-20 15:33:54 +00:00
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:
parent
156ea4ad89
commit
9ce0e67f27
7 changed files with 339 additions and 16 deletions
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
62
run_agent.py
62
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=<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:
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
57
tests/hermes_state/test_conversation_root.py
Normal file
57
tests/hermes_state/test_conversation_root.py
Normal 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("") == ""
|
||||
Loading…
Add table
Add a link
Reference in a new issue