feat: session topic segmentation — lightweight sub-contexts within sessions

agent/conversation_compression.py

@@ -3549,7 +3549,7 @@ def _commit_compaction(
                 # Tail rows tagged by compress() are archived as superseded duplicates, not
                 # compacted=1. Count against the FINAL list — salvage may have dropped rows.
                 agent._session_db.archive_and_compact(
-                    agent.session_id, compressed, model_config_patch={PROACTIVE_PRUNE_REARM_MODEL_CONFIG_KEY: None},
+                    agent.session_id, compressed, topic_id=getattr(agent, "_active_topic_id", None), model_config_patch={PROACTIVE_PRUNE_REARM_MODEL_CONFIG_KEY: None},
                     watermark=lease.watermark, lock_holder=lease.holder,
                     tail_count=sum(1 for m in compressed if id(m) in _tail_tagged_ids),
                 )

agent/session_persistence.py

@@ -155,6 +155,11 @@ def _db_flush_row(agent, msg: Dict, is_current_turn_user: bool) -> Dict[str, Any
     """Build the session-db row for ``msg``, applying the persist override to THIS row only."""
     role = msg.get("role", "unknown")
     content = msg.get("content")
+    topic_id = getattr(agent, "_active_topic_id", None)
+    if topic_id is None and role == "user":
+        topic_id = agent._auto_create_first_topic(content if isinstance(content, str) else "")
+    if role == "assistant" and isinstance(content, str):
+        content = agent._process_topic_signals(content)
     # api_content sidecar: exact bytes sent to the API when they differ from clean content (replay parity).
     api_content = msg.get("api_content") if isinstance(msg.get("api_content"), str) else None
     timestamp = msg.get("timestamp")
@@ -181,6 +186,7 @@ def _db_flush_row(agent, msg: Dict, is_current_turn_user: bool) -> Dict[str, Any
         "timestamp": timestamp, "api_content": api_content,
         "display_kind": _summary_display_kind(msg), "display_metadata": msg.get("display_metadata"),
         "platform_message_id": msg.get("platform_message_id"),  # load-bearing for restart drain-window recovery dedup
+        "topic_id": topic_id,
     }
     if isinstance(msg.get("_row_id"), int):
         row["_row_id"] = msg["_row_id"]
@@ -235,6 +241,13 @@ def _db_flush_write(agent, batch_rows: List[Dict[str, Any]], batch_msgs: List[Di
         # the live transcript so forks/compaction built from memory carry one checkpoint too. Markers stay:
         # the rows are durable exactly as the dicts now read.
         drop_shadowed_checkpoints(messages)
+    for written_row in batch_rows:
+        topic_id = written_row.get("topic_id")
+        if topic_id is not None:
+            try:
+                agent._session_db.update_topic_message_count(topic_id, 1)
+            except Exception:
+                pass
 
 
 def _db_flush_adopt_compression_tip(agent) -> bool:
@@ -316,6 +329,7 @@ def _persist_session(self, messages: List[Dict], conversation_history: List[Dict
         list used by the API call (#48677 is thus closed for every persist caller, not just this one).
         """
         from agent.agent_runtime_helpers import note_turn_persisted
+        self._ensure_topic_for_session()
         with _persist_lock(self):
             self._drop_trailing_empty_response_scaffolding(messages)
             self._session_messages = messages

agent/system_prompt.py

@@ -270,6 +270,49 @@ def _profile_name_for_home(home: Path) -> str:
         return "default"
 
 
+def _build_topic_detection_block(agent: Any) -> str:
+    """Build the session topic index and auto-detection instruction.
+
+    Returns an empty string if topic segmentation is not active or no
+    topics exist yet. Placed in the volatile prompt tail so it does not alter
+    the stable cached prefix.
+    """
+    db = getattr(agent, "_session_db", None)
+    session_id = getattr(agent, "session_id", None)
+    if not db or not session_id:
+        return ""
+
+    try:
+        topics = db.get_topics(session_id)
+    except Exception:
+        return ""
+
+    topic_rows = topics[:5]
+    lines = []
+    if topic_rows:
+        lines.append("## Session Topics")
+        lines.append("")
+        lines.append("| # | Topic | Msgs | State |")
+        lines.append("|---|-------|------|-------|")
+        for topic in topic_rows:
+            state_marker = "**active**" if topic["state"] == "active" else topic["state"]
+            lines.append(
+                f"| {topic['id']} | {topic['title']} | {topic['message_count']} | {state_marker} |"
+            )
+        if len(topics) > 5:
+            lines.extend(("", f"... and {len(topics) - 5} more archived topics."))
+        lines.append("")
+
+    # Ask agent to classify the topic as part of its response
+    lines.append(
+        "Append exactly one line to every response: TOPIC: <name>"
+    )
+    lines.append(
+        "Use the same name for follow-ups on the same subject."
+    )
+    return "\n".join(lines)
+
+
 def _tool_guidance_block(agent: Any) -> Optional[str]:
     """Tool-aware behavioral guidance, injected only when the tools are loaded."""
     names = agent.valid_tool_names
@@ -710,6 +753,9 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
     # a resumed process can reconstruct the stable prefix without re-running plugins.
     volatile_parts.extend(_plugin_section_blocks(_frozen_plugin_prompt_sections(agent), "after_memory"))
     volatile_parts.append(_timestamp_line(agent))
+    _topic_detection = _build_topic_detection_block(agent)
+    if _topic_detection:
+        volatile_parts.append(_topic_detection)
     # Keep the renderer-owned runtime anchor after all user/plugin prose so quoted
     # host examples cannot shadow it during persisted-prompt validation.
     if environment_hints:

agent/topic_detection.py

@@ -0,0 +1,208 @@
+"""Shared topic-shift detection for skill-graph injection and session topics.
+
+Two independent features need to answer the same question — "is this user
+message continuing the same topic as the previous one, or is it a new topic?":
+
+1.  **skill-graph** (plugin, ``pre_llm_call``): decides whether to inject a
+    full candidate list (new topic) or only the delta (same topic).
+2.  **session-topics** (core, ``system_prompt`` / ``run_agent``): provides a
+    pre-LLM hint so the model emits a more reliable ``TOPIC:`` signal.
+
+This module is a **pure-function library** — no hermes-agent imports, no
+global state, no I/O.  Callers manage their own per-session state and pass
+it in.  This keeps the module importable from both core and plugins (via
+``importlib``), and trivially unit-testable.
+
+Detection strategy: **AND gate** (calibrated on 18 real session transitions,
+94.4% accuracy vs 83% for LLM-only or S6-only):
+
+    topic_continuation = (
+        llm_topic_continuation is True
+        AND s6_cosine >= S6_THRESHOLD
+    )
+
+Both signals must agree on "same topic" to classify as continuation.  If
+either says "shift", we treat it as a shift.  This is deliberately biased
+toward false-shift (over-injection) over false-same (missing a new topic's
+skills).
+
+The primary signal is an LLM judgment piggy-backed onto the existing
+intent-split call (zero extra API cost).  The secondary signal is the
+cosine similarity between the first 100 characters of consecutive messages
+(S6 — first-100-char intent embedding).  S6 alone achieves 88.9% accuracy;
+combined with the LLM signal via AND gate, accuracy rises to 94.4%.
+
+When either signal is unavailable (LLM failed, no embedding backend), the
+available signal is used alone.  When both are unavailable, the function
+defaults to ``True`` (continuation) so the caller falls back to
+conservative full injection rather than blocking all injection.
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+from typing import Any, Optional
+
+logger = logging.getLogger("topic_detection")
+
+# Calibrated on 18 real session transitions (20260806_005634_165184).
+# S6 cosine at this threshold catches 2/3 LLM false-sames (8→9, 14→15)
+# while passing all 12 genuine same-topic transitions.
+S6_THRESHOLD: float = 0.47
+
+# Number of leading characters to use for the S6 embedding comparison.
+# Long messages dilute the core intent with context, quotes, and numbered
+# lists; the first ~100 chars capture what the user actually wants.
+INTENT_PREFIX_LEN: int = 100
+
+
+def compute_s6_cosine(
+    current_msg: str,
+    prev_msg: str,
+    embed_fn: Optional[Any] = None,
+) -> Optional[float]:
+    """Compute S6 cosine similarity between two messages.
+
+    Takes the first ``INTENT_PREFIX_LEN`` characters of each message,
+    embeds them via ``embed_fn``, and returns cosine similarity.
+
+    Args:
+        current_msg: Current user message text.
+        prev_msg: Previous user message text.
+        embed_fn: Callable that takes a string and returns a list[float]
+            embedding vector.  If ``None``, returns ``None``.
+
+    Returns:
+        Cosine similarity in [-1, 1], or ``None`` if embedding fails.
+    """
+    if not embed_fn:
+        return None
+    if not current_msg or not prev_msg:
+        return None
+
+    try:
+        intent_a = current_msg[:INTENT_PREFIX_LEN].strip()
+        intent_b = prev_msg[:INTENT_PREFIX_LEN].strip()
+        if not intent_a or not intent_b:
+            return None
+
+        emb_a = embed_fn(intent_a)
+        emb_b = embed_fn(intent_b)
+        if emb_a is None or emb_b is None:
+            return None
+
+        return _cosine(emb_a, emb_b)
+    except Exception as exc:
+        logger.debug("S6 cosine computation failed: %s", exc)
+        return None
+
+
+def detect_topic_shift(
+    current_msg: str,
+    prev_msg: Optional[str] = None,
+    llm_topic_continuation: Optional[bool] = None,
+    embed_fn: Optional[Any] = None,
+    *,
+    s6_threshold: float = S6_THRESHOLD,
+) -> dict[str, Any]:
+    """Determine whether ``current_msg`` continues the previous topic.
+
+    Uses the AND-gate strategy: both the LLM signal and S6 cosine must
+    agree on "same topic" for the result to be continuation.
+
+    Args:
+        current_msg: Current user message text.
+        prev_msg: Previous user message text, or ``None`` if this is the
+            first message in the session.
+        llm_topic_continuation: LLM judgment from the intent-split call
+            (``True`` = same topic, ``False`` = shift, ``None`` = LLM
+            unavailable).
+        embed_fn: Embedding callable for S6 cosine.  ``None`` to skip S6.
+        s6_threshold: Override the default S6 cosine threshold.
+
+    Returns:
+        Dict with keys:
+
+        - ``topic_continuation`` (bool): ``True`` if same topic, ``False``
+          if shift.  Defaults to ``True`` when no signals are available
+          (caller should fall back to full injection).
+        - ``confidence`` (float): 0.0–1.0, rough confidence based on how
+          many signals agreed.
+        - ``method`` (str): Which signal(s) were used — ``"and_gate"``,
+          ``"llm_only"``, ``"s6_only"``, or ``"fallback"``.
+        - ``s6_cosine`` (float|None): Raw S6 cosine value if computed.
+        - ``llm_signal`` (bool|None): Raw LLM signal if provided.
+    """
+    # First turn or no previous message → always "new topic" (full injection)
+    if not prev_msg:
+        return {
+            "topic_continuation": False,
+            "confidence": 1.0,
+            "method": "first_turn",
+            "s6_cosine": None,
+            "llm_signal": llm_topic_continuation,
+        }
+
+    # Compute S6 cosine (secondary signal)
+    s6_cosine = compute_s6_cosine(current_msg, prev_msg, embed_fn)
+
+    # ── AND gate: both signals must agree on "same" ──
+
+    has_llm = llm_topic_continuation is not None
+    has_s6 = s6_cosine is not None
+    s6_same = s6_cosine is not None and s6_cosine >= s6_threshold
+    llm_same = llm_topic_continuation is True
+
+    if has_llm and has_s6:
+        # Both available → AND gate
+        is_same = llm_same and s6_same
+        method = "and_gate"
+        # Confidence: both agree = high; disagree = the "no" wins but
+        # with lower confidence
+        if llm_same == s6_same:
+            confidence = 0.95
+        else:
+            confidence = 0.70
+    elif has_llm:
+        # LLM only
+        is_same = llm_same
+        method = "llm_only"
+        confidence = 0.83  # measured LLM-only accuracy
+    elif has_s6:
+        # S6 only
+        is_same = s6_same
+        method = "s6_only"
+        confidence = 0.89  # measured S6-only accuracy
+    else:
+        # Neither signal available → conservative: treat as continuation
+        # so the caller injects full (safer than silently suppressing)
+        # Actually — "continuation" means delta, which could suppress
+        # needed candidates. Better to say "shift" → full injection.
+        # But if prev_msg exists and we have NO signals at all, full
+        # injection is the right call.
+        return {
+            "topic_continuation": False,
+            "confidence": 0.0,
+            "method": "fallback",
+            "s6_cosine": None,
+            "llm_signal": None,
+        }
+
+    return {
+        "topic_continuation": is_same,
+        "confidence": confidence,
+        "method": method,
+        "s6_cosine": s6_cosine,
+        "llm_signal": llm_topic_continuation,
+    }
+
+
+def _cosine(a: list[float], b: list[float]) -> float:
+    """Cosine similarity between two vectors."""
+    dot = sum(x * y for x, y in zip(a, b))
+    norm_a = math.sqrt(sum(x * x for x in a))
+    norm_b = math.sqrt(sum(x * x for x in b))
+    if norm_a == 0 or norm_b == 0:
+        return 0.0
+    return dot / (norm_a * norm_b)

hermes_cli/cli_commands_mixin.py

@@ -1364,6 +1364,55 @@ def _resolve_resume_target(self, target: str):
             session_meta = self._session_db.get_session(target_id) or session_meta
         return target_id, session_meta
 
+    def _handle_topic_command(self, cmd_original: str) -> None:
+        """Handle /topic [list|switch N|new name] — session topic management."""
+        from cli import _cprint
+        parts = cmd_original.split(None, 2)
+        sub = parts[1].strip().lower() if len(parts) > 1 else ""
+        arg = parts[2].strip() if len(parts) > 2 else ""
+
+        db = getattr(self, "_session_db", None)
+        sid = getattr(self, "session_id", None)
+
+        if not db or not sid:
+            _cprint("  Session database not available.")
+            return
+
+        if sub == "list" or not sub:
+            topics = db.get_topics(sid)
+            if not topics:
+                _cprint("  No topics in this session yet.")
+                return
+            _cprint("  Session Topics:")
+            for topic in topics:
+                marker = " *" if topic["state"] == "active" else "  "
+                _cprint(f"  {marker} [{topic['id']}] {topic['title']} ({topic['message_count']} msgs, {topic['state']})")
+
+        elif sub == "switch" or sub == "sw":
+            if not arg:
+                _cprint("  Usage: /topic switch <id>")
+                return
+            try:
+                topic_id = int(arg)
+            except ValueError:
+                _cprint(f"  Invalid topic id: {arg}")
+                return
+            if db.set_active_topic(sid, topic_id):
+                _cprint(f"  Switched to topic {topic_id}.")
+            else:
+                _cprint(f"  Topic {topic_id} not found.")
+
+        elif sub == "new":
+            name = arg if arg else "unnamed"
+            # Archive current active first
+            db.set_active_topic(sid, 0)
+            topic_id = db.create_topic(sid, name)
+            db.set_topic_session_title(sid)
+            _cprint(f"  Created topic [{topic_id}] '{name}'.")
+
+        else:
+            _cprint("  Usage: /topic [list|switch N|new name]")
+
     def _handle_sessions_command(self, cmd_original: str) -> None:
         """Handle /sessions [list|<id_or_title>] — bare/``list`` prints the recent-sessions table;
         an explicit target delegates to /resume so both spellings behave identically."""

hermes_cli/commands.py

@@ -53,8 +53,9 @@ class CommandDef:
     CommandDef("new", "Start a new session (fresh session ID + history)", "Session",
                aliases=("reset",), args_hint="[name]",
                busy_policy="interrupt_then_dispatch", busy_handler="new"),
-    CommandDef("topic", "Enable or inspect Telegram DM topic sessions", "Session",
-               gateway_only=True, args_hint="[off|help|session-id]"),
+    CommandDef("topic", "Manage session conversation topics — list, switch, or create", "Session",
+               aliases=("topics",), args_hint="[list|switch N|new name]",
+               subcommands=("list", "switch", "new")),
     CommandDef("clear", "Clear screen and start a new session", "Session",
                cli_only=True, desktop="terminal"),
     CommandDef("redraw", "Force a full UI repaint (recovers from terminal drift)", "Session",