Honcho peer identity and pinPeerName setup improvements

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #30077 fix(honcho): runtime peer mapping — correctness follow-ups + setup wizard + docs

  merged to main

  files: agent/agent_init.py, agent/memory_provider.py, gateway/run.py, plugins/memory/honcho/README.md, plugins/memory/honcho/__init__.py, plugins/memory/honcho/cli.py, plugins/memory/honcho/client.py, plugins/memory/honcho/session.py, run_agent.py, scripts/release.py, tests/gateway/test_agent_cache.py, tests/honcho_plugin/test_cli.py, tests/honcho_plugin/test_pin_peer_name.py, tests/honcho_plugin/test_session.py, tests/run_agent/test_memory_provider_init.py, tests/test_honcho_session_context.py

  Show the diff

  • agent/agent_init.py

    @@ -183,6 +183,7 @@ def init_agent(
         prefill_messages: List[Dict[str, Any]] = None,
         platform: str = None,
         user_id: str = None,
    +    user_id_alt: str = None,
         user_name: str = None,
         chat_id: str = None,
         chat_name: str = None,
    @@ -265,6 +266,7 @@ def init_agent(
         agent.ephemeral_system_prompt = ephemeral_system_prompt
         agent.platform = platform  # "cli", "telegram", "discord", "whatsapp", etc.
         agent._user_id = user_id  # Platform user identifier (gateway sessions)
    +    agent._user_id_alt = user_id_alt  # Optional stable alternate platform identifier
         agent._user_name = user_name
         agent._chat_id = chat_id
         agent._chat_name = chat_name
    @@ -1089,6 +1091,8 @@ def init_agent(
                         # Thread gateway user identity for per-user memory scoping
                         if agent._user_id:
                             _init_kwargs["user_id"] = agent._user_id
    +                    if agent._user_id_alt:
    +                        _init_kwargs["user_id_alt"] = agent._user_id_alt
                         if agent._user_name:
                             _init_kwargs["user_name"] = agent._user_name
                         if agent._chat_id:

  • agent/memory_provider.py

    @@ -78,6 +78,7 @@ def initialize(self, session_id: str, **kwargs) -> None:
               - agent_workspace (str): Shared workspace name (e.g. "hermes").
               - parent_session_id (str): For subagents, the parent's session_id.
               - user_id (str): Platform user identifier (gateway sessions).
    +          - user_id_alt (str): Optional alternate stable platform user identifier.
             """
     
         def system_prompt_block(self) -> str:

  • gateway/run.py

    @@ -11436,6 +11436,7 @@ def run_sync():
                         session_id=task_id,
                         platform=platform_key,
                         user_id=source.user_id,
    +                    user_id_alt=source.user_id_alt,
                         user_name=source.user_name,
                         chat_id=source.chat_id,
                         chat_name=source.chat_name,
    @@ -14672,6 +14673,29 @@ def _extract_cache_busting_config(cls, user_config: dict | None) -> dict:
                 out["tools.registry_generation"] = getattr(registry, "_generation", None)
             except Exception:
                 out["tools.registry_generation"] = None
    +
    +        # Honcho identity-mapping keys live in honcho.json, not user_config.
    +        # HonchoSessionManager freezes the resolved peer_name / ai_peer /
    +        # pin / aliases / prefix at construction; without busting here,
    +        # mid-flight honcho.json edits go unread until the next unrelated
    +        # cache eviction.
    +        try:
    +            from plugins.memory.honcho.client import HonchoClientConfig
    +
    +            hcfg = HonchoClientConfig.from_global_config()
    +            out["honcho.peer_name"] = hcfg.peer_name
    +            out["honcho.ai_peer"] = hcfg.ai_peer
    +            out["honcho.pin_peer_name"] = bool(hcfg.pin_peer_name)
    +            out["honcho.runtime_peer_prefix"] = hcfg.runtime_peer_prefix or ""
    +            aliases = hcfg.user_peer_aliases or {}
    +            out["honcho.user_peer_aliases"] = sorted(aliases.items()) if isinstance(aliases, dict) else []
    +        except Exception:
    +            out["honcho.peer_name"] = None
    +            out["honcho.ai_peer"] = None
    +            out["honcho.pin_peer_name"] = None
    +            out["honcho.runtime_peer_prefix"] = None
    +            out["honcho.user_peer_aliases"] = None
    +
             return out
     
         @staticmethod
    @@ -14681,6 +14705,8 @@ def _agent_config_signature(
             enabled_toolsets: list,
             ephemeral_prompt: str,
             cache_keys: dict | None = None,
    +        user_id: str | None = None,
    +        user_id_alt: str | None = None,
         ) -> str:
             """Compute a stable string key from agent config values.
     
    @@ -14694,6 +14720,20 @@ def _agent_config_signature(
             the output of ``_extract_cache_busting_config(user_config)`` so
             edits to model.context_length / compression.* in config.yaml are
             picked up on the next gateway message without a manual restart.
    +
    +        ``user_id`` and ``user_id_alt`` are the runtime user identities
    +        carried by the current message's gateway source.  They participate
    +        in the cache key because the Honcho memory provider freezes them
    +        into ``HonchoSessionManager`` at first-message init (see
    +        ``plugins/memory/honcho/__init__.py::_do_session_init``).  Without
    +        them in the signature, a shared-thread session_key (one in which
    +        ``build_session_key`` intentionally omits the participant ID,
    +        e.g. ``thread_sessions_per_user=False``) would reuse the cached
    +        AIAgent across distinct users, causing the second user's messages
    +        to be attributed to the first user's resolved Honcho peer.  This
    +        broke #27371's per-user-peer contract in multi-user gateways.
    +        Per-user agent rebuilds in shared threads trade prompt-cache
    +        warmth for correct memory attribution.
             """
             import hashlib, json as _j
     
    @@ -14718,6 +14758,8 @@ def _agent_config_signature(
                     # cached agent and doesn't affect system prompt or tools.
                     ephemeral_prompt or "",
                     _cache_keys_sorted,
    +                str(user_id or ""),
    +                str(user_id_alt or ""),
                 ],
                 sort_keys=True,
                 default=str,
    @@ -16259,6 +16301,8 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
                     enabled_toolsets,
                     combined_ephemeral,
                     cache_keys=self._extract_cache_busting_config(user_config),
    +                user_id=getattr(source, "user_id", None),
    +                user_id_alt=getattr(source, "user_id_alt", None),
                 )
                 agent = None
                 _cache_lock = getattr(self, "_agent_cache_lock", None)
    @@ -16302,6 +16346,7 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
                         session_id=session_id,
                         platform=platform_key,
                         user_id=source.user_id,
    +                    user_id_alt=source.user_id_alt,
                         user_name=source.user_name,
                         chat_id=source.chat_id,
                         chat_name=source.chat_name,

  • plugins/memory/honcho/README.md

    @@ -127,6 +127,41 @@ For every key, resolution order is: **host block > root > env var > default**.
     | `peerName` | string | — | User peer identity |
     | `aiPeer` | string | host key | AI peer identity |
     
    +### Identity Mapping (Gateway Multi-User)
    +
    +In gateway deployments (Telegram, Discord, Slack, etc.) each user arrives with a platform-native runtime ID (Telegram UID, Discord snowflake, Slack user). These three keys control how those runtime IDs map to Honcho peers. The resolver is config-driven and deterministic — no automatic merging or runtime inference.
    +
    +| Key | Type | Default | Description |
    +|-----|------|---------|-------------|
    +| `pinUserPeer` | bool | `false` | When `true`, every gateway runtime user collapses to `peerName`. Single-operator deployments where you want all your platforms (and any other users) to share one peer. Also accepted as `pinPeerName` |
    +| `pinPeerName` | bool | `false` | Alias for `pinUserPeer`; same effect |
    +| `userPeerAliases` | object | `{}` | Map of runtime IDs to peer IDs (`{"86701400": "eri"}`). Many-to-one is the intended pattern — alias all your runtime IDs to one peer name. One-to-many is not supported; one runtime ID resolves to exactly one peer |
    +| `runtimePeerPrefix` | string | `""` | Prepended to unknown runtime IDs to namespace them (e.g. `"telegram_"` → `telegram_86701400`). Used only when no alias matches. Prevents collisions between platforms whose runtime IDs share the same shape |
    +
    +**Resolver ladder** (first match wins):
    +
    +```
    +1. pinUserPeer / pinPeerName=true → return peerName (ignore runtime ID)
    +2. userPeerAliases[runtime_id]   → return aliased peer
    +3. userPeerAliases[runtime_id_alt] → check alt-ID too (Telegram UID + username, etc.)
    +4. runtimePeerPrefix + runtime_id → namespaced peer, with sha256 collision escalation
    +5. raw sanitized runtime_id      → fallback peer
    +6. peerName                      → no runtime ID at all (CLI/TUI)
    +7. session-key fallback          → no config either
    +```
    +
    +**Why no `pinAiPeer`?** The AI peer is already pinned by construction — `aiPeer` is the only AI-side identity setting and the resolver never overrides it. Only the user-side peer has the runtime-vs-config tension that `pinUserPeer` resolves.
    +
    +**Host vs root semantics.** All three keys are accepted at both root and `hosts.<host>` levels. Host-level wins. For maps and prefixes, host-level *replaces* the root value as a whole (not merge), so a host can intentionally own its identity universe or wipe it with `userPeerAliases: {}` / `runtimePeerPrefix: ""`.
    +
    +**Deployment shapes** (`hermes honcho setup` asks one prompt to set these):
    +
    +- **Single-operator** — `pinUserPeer: true`. All gateway users → `peerName`. Recommended for personal use where you connect Hermes to your own Telegram/Discord/etc.
    +- **Multi-user gateway** — `pinUserPeer: false`, optional `runtimePeerPrefix`. Each runtime user → own peer. Recommended for bots serving many humans.
    +- **Hybrid** — `pinUserPeer: false`, `userPeerAliases` mapping the operator's runtime IDs to `peerName`. Multi-user gateway where YOU are routed but others stay distinct.
    +
    +**Migrating single → multi.** Flipping `pinUserPeer` from `true` to `false` does not migrate data. Memory accumulated under `peerName` while pinned stays there; runtime users now resolve to fresh, empty peers. To preserve your own continuity, use the **hybrid** shape — alias your runtime IDs back to `peerName` so your turns keep landing on the pooled history while other users get their own peers. The setup wizard offers this path automatically when it detects a single → multi transition.
    +
     ### Memory & Recall
     
     | Key | Type | Default | Description |

  • plugins/memory/honcho/__init__.py

    @@ -321,10 +321,8 @@ def initialize(self, session_id: str, **kwargs) -> None:
                 except Exception as e:
                     logger.debug("Honcho cost-awareness config parse error: %s", e)
     
    -            # ----- Port #1969: aiPeer sync from SOUL.md — REMOVED -----
    -            # SOUL.md is persona content, not identity config. aiPeer should
    -            # only come from honcho.json (host block or root) or the default.
    -            # See scratch/memory-plugin-ux-specs.md #10 for rationale.
    +            # aiPeer comes from honcho.json (host block or root) only.
    +            # SOUL.md is persona content, not identity config.
     
                 # ----- Port #1957: lazy session init for tools-only mode -----
                 if self._recall_mode == "tools":
    @@ -360,6 +358,7 @@ def _do_session_init(self, cfg, session_id: str, **kwargs) -> None:
                 config=cfg,
                 context_tokens=[redacted:secret-assignment]
                 runtime_user_peer_name=kwargs.get("user_id") or None,
    +            runtime_user_peer_name_alt=kwargs.get("user_id_alt") or None,
             )
     
             # ----- B3: resolve_session_name -----

  • plugins/memory/honcho/cli.py

    @@ -40,12 +40,20 @@ def clone_honcho_for_profile(profile_name: str) -> bool:
         if new_host in hosts:
             return False  # already exists
     
    -    # Clone settings from default block, override identity fields
    +    # Clone settings from default block, override identity fields.
    +    # Identity-mapping keys (pinPeerName/pinUserPeer, userPeerAliases,
    +    # runtimePeerPrefix) carry the operator's runtime-to-peer routing
    +    # intent from #27371.  Both pin keys are inherited because
    +    # HonchoClientConfig prefers pinUserPeer over pinPeerName — leaving
    +    # the canonical key off this allowlist silently drops the pin on
    +    # cloned profiles when the default uses the newer name.
         new_block = {}
         for key in ("recallMode", "writeFrequency", "sessionStrategy",
                     "sessionPeerPrefix", "contextTokens", "dialecticReasoningLevel",
                     "dialecticDynamic", "dialecticMaxChars", "messageMaxChars",
    -                "dialecticMaxInputChars", "saveMessages", "observation"):
    +                "dialecticMaxInputChars", "saveMessages", "observation",
    +                "pinPeerName", "pinUserPeer", "userPeerAliases",
    +                "runtimePeerPrefix"):
             val = default_block.get(key)
             if val is not None:
                 new_block[key] = val
    @@ -308,6 +316,72 @@ def _resolve_api_key(cfg: dict) -> str:
         return key
     
     
    +_IDENTITY_MAPPING_KEYS = (
    +    "pinPeerName",
    +    "pinUserPeer",
    +    "userPeerAliases",
    +    "runtimePeerPrefix",
    +)
    +
    +
    +def _resolve_effective_identity_mapping(
    +    cfg: dict, hermes_host: dict
    +) -> tuple[bool, dict, str, bool, bool]:
    +    """Resolve the effective identity-mapping state for the active host.
    +
    +    Matches the precedence used by ``HonchoClientConfig.from_global_config``
    +    so the wizard reads the same shape the gateway will actually run with.
    +    Without this, root-level overrides and ``pinUserPeer`` (which wins over
    +    ``pinPeerName`` at the same level) are invisible to detection, letting
    +    setup mis-classify the current shape and silently change effective
    +    routing on the next save.
    +
    +    Returns ``(pin, aliases, prefix, aliases_from_root, prefix_from_root)``.
    +    The ``*_from_root`` flags let the write step skip touching host keys
    +    whose value is actually inherited.
    +    """
    +    pin = False
    +    for val in (
    +        hermes_host.get("pinUserPeer"),
    +        hermes_host.get("pinPeerName"),
    +        cfg.get("pinUserPeer"),
    +        cfg.get("pinPeerName"),
    +    ):
    +        if val is not None:
    +            pin = bool(val)
    +            break
    +
    +    if "userPeerAliases" in hermes_host:
    +        aliases_src = hermes_host.get("userPeerAliases")
    +        aliases_from_root = False
    +    else:
    +        aliases_src = cfg.get("userPeerAliases")
    +        aliases_from_root = aliases_src is not None
    +    aliases = aliases_src if isinstance(aliases_src, dict) else {}
    +
    +    if "runtimePeerPrefix" in hermes_host:
    +        prefix_src = hermes_host.get("runtimePeerPrefix")
    +        prefix_from_root = False
    +    else:
    +        prefix_src = cfg.get("runtimePeerPrefix")
    +        prefix_from_root = prefix_src is not None
    +    prefix = str(prefix_src or "")
    +
    +    return pin, aliases, prefix, aliases_from_root, prefix_from_root
    +
    +
    +def _scrub_identity_mapping(hermes_host: dict) -> None:
    +    """Drop every peer-mapping key from the host block.
    +
    +    Called before the wizard writes a chosen shape so latent precedence
    +    conflicts can't survive — e.g. a stray host ``pinUserPeer: false``
    +    that would silently outrank a freshly written ``pinPeerName: true``
    +    (host ``pinUserPeer`` is first in the resolver ladder).
    +    """
    +    for key in _IDENTITY_MAPPING_KEYS:
    +        hermes_host.pop(key, None)
    +
    +
     def _prompt(label: str, default: str | None = None, secret: bool = False) -> str:
         suffix = f" [{default}]" if default else ""
         sys.stdout.write(f"  {label}{suffix}: ")
    @@ -435,6 +509,131 @@ def cmd_setup(args) -> None:
         if new_workspace:
             hermes_host["workspace"] = new_workspace
     
    +    # --- 3b. Deployment shape ---
    +    # Determines how runtime user identities (Telegram UIDs, Discord
    +    # snowflakes, etc.) map to Honcho peers in gateway sessions.  Three
    +    # shapes cover the realistic deployments; each writes a different
    +    # combination of pinPeerName / userPeerAliases / runtimePeerPrefix.
    +    # See plugins/memory/honcho/README.md for the resolver ladder.
    +    #
    +    # Detection must mirror the gateway resolver: root-level config and
    +    # ``pinUserPeer`` (which outranks ``pinPeerName`` at the same level)
    +    # both affect effective routing, so reading host-only fields would
    +    # mis-classify a profile that inherits its mapping from root or uses
    +    # the newer canonical key.
    +    (
    +        current_pin,
    +        current_aliases,
    +        current_prefix,
    +        aliases_from_root,
    +        prefix_from_root,
    +    ) = _resolve_effective_identity_mapping(cfg, hermes_host)
    +
    +    if current_pin:
    +        current_shape = "single"
    +    elif current_aliases:
    +        current_shape = "hybrid"
    +    else:
    +        current_shape = "multi"
    +
    +    print("\n  Deployment shape (how gateway users map to peers):")
    +    print("    single -- all platforms route to your peer (recommended for personal use)")
    +    print("    multi  -- each platform user gets their own peer (multi-user bots)")
    +    print("    hybrid -- multi-user, but YOUR runtime IDs alias to your peer")
    +    print("    skip   -- don't touch identity-mapping config")
    +    new_shape = _prompt("Deployment shape", default=current_shape).strip().lower()
    +
    +    # Transitioning single → multi orphans the peerName pool for runtime users
    +    # (their resolved peers go from peerName to runtime-derived IDs with empty
    +    # history).  Steer the operator toward hybrid so their own continuity is
    +    # preserved via alias mappings.
    +    if current_shape == "single" and new_shape == "multi":
    +        peer_target = hermes_host.get("peerName") or current_peer or "user"
    +        print(
    +            f"\n  ⚠ Switching from single to multi will orphan memory accumulated\n"
    +            f"    under peer '{peer_target}'.  Existing runtime users (Telegram,\n"
    +            f"    Discord, etc.) will resolve to fresh, empty peers."
    +        )
    +        print("    To keep your own continuity, choose 'hybrid' and alias your\n"
    +              "    runtime IDs back to peerName.")
    +        confirm = _prompt("Continue with multi anyway? (yes/hybrid/no)", default="hybrid").strip().lower()
    +        if confirm in {"hybrid", "h"}:
    +            new_shape = "hybrid"
    +        elif confirm not in {"yes", "y"}:
    +            new_shape = "skip"
    +
    +    # Each shape branch scrubs every peer-mapping key before writing its own,
    +    # so a stale ``pinUserPeer`` left behind by an earlier setup run can't
    +    # outrank the freshly written ``pinPeerName`` via host-level precedence.
    +    if new_shape == "single":
    +        _scrub_identity_mapping(hermes_host)
    +        hermes_host["pinPeerName"] = True
    +        print(f"  pinPeerName=true → all gateway users route to '{hermes_host.get('peerName', '?')}'.")
    +    elif new_shape == "multi":
    +        # Preserve operator-curated, host-level aliases so multi → multi
    +        # re-runs don't drop them.  Root-sourced aliases are left to
    +        # cascade naturally and are NOT copied down into the host.
    +        prior_aliases = (
    +            dict(current_aliases)
    +            if isinstance(current_aliases, dict) and not aliases_from_root
    +            else {}
    +        )
    +        _scrub_identity_mapping(hermes_host)
    +        hermes_host["pinPeerName"] = False
    +        # Do NOT auto-write ``userPeerAliases: {}``: an empty host map
    +        # would override any root-level ``userPeerAliases`` the operator
    +        # set as a cross-host baseline, silently disabling those aliases.
    +        # Absence is the right "no host opinion" signal.
    +        if prior_aliases:
    +            hermes_host["userPeerAliases"] = prior_aliases
    +        _prefix_default = current_prefix or ""
    +        _new_prefix = _prompt(
    +            "Runtime peer prefix (e.g. 'telegram_', blank for none)",
    +            default=_prefix_default,
    +        ).strip()
    +        # Only write a host-level prefix when the operator typed one that
    +        # diverges from the inherited root value; otherwise let the root
    +        # cascade continue unmodified.
    +        if _new_prefix and not (prefix_from_root and _new_prefix == current_prefix):
    +            hermes_host["runtimePeerPrefix"] = _new_prefix
    +        print("  Multi-user mode: each runtime ID → own peer. Use 'hermes honcho status' to inspect.")
    +    elif new_shape == "hybrid":
    +        # Hybrid encodes operator intent at the host level: collect existing
    +        # entries (host or root) so the wizard never silently drops a known
    +        # alias, then write the combined map.  Materialising root entries
    +        # into the host is the right move here — once the operator answers
    +        # the alias prompts for a host, they're declaring "this host owns
    +        # the mapping".
    +        existing_aliases = dict(current_aliases) if isinstance(current_aliases, dict) else {}
    +        _scrub_identity_mapping(hermes_host)
    +        hermes_host["pinPeerName"] = False
    +        peer_target = hermes_host.get("peerName") or current_peer or "user"
    +        print(f"\n  Add runtime IDs that should alias to peer '{peer_target}'.")
    +        print("  Leave blank to skip a platform.  Existing aliases are preserved.")
    +        for platform_label, alias_hint in (
    +            ("Telegram UID", "e.g. 86701400"),
    +            ("Discord snowflake", "e.g. 491827364"),
    +            ("Slack user ID", "e.g. U04ABCDEF"),
    +            ("Matrix MXID", "e.g. @you:matrix.org"),
    +        ):
    +            entered = _prompt(f"  {platform_label} ({alias_hint})", default="").strip()
    +            if entered:
    +                existing_aliases[entered] = peer_target
    +        if existing_aliases:
    +            hermes_host["userPeerAliases"] = existing_aliases
    +        _prefix_default = current_prefix or ""
    +        _new_prefix = _prompt(
    +            "Runtime peer prefix for unknown users (e.g. 'telegram_', blank for none)",
    +            default=_prefix_default,
    +        ).strip()
    +        if _new_prefix and not (prefix_from_root and _new_prefix == current_prefix):
    +            hermes_host["runtimePeerPrefix"] = _new_prefix
    +        print(f"  Hybrid mode: your runtime IDs → '{peer_target}', others → own peer.")
    +    elif new_shape == "skip":
    +        pass  # leave config untouched
    +    else:
    +        print(f"  Unknown shape '{new_shape}' — leaving identity-mapping config untouched.")
    +
         # --- 4. Observation mode ---
         current_obs = hermes_host.get("observationMode") or cfg.get("observationMode", "directional")
         print("\n  Observation mode:")

  • plugins/memory/honcho/client.py

    @@ -91,12 +91,17 @@ def _normalize_recall_mode(val: str) -> str:
         return val if val in _VALID_RECALL_MODES else "hybrid"
     
     
    -def _resolve_bool(host_val, root_val, *, default: bool) -> bool:
    -    """Resolve a bool config field: host wins, then root, then default."""
    -    if host_val is not None:
    -        return bool(host_val)
    -    if root_val is not None:
    -        return bool(root_val)
    +def _resolve_bool(*vals, default: bool) -> bool:
    +    """Resolve a bool config field: first non-None wins, else default.
    +
    +    Variadic to support aliased keys (e.g. ``pinUserPeer`` shadowing
    +    ``pinPeerName`` for backwards compatibility).  Pass values in
    +    precedence order: caller's preferred alias first, then fallback
    +    aliases, in (host, root) interleaving as needed.
    +    """
    +    for val in vals:
    +        if val is not None:
    +            return bool(val)
         return default
     
     
    @@ -122,6 +127,34 @@ def _parse_int_config(host_val, root_val, default: int) -> int:
         return default
     
     
    +def _parse_string_map(host_obj: dict, root_obj: dict, key: str) -> dict[str, str]:
    +    """Parse a string-to-string map with host-level whole-map override."""
    +    source = host_obj[key] if key in host_obj else root_obj.get(key)
    +    if not isinstance(source, dict):
    +        return {}
    +
    +    result: dict[str, str] = {}
    +    for raw_key, raw_value in source.items():
    +        alias_key = str(raw_key).strip()
    +        alias_value = str(raw_value).strip() if raw_value is not None else ""
    +        if alias_key and alias_value:
    +            result[alias_key] = alias_value
    +    return result
    +
    +
    +def _parse_optional_string(
    +    host_obj: dict, root_obj: dict, key: str, default: str = ""
    +) -> str:
    +    """Parse a string field where host-level empty string can override root."""
    +    if key in host_obj:
    +        value = host_obj.get(key)
    +    else:
    +        value = root_obj.get(key, default)
    +    if value is None:
    +        return default
    +    return str(value).strip()
    +
    +
     def _parse_dialectic_depth(host_val, root_val) -> int:
         """Parse dialecticDepth: host wins, then root, then 1. Clamped to 1-3."""
         for val in (host_val, root_val):
    @@ -259,6 +292,12 @@ class HonchoClientConfig:
         # each platform would fork memory into its own peer (#14984).  Default
         # ``False`` preserves existing multi-user behaviour.
         pin_peer_name: bool = False
    +    # Map gateway runtime user IDs to stable Honcho user peers. Host-level
    +    # config replaces the root map as a whole so profiles can intentionally
    +    # own their identity mappings.
    +    user_peer_aliases: dict[str, str] = field(default_factory=dict)
    +    # Optional prefix for unknown gateway runtime user IDs, e.g. "telegram_".
    +    runtime_peer_prefix: str = ""
         # Toggles
         enabled: bool = False
         save_messages: bool = True
    @@ -454,10 +493,28 @@ def from_global_config(
                 peer_name=host_block.get("peerName") or raw.get("peerName"),
                 ai_peer=ai_peer,
                 pin_peer_name=_resolve_bool(
    +                # ``pinUserPeer`` is the clearer name (the resolver pins
    +                # the user-side peer to ``peerName``, ignoring runtime
    +                # identity).  ``pinPeerName`` is the original key from
    +                # #14984 and stays accepted for backward compatibility.
    +                # Host-level keys win over root-level; among same-level
    +                # keys, ``pinUserPeer`` wins over ``pinPeerName``.
    +                host_block.get("pinUserPeer"),
                     host_block.get("pinPeerName"),
    +                raw.get("pinUserPeer"),
                     raw.get("pinPeerName"),
                     default=False,
                 ),
    +            user_peer_aliases=_parse_string_map(
    +                host_block,
    +                raw,
    +                "userPeerAliases",
    +            ),
    +            runtime_peer_prefix=_parse_optional_string(
    +                host_block,
    +                raw,
    +                "runtimePeerPrefix",
    +            ),
                 enabled=enabled,
                 save_messages=save_messages,
                 write_frequency=write_frequency,

  • plugins/memory/honcho/session.py

    @@ -2,6 +2,7 @@
     
     from __future__ import annotations
     
    +import hashlib
     import queue
     import re
     import logging
    @@ -19,6 +20,8 @@
     
     # Sentinel to signal the async writer thread to shut down
     _ASYNC_SHUTDOWN = object()
    +_PEER_ID_HASH_LEN = 8
    +_PEER_ID_HASH_ESCALATION_LENGTHS = (_PEER_ID_HASH_LEN, 12, 16, 24, 32, 64)
     
     
     @dataclass
    @@ -79,6 +82,7 @@ def __init__(
             context_tokens: int | None = None,
             config: Any | None = None,
             runtime_user_peer_name: str | None = None,
    +        runtime_user_peer_name_alt: str | None = None,
         ):
             """
             Initialize the session manager.
    @@ -89,11 +93,13 @@ def __init__(
                 config: HonchoClientConfig from global config (provides peer_name, ai_peer,
                         write_frequency, observation, etc.).
                 runtime_user_peer_name: Gateway user identity for per-user memory scoping.
    +            runtime_user_peer_name_alt: Optional stable alternate gateway identity.
             """
             self._honcho = honcho
             self._context_tokens = [redacted:secret-assignment]
             self._config = config
             self._runtime_user_peer_name = runtime_user_peer_name
    +        self._runtime_user_peer_name_alt = runtime_user_peer_name_alt
             self._cache: dict[str, HonchoSession] = {}
             self._cache_lock = threading.RLock()
             self._peers_cache: dict[str, Any] = {}
    @@ -267,6 +273,90 @@ def _sanitize_id(self, id_str: str) -> str:
             """Sanitize an ID to match Honcho's pattern: ^[a-zA-Z0-9_-]+"""
             return re.sub(r'[^a-zA-Z0-9_-]', '-', id_str)
     
    +    def _runtime_user_ids(self) -> list[str]:
    +        """Return runtime identity candidates in lookup order."""
    +        candidates: list[str] = []
    +        for value in (self._runtime_user_peer_name, self._runtime_user_peer_name_alt):
    +            if value is None:
    +                continue
    +            candidate = str(value).strip()
    +            if candidate and candidate not in candidates:
    +                candidates.append(candidate)
    +        return candidates
    +
    +    def _session_key_fallback_peer_id(self, key: str) -> str:
    +        parts = key.split(":", 1)
    +        channel = parts[0] if len(parts) > 1 else "default"
    +        chat_id = parts[1] if len(parts) > 1 else key
    +        return self._sanitize_id(f"user-{channel}-{chat_id}")
    +
    +    def _explicit_user_peer_ids(self) -> set[str]:
    +        """Return sanitized user peer IDs that came from explicit config."""
    +        if self._config is None:
    +            return set()
    +
    +        explicit_ids: set[str] = set()
    +        peer_name = getattr(self._config, "peer_name", None)
    +        if peer_name:
    +            explicit_ids.add(self._sanitize_id(str(peer_name).strip()))
    +
    +        aliases = getattr(self._config, "user_peer_aliases", {})
    +        if isinstance(aliases, dict):
    +            for alias in aliases.values():
    +                if isinstance(alias, str) and alias.strip():
    +                    explicit_ids.add(self._sanitize_id(alias.strip()))
    +
    +        return explicit_ids
    +
    +    def _generated_runtime_peer_id(self, prefix: str, runtime_id: str) -> str:
    +        """Return a stable peer ID for an unknown prefixed runtime user."""
    +        raw_peer_id = f"{prefix}{runtime_id}"
    +        sanitized_peer_id = self._sanitize_id(raw_peer_id)
    +        explicit_ids = self._explicit_user_peer_ids()
    +        if (
    +            sanitized_peer_id != raw_peer_id
    +            or sanitized_peer_id in explicit_ids
    +        ):
    +            digest = hashlib.sha256(raw_peer_id.encode("utf-8")).hexdigest()
    +            for hash_len in _PEER_ID_HASH_ESCALATION_LENGTHS:
    +                candidate = f"{sanitized_peer_id}-{digest[:hash_len]}"
    +                if candidate not in explicit_ids:
    +                    return candidate
    +            return f"{sanitized_peer_id}-{digest}"
    +        return sanitized_peer_id
    +
    +    def _resolve_user_peer_id(self, key: str) -> str:
    +        """Resolve the Honcho user peer ID for this manager/session."""
    +        pin_peer_name = (
    +            self._config is not None
    +            and bool(getattr(self._config, "peer_name", None))
    +            and getattr(self._config, "pin_peer_name", False) is True
    +        )
    +        if pin_peer_name:
    +            return self._sanitize_id(self._config.peer_name)
    +
    +        runtime_ids = self._runtime_user_ids()
    +        if runtime_ids:
    +            aliases = getattr(self._config, "user_peer_aliases", {}) if self._config else {}
    +            if not isinstance(aliases, dict):
    +                aliases = {}
    +            for runtime_id in runtime_ids:
    +                alias = aliases.get(runtime_id)
    +                if isinstance(alias, str) and alias.strip():
    +                    return self._sanitize_id(alias.strip())
    +
    +            primary_runtime_id = runtime_ids[0]
    +            prefix = getattr(self._config, "runtime_peer_prefix", "") if self._config else ""
    +            prefix = prefix.strip() if isinstance(prefix, str) else ""
    +            if prefix:
    +                return self._generated_runtime_peer_id(prefix, primary_runtime_id)
    +            return self._sanitize_id(primary_runtime_id)
    +
    +        if self._config and self._config.peer_name:
    +            return self._sanitize_id(self._config.peer_name)
    +
    +        return self._session_key_fallback_peer_id(key)
    +
         def get_or_create(self, key: str) -> HonchoSession:
             """
             Get an existing session or create a new one.
    @@ -285,31 +375,11 @@ def get_or_create(self, key: str) -> HonchoSession:
             # Determine peer IDs — no lock needed (read-only, no shared state mutation).
             # Gateway sessions normally use the runtime user identity (the
             # platform-native ID: Telegram UID, Discord snowflake, Slack user,
    -        # etc.) so multi-user bots scope memory per user.  For a single-user
    -        # deployment the config-supplied ``peer_name`` is an unambiguous
    -        # identity and we should keep it unified across platforms — see
    -        # #14984.  Opt into that with ``hosts.<host>.pinPeerName: true`` in
    -        # ``honcho.json`` (or root-level ``pinPeerName: true``).
    -        # `is True` (not `bool(...)`) is deliberate: several multi-user tests
    -        # pass a ``MagicMock`` for ``config`` where ``mock.pin_peer_name``
    -        # silently returns another MagicMock — truthy by default.  Requiring
    -        # strict ``True`` keeps pinning as opt-in even for callers that
    -        # haven't updated their mocks yet; real configs built via
    -        # ``from_global_config`` always produce a proper boolean.
    -        pin_peer_name = (
    -            self._config is not None
    -            and bool(getattr(self._config, "peer_name", None))
    -            and getattr(self._config, "pin_peer_name", False) is True
    -        )
    -        if self._runtime_user_peer_name and not pin_peer_name:
    -            user_peer_id = self._sanitize_id(self._runtime_user_peer_name)
    -        elif self._config and self._config.peer_name:
    -            user_peer_id = self._sanitize_id(self._config.peer_name)
    -        else:
    -            parts = key.split(":", 1)
    -            channel = parts[0] if len(parts) > 1 else "default"
    -            chat_id = parts[1] if len(parts) > 1 else key
    -            user_peer_id = self._sanitize_id(f"user-{channel}-{chat_id}")
    +        # etc.) so multi-user bots scope memory per user.  Config can alias
    +        # known runtime IDs or prefix unknown IDs.  For a single-user
    +        # deployment, ``pinPeerName`` still pins all runtime identities to
    +        # ``peerName`` (see #14984).
    +        user_peer_id = self._resolve_user_peer_id(key)
     
             assistant_peer_id = self._sanitize_id(
                 self._config.ai_peer if self._config else "hermes-assistant"
    @@ -937,11 +1007,11 @@ def get_session_context(self, session_key: str, peer: str = "user") -> dict[str,
                 return self._fetch_peer_context(peer_id, target=peer_id)
     
             try:
    -            peer_id = self._resolve_peer_id(session, peer)
    +            observer_peer_id, target_peer_id = self._resolve_observer_target(session, peer)
                 ctx = honcho_session.context(
                     summary=True,
    -                peer_target=peer_id,
    -                peer_perspective=session.user_peer_id if peer == "user" else session.assistant_peer_id,
    +                peer_target=target_peer_id or observer_peer_id,
    +                peer_perspective=observer_peer_id,
                 )
     
                 result: dict[str, Any] = {}
    @@ -1017,7 +1087,14 @@ def get_peer_card(self, session_key: str, peer: str = "user") -> list[str]:
     
             try:
                 observer_peer_id, target_peer_id = self._resolve_observer_target(session, peer)
    -            return self._fetch_peer_card(observer_peer_id, target=target_peer_id)
    +            card = self._fetch_peer_card(observer_peer_id, target=target_peer_id)
    +            if card:
    +                return card
    +            # Some backends store cards directly on the target peer, not the
    +            # observer-target slot. Fall back so honcho_profile still works.
    +            if target_peer_id:
    +                return self._fetch_peer_card(target_peer_id)
    +            return []
             except Exception as e:
                 logger.debug("Failed to fetch peer card from Honcho: %s", e)
                 return []
    @@ -1164,13 +1241,22 @@ def set_peer_card(self, session_key: str, card: list[str], peer: str = "user") -
             if not session:
                 return None
             try:
    -            peer_id = self._resolve_peer_id(session, peer)
    -            if peer_id is None:
    +            observer_peer_id, target_peer_id = self._resolve_observer_target(session, peer)
    +            if observer_peer_id is None:
                     logger.warning("Could not resolve peer '%s' for set_peer_card in session '%s'", peer, session_key)
                     return None
    -            peer_obj = self._get_or_create_peer(peer_id)
    -            result = peer_obj.set_card(card)
    -            logger.info("Updated peer card for %s (%d facts)", peer_id, len(card))
    +            peer_obj = self._get_or_create_peer(observer_peer_id)
    +            result = (
    +                peer_obj.set_card(card, target=target_peer_id)
    +                if target_peer_id is not None
    +                else peer_obj.set_card(card)
    +            )
    +            logger.info(
    +                "Updated peer card observer=%s target=%s (%d facts)",
    +                observer_peer_id,
    +                target_peer_id or observer_peer_id,
    +                len(card),
    +            )
                 return result
             except Exception as e:
                 logger.error("Failed to set peer card: %s", e)

  • run_agent.py

    @@ -393,6 +393,7 @@ def __init__(
             prefill_messages: List[Dict[str, Any]] = None,
             platform: str = None,
             user_id: str = None,
    +        user_id_alt: str = None,
             user_name: str = None,
             chat_id: str = None,
             chat_name: str = None,
    @@ -462,6 +463,7 @@ def __init__(
                 prefill_messages=prefill_messages,
                 platform=platform,
                 user_id=user_id,
    +            user_id_alt=user_id_alt,
                 user_name=user_name,
                 chat_id=chat_id,
                 chat_name=chat_name,

  • scripts/release.py

    @@ -213,6 +213,8 @@
         "[email protected]": "maciekczech",
         "[email protected]": "LeonSGP43",
         "[redacted:email]": "LeonSGP43",
    +    "[redacted:email]": "cyb0rgk1tty",
    +    "[redacted:email]": "cmullins70",
         "[redacted:email]": "zeejaytan",
         "[redacted:email]": "Aslaaen",
         "[redacted:email]": "Tkander1715",

  • tests/gateway/test_agent_cache.py

    @@ -1344,3 +1344,71 @@ def test_watchdog_accumulation_across_recursive_turns(self):
                 f"Watchdog would see {idle_secs:.0f}s idle, expected ~{STUCK_FOR}s. "
                 "Inactivity timeout could not fire for a stuck interrupted turn."
             )
    +
    +
    +class TestAgentConfigSignatureUserId:
    +    """Shared-thread cache must not reuse an agent across users.
    +
    +    HonchoSessionManager freezes the resolved runtime user identity at
    +    first-message init.  When the gateway session_key omits the participant
    +    ID (``thread_sessions_per_user=False``), a cached AIAgent created by
    +    user A would otherwise be reused for user B, attributing B's writes to
    +    A's resolved peer.  Including ``user_id`` / ``user_id_alt`` in the
    +    signature forces per-user agent builds in shared threads.
    +
    +    Tradeoff: cold prompt cache for each user's first turn in a shared
    +    thread, in exchange for correct memory attribution.
    +    """
    +
    +    def test_signature_changes_with_user_id(self):
    +        from gateway.run import GatewayRunner
    +        runtime = {"provider": "anthropic", "api_key": "k", "base_url": "", "api_mode": "chat_completions"}
    +        sig_a = GatewayRunner._agent_config_signature(
    +            "claude-sonnet-4", runtime, ["hermes-telegram"], "", user_id="86701400"
    +        )
    +        sig_b = GatewayRunner._agent_config_signature(
    +            "claude-sonnet-4", runtime, ["hermes-telegram"], "", user_id="491827364"
    +        )
    +        assert sig_a != sig_b
    +
    +    def test_signature_stable_with_same_user_id(self):
    +        from gateway.run import GatewayRunner
    +        runtime = {"provider": "anthropic", "api_key": "k", "base_url": "", "api_mode": "chat_completions"}
    +        sig_1 = GatewayRunner._agent_config_signature(
    +            "claude-sonnet-4", runtime, ["hermes-telegram"], "", user_id="86701400"
    +        )
    +        sig_2 = GatewayRunner._agent_config_signature(
    +            "claude-sonnet-4", runtime, ["hermes-telegram"], "", user_id="86701400"
    +        )
    +        assert sig_1 == sig_2
    +
    +    def test_signature_changes_with_user_id_alt(self):
    +        from gateway.run import GatewayRunner
    +        runtime = {"provider": "anthropic", "api_key": "k", "base_url": "", "api_mode": "chat_completions"}
    +        sig_a = GatewayRunner._agent_config_signature(
    +            "claude-sonnet-4", runtime, ["hermes-telegram"], "",
    +            user_id="86701400", user_id_alt="@igor_tg",
    +        )
    +        sig_b = GatewayRunner._agent_config_signature(
    +            "claude-sonnet-4", runtime, ["hermes-telegram"], "",
    +            user_id="86701400", user_id_alt="@erosika_tg",
    +        )
    +        assert sig_a != sig_b
    +
    +    def test_signature_omits_user_id_when_absent(self):
    +        """Default-None user_id must not change signatures vs unset call.
    +
    +        Callers that pass no user_id kwarg must produce a signature
    +        byte-identical to ``user_id=None`` so in-flight caches survive
    +        the rollout of this fix.
    +        """
    +        from gateway.run import GatewayRunner
    +        runtime = {"provider": "anthropic", "api_key": "k", "base_url": "", "api_mode": "chat_completions"}
    +        sig_implicit = GatewayRunner._agent_config_signature(
    +            "claude-sonnet-4", runtime, ["hermes-telegram"], "",
    +        )
    +        sig_explicit_none = GatewayRunner._agent_config_signature(
    +            "claude-sonnet-4", runtime, ["hermes-telegram"], "",
    +            user_id=None, user_id_alt=None,
    +        )
    +        assert sig_implicit == sig_explicit_none

  • tests/honcho_plugin/test_cli.py

    @@ -153,4 +153,424 @@ def _boom(hcfg, client):
     
             out = capsys.readouterr().out
             assert "FAILED (Invalid API key)" in out
    -        assert "Connection... OK" not in out
    \ No newline at end of file
    +        assert "Connection... OK" not in out
    +
    +
    +class TestCloneHonchoForProfile:
    +    """Identity-key carryover during profile cloning.
    +
    +    The host-scoped identity-mapping keys (``userPeerAliases``,
    +    ``runtimePeerPrefix``, ``pinPeerName``) must survive a clone; otherwise
    +    the new profile silently fragments memory by resolving gateway users to
    +    raw runtime IDs instead of operator-declared peers.
    +    """
    +
    +    def _setup_clone_env(self, monkeypatch, tmp_path, cfg):
    +        import plugins.memory.honcho.cli as honcho_cli
    +        cfg_path = tmp_path / "config.json"
    +        cfg_path.write_text("{}")
    +        monkeypatch.setattr(honcho_cli, "_read_config", lambda: cfg)
    +        monkeypatch.setattr(honcho_cli, "_config_path", lambda: cfg_path)
    +        monkeypatch.setattr(honcho_cli, "_local_config_path", lambda: cfg_path)
    +        monkeypatch.setattr(honcho_cli, "_ensure_peer_exists", lambda host_key=None: True)
    +        written = {}
    +        def _write(c, path=None):
    +            written["cfg"] = c
    +        monkeypatch.setattr(honcho_cli, "_write_config", _write)
    +        return honcho_cli, written
    +
    +    def test_user_peer_aliases_carry_into_cloned_profile(self, monkeypatch, tmp_path):
    +        cfg = {
    +            "apiKey": "***",
    +            "hosts": {
    +                "hermes": {
    +                    "userPeerAliases": {"86701400": "eri", "discord-491827364": "eri"},
    +                    "peerName": "eri",
    +                },
    +            },
    +        }
    +        honcho_cli, written = self._setup_clone_env(monkeypatch, tmp_path, cfg)
    +        ok = honcho_cli.clone_honcho_for_profile("coder")
    +        assert ok is True
    +        new_block = written["cfg"]["hosts"]["hermes.coder"]
    +        assert new_block["userPeerAliases"] == {"86701400": "eri", "discord-491827364": "eri"}
    +
    +    def test_runtime_peer_prefix_carries_into_cloned_profile(self, monkeypatch, tmp_path):
    +        cfg = {
    +            "apiKey": "***",
    +            "hosts": {
    +                "hermes": {
    +                    "runtimePeerPrefix": "telegram_",
    +                    "peerName": "eri",
    +                },
    +            },
    +        }
    +        honcho_cli, written = self._setup_clone_env(monkeypatch, tmp_path, cfg)
    +        ok = honcho_cli.clone_honcho_for_profile("coder")
    +        assert ok is True
    +        new_block = written["cfg"]["hosts"]["hermes.coder"]
    +        assert new_block["runtimePeerPrefix"] == "telegram_"
    +
    +    def test_pin_peer_name_carries_into_cloned_profile(self, monkeypatch, tmp_path):
    +        cfg = {
    +            "apiKey": "***",
    +            "hosts": {
    +                "hermes": {
    +                    "pinPeerName": True,
    +                    "peerName": "eri",
    +                },
    +            },
    +        }
    +        honcho_cli, written = self._setup_clone_env(monkeypatch, tmp_path, cfg)
    +        ok = honcho_cli.clone_honcho_for_profile("coder")
    +        assert ok is True
    +        new_block = written["cfg"]["hosts"]["hermes.coder"]
    +        assert new_block["pinPeerName"] is True
    +
    +    def test_unset_identity_keys_do_not_appear_in_cloned_profile(self, monkeypatch, tmp_path):
    +        cfg = {
    +            "apiKey": "***",
    +            "hosts": {"hermes": {"peerName": "eri"}},
    +        }
    +        honcho_cli, written = self._setup_clone_env(monkeypatch, tmp_path, cfg)
    +        ok = honcho_cli.clone_honcho_for_profile("coder")
    +        assert ok is True
    +        new_block = written["cfg"]["hosts"]["hermes.coder"]
    +        assert "userPeerAliases" not in new_block
    +        assert "runtimePeerPrefix" not in new_block
    +        assert "pinPeerName" not in new_block
    +
    +
    +class TestSetupWizardDeploymentShape:
    +    """The deployment-shape step writes pinPeerName / userPeerAliases /
    +    runtimePeerPrefix based on the operator's chosen shape.
    +
    +    Single-operator deployments collapse all platforms to peerName.
    +    Multi-user gateways leave the resolver to route per-runtime.
    +    Hybrid deployments alias the operator's own runtime IDs only.
    +
    +    These tests script the interactive _prompt calls and assert the
    +    resulting hermes_host block, so the wizard's deployment-shape
    +    semantics stay locked even as adjacent prompts are added.
    +    """
    +
    +    def _run_setup(self, monkeypatch, tmp_path, *, answers, initial_cfg=None):
    +        import plugins.memory.honcho.cli as honcho_cli
    +
    +        cfg_path = tmp_path / "config.json"
    +        cfg_path.write_text("{}")
    +        cfg = initial_cfg if initial_cfg is not None else {"apiKey": "***"}
    +
    +        monkeypatch.setattr(honcho_cli, "_read_config", lambda: cfg)
    +        monkeypatch.setattr(honcho_cli, "_config_path", lambda: cfg_path)
    +        monkeypatch.setattr(honcho_cli, "_local_config_path", lambda: cfg_path)
    +        monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
    +        monkeypatch.setattr(honcho_cli, "_ensure_sdk_installed", lambda: True)
    +        monkeypatch.setattr(honcho_cli, "_write_config", lambda *a, **k: None)
    +
    +        # Bypass config.yaml + connection test side effects.
    +        monkeypatch.setattr(
    +            "hermes_cli.config.load_config", lambda: {"memory": {}}, raising=False,
    +        )
    +        monkeypatch.setattr(
    +            "hermes_cli.config.save_config", lambda c: None, raising=False,
    +        )
    +
    +        class _FakeClientCfg:
    +            def resolve_session_name(self):
    +                return "hermes-test"
    +            workspace_id = "hermes"
    +            peer_name = "eri"
    +            ai_peer = "hermetika"
    +            observation_mode = "directional"
    +            write_frequency = "async"
    +            recall_mode = "hybrid"
    +            session_strategy = "per-session"
    +
    +        monkeypatch.setattr(
    +            "plugins.memory.honcho.client.HonchoClientConfig.from_global_config",
    +            lambda host=None: _FakeClientCfg(),
    +        )
    +        monkeypatch.setattr(
    +            "plugins.memory.honcho.client.reset_honcho_client",
    +            lambda: None,
    +        )
    +        monkeypatch.setattr(
    +            "plugins.memory.honcho.client.get_honcho_client",
    +            lambda hcfg: object(),
    +        )
    +
    +        # Scripted _prompt: pop answers in order. Default-return for unconsumed prompts.
    +        answer_iter = iter(answers)
    +        def _scripted_prompt(label, default=None, secret=[redacted:secret-assignment]
    +            try:
    +                return next(answer_iter)
    +            except StopIteration:
    +                return default if default is not None else ""
    +        monkeypatch.setattr(honcho_cli, "_prompt", _scripted_prompt)
    +
    +        honcho_cli.cmd_setup(SimpleNamespace())
    +        return cfg["hosts"]["hermes"]
    +
    +    def test_single_shape_sets_pin_peer_name_and_clears_aliases(self, monkeypatch, tmp_path):
    +        answers = [
    +            "cloud",           # deployment
    +            "",                # api key (keep)
    +            "eri",             # peer name
    +            "hermetika",       # ai peer
    +            "hermes",          # workspace
    +            "single",          # deployment shape ← key answer
    +            # remaining prompts fall through to defaults
    +        ]
    +        initial_cfg = {
    +            "apiKey": "***",
    +            "hosts": {"hermes": {
    +                "userPeerAliases": {"old": "stale"},
    +                "runtimePeerPrefix": "old_",
    +            }},
    +        }
    +        host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
    +        assert host["pinPeerName"] is True
    +        assert "userPeerAliases" not in host
    +        assert "runtimePeerPrefix" not in host
    +
    +    def test_multi_shape_leaves_pin_false_and_accepts_prefix(self, monkeypatch, tmp_path):
    +        answers = [
    +            "cloud",           # deployment
    +            "",                # api key (keep)
    +            "eri",             # peer name
    +            "hermetika",       # ai peer
    +            "hermes",          # workspace
    +            "multi",           # deployment shape
    +            "telegram_",       # runtime peer prefix
    +        ]
    +        host = self._run_setup(monkeypatch, tmp_path, answers=answers)
    +        assert host["pinPeerName"] is False
    +        # Multi must NOT auto-write ``userPeerAliases: {}``: an empty host
    +        # map would silently override a root-level baseline.  Absence is
    +        # the correct "no host opinion" signal.
    +        assert "userPeerAliases" not in host
    +        assert host["runtimePeerPrefix"] == "telegram_"
    +
    +    def test_hybrid_shape_aliases_operator_runtime_ids_to_peer_name(self, monkeypatch, tmp_path):
    +        answers = [
    +            "cloud",           # deployment
    +            "",                # api key (keep)
    +            "eri",             # peer name
    +            "hermetika",       # ai peer
    +            "hermes",          # workspace
    +            "hybrid",          # deployment shape
    +            "86701400",        # telegram uid
    +            "491827364",       # discord snowflake
    +            "",                # slack (skip)
    +            "",                # matrix (skip)
    +            "",                # runtime peer prefix (skip)
    +        ]
    +        host = self._run_setup(monkeypatch, tmp_path, answers=answers)
    +        assert host["pinPeerName"] is False
    +        assert host["userPeerAliases"] == {
    +            "86701400": "eri",
    +            "491827364": "eri",
    +        }
    +        assert "runtimePeerPrefix" not in host
    +
    +    def test_skip_shape_preserves_existing_identity_config(self, monkeypatch, tmp_path):
    +        initial_cfg = {
    +            "apiKey": "***",
    +            "hosts": {"hermes": {
    +                "pinPeerName": True,
    +                "userPeerAliases": {"keep": "me"},
    +                "runtimePeerPrefix": "keep_",
    +            }},
    +        }
    +        answers = [
    +            "cloud", "", "eri", "hermetika", "hermes", "skip",
    +        ]
    +        host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
    +        assert host["pinPeerName"] is True
    +        assert host["userPeerAliases"] == {"keep": "me"}
    +        assert host["runtimePeerPrefix"] == "keep_"
    +
    +    def test_single_to_multi_steers_to_hybrid_by_default(self, monkeypatch, tmp_path):
    +        """Flipping single → multi triggers a warning that auto-steers the
    +        operator to ``hybrid`` (default), so their own runtime IDs keep
    +        landing on peerName instead of orphaning the pinned-pool history.
    +        """
    +        initial_cfg = {
    +            "apiKey": "***",
    +            "hosts": {"hermes": {"pinPeerName": True, "peerName": "eri"}},
    +        }
    +        answers = [
    +            "cloud",           # deployment
    +            "",                # api key (keep)
    +            "eri",             # peer name
    +            "hermetika",       # ai peer
    +            "hermes",          # workspace
    +            "multi",           # deployment shape — triggers the guard
    +            "hybrid",          # guard response: accept the steer
    +            "86701400",        # telegram uid
    +            "",                # discord (skip)
    +            "",                # slack (skip)
    +            "",                # matrix (skip)
    +            "",                # runtime prefix (skip)
    +        ]
    +        host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
    +        assert host["pinPeerName"] is False
    +        assert host["userPeerAliases"] == {"86701400": "eri"}
    +
    +    def test_single_to_multi_yes_override_keeps_multi(self, monkeypatch, tmp_path):
    +        """Operator can override the steer by answering ``yes`` and accept
    +        the orphaning consequences.  This is the explicit undo-the-pin path.
    +        """
    +        initial_cfg = {
    +            "apiKey": "***",
    +            "hosts": {"hermes": {"pinPeerName": True, "peerName": "eri"}},
    +        }
    +        answers = [
    +            "cloud", "", "eri", "hermetika", "hermes",
    +            "multi",           # deployment shape — triggers the guard
    +            "yes",             # guard response: confirm multi
    +            "telegram_",       # runtime peer prefix
    +        ]
    +        host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
    +        assert host["pinPeerName"] is False
    +        # See test_multi_shape_leaves_pin_false_and_accepts_prefix.
    +        assert "userPeerAliases" not in host
    +        assert host["runtimePeerPrefix"] == "telegram_"
    +
    +    def test_host_pin_user_peer_true_is_detected_as_single(self, monkeypatch, tmp_path):
    +        """Host-level ``pinUserPeer: true`` must classify as ``single``.
    +
    +        Pressing Enter at the shape prompt then preserves the pin instead
    +        of falling through to ``multi`` and orphaning the user's memory
    +        pool — the bug the wizard regressed when ``pinUserPeer`` landed
    +        as a higher-precedence alias.
    +        """
    +        initial_cfg = {
    +            "apiKey": "***",
    +            "hosts": {"hermes": {"pinUserPeer": True, "peerName": "eri"}},
    +        }
    +        # Exhaust the iterator before the shape prompt so the scripted
    +        # mock falls through to the prompt's default (which is the
    +        # wizard-detected shape).  Scripting an explicit "" would NOT
    +        # exercise that fallthrough — the mock returns it literally.
    +        answers = ["cloud", "", "eri", "hermetika", "hermes"]
    +        host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
    +        # Scrub-then-write normalises onto pinPeerName and drops the alias
    +        # so resolver precedence can't reintroduce ambiguity.
    +        assert host["pinPeerName"] is True
    +        assert "pinUserPeer" not in host
    +
    +    def test_host_pin_user_peer_false_overrides_root_pin_peer_name(
    +        self, monkeypatch, tmp_path
    +    ):
    +        """Host ``pinUserPeer: false`` outranks host ``pinPeerName`` in the
    +        resolver.  Detection must agree, otherwise the wizard would offer
    +        ``single`` as the default and silently re-pin a profile the
    +        operator explicitly unpinned via the newer key.
    +        """
    +        initial_cfg = {
    +            "apiKey": "***",
    +            "hosts": {"hermes": {
    +                "pinUserPeer": False,
    +                "pinPeerName": True,
    +                "peerName": "eri",
    +            }},
    +        }
    +        answers = ["cloud", "", "eri", "hermetika", "hermes"]
    +        host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
    +        assert host["pinPeerName"] is False
    +        assert "pinUserPeer" not in host
    +
    +    def test_root_user_peer_aliases_detected_as_hybrid(self, monkeypatch, tmp_path):
    +        """Root-level ``userPeerAliases`` must classify as ``hybrid`` even
    +        when the host block has no aliases of its own.
    +        """
    +        initial_cfg = {
    +            "apiKey": "***",
    +            "userPeerAliases": {"86701400": "eri"},
    +            "hosts": {"hermes": {"peerName": "eri"}},
    +        }
    +        answers = ["cloud", "", "eri", "hermetika", "hermes"]
    +        host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
    +        assert host["pinPeerName"] is False
    +        # Hybrid materialises the root aliases into the host so subsequent
    +        # operator edits live on the host block they're inspecting.
    +        assert host["userPeerAliases"] == {"86701400": "eri"}
    +
    +    def test_multi_does_not_override_root_user_peer_aliases(self, monkeypatch, tmp_path):
    +        """Explicit ``multi`` must leave the host ``userPeerAliases`` key
    +        absent, preserving any root-level aliases as a cross-host baseline.
    +
    +        Picking ``multi`` here is an active choice — detection would have
    +        defaulted to ``hybrid`` because root aliases exist — so the
    +        operator's intent is to drop the alias mapping for this host.
    +        We honor that by writing ``pinPeerName: false`` only, and rely
    +        on the host's absence of ``userPeerAliases`` to inherit root.
    +        That inheritance is intentional: a true wipe would require the
    +        operator to delete the root key explicitly.
    +        """
    +        initial_cfg = {
    +            "apiKey": "***",
    +            "userPeerAliases": {"baseline": "eri"},
    +            "hosts": {"hermes": {"peerName": "eri"}},
    +        }
    +        answers = [
    +            "cloud", "", "eri", "hermetika", "hermes",
    +            "multi",           # explicit multi override of detected hybrid
    +        ]
    +        host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
    +        assert host["pinPeerName"] is False
    +        assert "userPeerAliases" not in host
    +
    +    def test_single_scrubs_stale_pin_user_peer_false(self, monkeypatch, tmp_path):
    +        """Choosing ``single`` must drop any host-level ``pinUserPeer``,
    +        otherwise an existing ``pinUserPeer: false`` would outrank the
    +        freshly written ``pinPeerName: true`` and leave the profile
    +        effectively unpinned (the P1 latent-precedence regression).
    +        """
    +        initial_cfg = {
    +            "apiKey": "***",
    +            "hosts": {"hermes": {
    +                "pinUserPeer": False,
    +                "peerName": "eri",
    +            }},
    +        }
    +        answers = [
    +            "cloud", "", "eri", "hermetika", "hermes",
    +            "single",
    +        ]
    +        host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
    +        assert host["pinPeerName"] is True
    +        assert "pinUserPeer" not in host
    +
    +
    +class TestCloneCarriesPinUserPeer:
    +    """``pinUserPeer`` (canonical name for ``pinPeerName``) must survive a
    +    profile clone.  Without this, a default profile that uses the newer
    +    key would silently produce cloned profiles without the pin even
    +    though the resolver prefers ``pinUserPeer`` over ``pinPeerName``.
    +    """
    +
    +    def test_clone_inherits_host_pin_user_peer(self, monkeypatch, tmp_path):
    +        import plugins.memory.honcho.cli as honcho_cli
    +
    +        cfg = {
    +            "apiKey": "***",
    +            "hosts": {"hermes": {"pinUserPeer": True, "peerName": "eri"}},
    +        }
    +        cfg_path = tmp_path / "config.json"
    +        cfg_path.write_text("{}")
    +        monkeypatch.setattr(honcho_cli, "_read_config", lambda: cfg)
    +        monkeypatch.setattr(honcho_cli, "_config_path", lambda: cfg_path)
    +        monkeypatch.setattr(honcho_cli, "_local_config_path", lambda: cfg_path)
    +        monkeypatch.setattr(honcho_cli, "_ensure_peer_exists", lambda host_key=None: True)
    +        written = {}
    +        monkeypatch.setattr(
    +            honcho_cli, "_write_config", lambda c, path=None: written.setdefault("cfg", c),
    +        )
    +
    +        ok = honcho_cli.clone_honcho_for_profile("partner")
    +        assert ok is True
    +        new_block = written["cfg"]["hosts"]["hermes.partner"]
    +        assert new_block["pinUserPeer"] is True

  • tests/honcho_plugin/test_pin_peer_name.py

    @@ -1,24 +1,20 @@
    -"""Tests for the ``pinPeerName`` config flag (#14984).
    -
    -By default, when Hermes runs under a gateway (Telegram, Discord, Slack, ...)
    -it passes the platform-native user ID as ``runtime_user_peer_name`` into
    -``HonchoSessionManager``.  That ID wins over any configured ``peer_name``
    -so multi-user bots scope memory per user.
    -
    -For a single-user personal deployment where the user connects over multiple
    -platforms, that default forks memory into one Honcho peer per platform
    -(Telegram UID, Discord snowflake, Slack user ID, ...).  The user asked for
    -an opt-in knob that pins the user peer to ``peer_name`` from ``honcho.json``
    -so the same person's memory stays unified regardless of which platform the
    -turn arrived on — ``hosts.<host>.pinPeerName: true`` (or root-level
    -``pinPeerName: true``).
    -
    -These tests exercise both the config parsing (``client.py::from_global_config``)
    -and the resolution order (``session.py::get_or_create``).  We stub the
    -Honcho API calls so we can assert the chosen ``user_peer_id`` without
    -touching the network.
    +"""Tests for the ``pinPeerName`` / ``pinUserPeer`` config flag.
    +
    +Under a gateway (Telegram, Discord, Slack, ...) Hermes passes the
    +platform-native user ID as ``runtime_user_peer_name`` into
    +``HonchoSessionManager``.  By default that ID wins over any configured
    +``peer_name`` so multi-user bots scope memory per user.
    +
    +For single-user deployments connecting over multiple platforms,
    +``pinUserPeer: true`` pins the user peer to ``peer_name`` so memory stays
    +unified across platforms.
    +
    +Tests cover config parsing (``client.py::from_global_config``) and resolver
    +order (``session.py::get_or_create``), stubbing Honcho API calls so the
    +chosen ``user_peer_id`` can be asserted without touching the network.
     """
     
    +import hashlib
     import json
     from unittest.mock import MagicMock
     
    @@ -99,6 +95,90 @@ def test_explicit_false_parses(self, tmp_path, monkeypatch):
             assert config.pin_peer_name is False
     
     
    +class TestRuntimePeerMappingConfigParsing:
    +    def test_defaults_are_empty(self):
    +        config = HonchoClientConfig()
    +        assert config.user_peer_aliases == {}
    +        assert config.runtime_peer_prefix == ""
    +
    +    def test_root_level_aliases_and_prefix_parse(self, tmp_path):
    +        config_file = tmp_path / "honcho.json"
    +        config_file.write_text(json.dumps({
    +            "apiKey": "k",
    +            "userPeerAliases": {
    +                " 86701400 ": " Igor ",
    +                "": "ignored",
    +                "empty-value": " ",
    +                "null-value": None,
    +            },
    +            "runtimePeerPrefix": "telegram_",
    +        }))
    +
    +        config = HonchoClientConfig.from_global_config(config_path=config_file)
    +
    +        assert config.user_peer_aliases == {"86701400": "Igor"}
    +        assert config.runtime_peer_prefix == "telegram_"
    +
    +    def test_host_aliases_override_root_aliases_as_whole_map(self, tmp_path):
    +        config_file = tmp_path / "honcho.json"
    +        config_file.write_text(json.dumps({
    +            "apiKey": "k",
    +            "userPeerAliases": {"root-user": "root-peer"},
    +            "hosts": {
    +                "hermes": {
    +                    "userPeerAliases": {"host-user": "host-peer"},
    +                },
    +            },
    +        }))
    +
    +        config = HonchoClientConfig.from_global_config(config_path=config_file)
    +
    +        assert config.user_peer_aliases == {"host-user": "host-peer"}
    +
    +    def test_host_empty_aliases_disable_root_aliases(self, tmp_path):
    +        config_file = tmp_path / "honcho.json"
    +        config_file.write_text(json.dumps({
    +            "apiKey": "k",
    +            "userPeerAliases": {"root-user": "root-peer"},
    +            "hosts": {
    +                "hermes": {
    +                    "userPeerAliases": {},
    +                },
    +            },
    +        }))
    +
    +        config = HonchoClientConfig.from_global_config(config_path=config_file)
    +
    +        assert config.user_peer_aliases == {}
    +
    +    def test_host_empty_prefix_disables_root_prefix(self, tmp_path):
    +        config_file = tmp_path / "honcho.json"
    +        config_file.write_text(json.dumps({
    +            "apiKey": "k",
    +            "runtimePeerPrefix": "telegram_",
    +            "hosts": {
    +                "hermes": {
    +                    "runtimePeerPrefix": "",
    +                },
    +            },
    +        }))
    +
    +        config = HonchoClientConfig.from_global_config(config_path=config_file)
    +
    +        assert config.runtime_peer_prefix == ""
    +
    +    def test_malformed_alias_config_is_ignored(self, tmp_path):
    +        config_file = tmp_path / "honcho.json"
    +        config_file.write_text(json.dumps({
    +            "apiKey": "k",
    +            "userPeerAliases": ["not", "a", "map"],
    +        }))
    +
    +        config = HonchoClientConfig.from_global_config(config_path=config_file)
    +
    +        assert config.user_peer_aliases == {}
    +
    +
     # ---------------------------------------------------------------------------
     # Peer resolution (the actual bug fix)
     # ---------------------------------------------------------------------------
    @@ -119,13 +199,24 @@ def _patch_manager_for_resolution_test(mgr: HonchoSessionManager) -> None:
     class TestPeerResolutionOrder:
         """Matrix of (runtime_id, pin_peer_name, peer_name) → expected user_peer_id."""
     
    -    def _config(self, *, peer_name: str | None, pin_peer_name: bool) -> HonchoClientConfig:
    +    def _config(
    +        self,
    +        *,
    +        peer_name: str | None,
    +        pin_peer_name: bool,
    +        user_peer_aliases: dict[str, str] | None = None,
    +        runtime_peer_prefix: str = "",
    +        session_peer_prefix: bool = False,
    +    ) -> HonchoClientConfig:
             # The test doesn't need auth / Honcho — disable the provider so
             # the manager doesn't try to open a real client.
             return HonchoClientConfig(
                 api_key="[redacted:secret-assignment]",
                 peer_name=peer_name,
                 pin_peer_name=pin_peer_name,
    +            user_peer_aliases=user_peer_aliases or {},
    +            runtime_peer_prefix=runtime_peer_prefix,
    +            session_peer_prefix=session_peer_prefix,
                 enabled=False,
                 write_frequency="turn",  # avoid spawning the async writer thread
             )
    @@ -148,11 +239,177 @@ def test_runtime_wins_when_pin_is_false(self):
                 "bot immediately merges memory across users."
             )
     
    +    def test_alias_wins_for_known_runtime_id(self):
    +        """Known platform IDs can preserve an existing stable Honcho peer."""
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name="Igor",
    +                pin_peer_name=False,
    +                user_peer_aliases={"86701400": "Igor"},
    +                runtime_peer_prefix="telegram_",
    +            ),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("telegram:86701400")
    +        assert session.user_peer_id == "Igor"
    +
    +    def test_unknown_runtime_id_uses_prefix(self):
    +        """Unknown gateway users stay isolated but become platform-scoped."""
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name="Igor",
    +                pin_peer_name=False,
    +                runtime_peer_prefix="telegram_",
    +            ),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("telegram:86701400")
    +        assert session.user_peer_id == "telegram_86701400"
    +
    +    def test_prefixed_runtime_id_hashes_when_sanitization_is_lossy(self):
    +        """Generated prefixed IDs avoid merges caused by lossy sanitization."""
    +        raw_peer_id = "telegram_user:42"
    +        expected_hash = hashlib.sha256(raw_peer_id.encode("utf-8")).hexdigest()[:8]
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name=None,
    +                pin_peer_name=False,
    +                runtime_peer_prefix="telegram_",
    +            ),
    +            runtime_user_peer_name="user:42",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("telegram:user:42")
    +        assert session.user_peer_id == f"telegram_user-42-{expected_hash}"
    +
    +    def test_prefixed_runtime_id_hashes_when_it_collides_with_peer_name(self):
    +        """Unknown generated peers should not silently merge into peerName."""
    +        raw_peer_id = "telegram_86701400"
    +        expected_hash = hashlib.sha256(raw_peer_id.encode("utf-8")).hexdigest()[:8]
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name="telegram_86701400",
    +                pin_peer_name=False,
    +                runtime_peer_prefix="telegram_",
    +            ),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("telegram:86701400")
    +        assert session.user_peer_id == f"telegram_86701400-{expected_hash}"
    +
    +    def test_prefixed_runtime_id_hashes_when_it_collides_with_alias_target(self):
    +        """Unknown generated peers should not silently merge into alias targets."""
    +        raw_peer_id = "telegram_86701400"
    +        expected_hash = hashlib.sha256(raw_peer_id.encode("utf-8")).hexdigest()[:8]
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name=None,
    +                pin_peer_name=False,
    +                user_peer_aliases={"known-user": "telegram_86701400"},
    +                runtime_peer_prefix="telegram_",
    +            ),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("telegram:86701400")
    +        assert session.user_peer_id == f"telegram_86701400-{expected_hash}"
    +
    +    def test_prefixed_runtime_id_extends_hash_when_short_hash_collides(self):
    +        raw_peer_id = "telegram_86701400"
    +        digest = hashlib.sha256(raw_peer_id.encode("utf-8")).hexdigest()
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name=None,
    +                pin_peer_name=False,
    +                user_peer_aliases={
    +                    "known-user": "telegram_86701400",
    +                    "reserved-user": f"telegram_86701400-{digest[:8]}",
    +                },
    +                runtime_peer_prefix="telegram_",
    +            ),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("telegram:86701400")
    +        assert session.user_peer_id == f"telegram_86701400-{digest[:12]}"
    +
    +    def test_alias_value_is_sanitized_after_selection(self):
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name=None,
    +                pin_peer_name=False,
    +                user_peer_aliases={"86701400": "Alice Smith!"},
    +            ),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("telegram:86701400")
    +        assert session.user_peer_id == "Alice-Smith-"
    +
    +    def test_alias_keys_match_raw_runtime_id_before_sanitization(self):
    +        """Alias selection is exact on platform IDs before Honcho ID cleanup."""
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name=None,
    +                pin_peer_name=False,
    +                user_peer_aliases={
    +                    "user:42": "raw-match",
    +                    "user-42": "sanitized-match",
    +                },
    +            ),
    +            runtime_user_peer_name="user:42",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("telegram:user:42")
    +        assert session.user_peer_id == "raw-match"
    +
    +    def test_session_peer_prefix_is_orthogonal_to_runtime_peer_prefix(self):
    +        """sessionPeerPrefix scopes session IDs; runtimePeerPrefix scopes user peers."""
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name="Igor",
    +                pin_peer_name=False,
    +                runtime_peer_prefix="telegram_",
    +                session_peer_prefix=True,
    +            ),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("telegram:86701400")
    +        assert session.user_peer_id == "telegram_86701400"
    +        assert session.honcho_session_id == "telegram-86701400"
    +
         def test_config_wins_when_pin_is_true(self):
    -        """The #14984 fix: single-user deployments opt into config pinning."""
    +        """With pin enabled, configured peer_name beats runtime ID."""
             mgr = HonchoSessionManager(
                 honcho=MagicMock(),
    -            config=self._config(peer_name="Igor", pin_peer_name=True),
    +            config=self._config(
    +                peer_name="Igor",
    +                pin_peer_name=True,
    +                user_peer_aliases={"86701400": "Alias"},
    +                runtime_peer_prefix="telegram_",
    +            ),
                 runtime_user_peer_name="86701400",  # Telegram pushes this in
             )
             _patch_manager_for_resolution_test(mgr)
    @@ -167,7 +424,23 @@ def test_config_wins_when_pin_is_true(self):
         def test_pin_noop_when_peer_name_missing(self):
             """Safety: pinPeerName alone (no peer_name) must not silently drop
             the runtime identity.  Without a configured peer_name there's
    -        nothing to pin to — fall back to runtime as before."""
    +        nothing to pin to — fall through to runtime mapping."""
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name=None,
    +                pin_peer_name=True,
    +                user_peer_aliases={"86701400": "Igor"},
    +                runtime_peer_prefix="telegram_",
    +            ),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("telegram:86701400")
    +        assert session.user_peer_id == "Igor"
    +
    +    def test_pin_noop_without_peer_name_or_mapping_preserves_runtime(self):
             mgr = HonchoSessionManager(
                 honcho=MagicMock(),
                 config=self._config(peer_name=None, pin_peer_name=True),
    @@ -176,11 +449,42 @@ def test_pin_noop_when_peer_name_missing(self):
             _patch_manager_for_resolution_test(mgr)
     
             session = mgr.get_or_create("telegram:86701400")
    -        assert session.user_peer_id == "86701400", (
    -            "pin_peer_name=True with no peer_name set must not strip the "
    -            "runtime ID — otherwise the user peer would collapse to the "
    -            "session-key fallback and lose per-user scoping entirely"
    +        assert session.user_peer_id == "86701400"
    +
    +    def test_alt_runtime_id_can_match_alias_without_changing_raw_fallback(self):
    +        """Stable alternate IDs can map known users while primary ID fallback stays unchanged."""
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name=None,
    +                pin_peer_name=False,
    +                user_peer_aliases={"union-user": "Igor"},
    +                runtime_peer_prefix="feishu_",
    +            ),
    +            runtime_user_peer_name="open-id",
    +            runtime_user_peer_name_alt="union-user",
             )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("feishu:chat")
    +        assert session.user_peer_id == "Igor"
    +
    +    def test_alt_runtime_id_does_not_replace_primary_prefix_fallback(self):
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._config(
    +                peer_name=None,
    +                pin_peer_name=False,
    +                user_peer_aliases={"other-union": "Igor"},
    +                runtime_peer_prefix="feishu_",
    +            ),
    +            runtime_user_peer_name="open-id",
    +            runtime_user_peer_name_alt="union-user",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +
    +        session = mgr.get_or_create("feishu:chat")
    +        assert session.user_peer_id == "feishu_open-id"
     
         def test_runtime_missing_falls_back_to_peer_name(self):
             """CLI-mode (no gateway runtime identity) uses config peer_name —
    @@ -233,9 +537,8 @@ def test_pin_does_not_affect_assistant_peer(self):
     
     
     class TestCrossPlatformMemoryUnification:
    -    """The user-visible outcome of the #14984 fix: the same physical user
    -    talking to Hermes via Telegram AND Discord should land on ONE peer
    -    (not two) when pinPeerName is opted in.
    +    """The same physical user talking to Hermes via Telegram AND Discord
    +    lands on ONE peer when ``pinPeerName`` is opted in.
         """
     
         def _config_pinned(self) -> HonchoClientConfig:
    @@ -305,3 +608,277 @@ def test_multiuser_default_keeps_platforms_separate(self):
                 "multi-user default MUST keep users separate — a regression "
                 "here would silently merge unrelated users' memory"
             )
    +
    +
    +class TestPinUserPeerAlias:
    +    """``pinUserPeer`` and ``pinPeerName`` both resolve to the same internal
    +    ``pin_peer_name`` field.  Precedence when both appear: host pinUserPeer →
    +    host pinPeerName → root pinUserPeer → root pinPeerName → default.
    +    """
    +
    +    def test_root_pinUserPeer_true_pins(self, tmp_path):
    +        from plugins.memory.honcho.client import HonchoClientConfig
    +        import json
    +        config_file = tmp_path / "honcho.json"
    +        config_file.write_text(json.dumps({
    +            "apiKey": "***",
    +            "peerName": "eri",
    +            "pinUserPeer": True,
    +        }))
    +        config = HonchoClientConfig.from_global_config(config_path=config_file)
    +        assert config.pin_peer_name is True
    +
    +    def test_host_pinUserPeer_wins_over_root_pinPeerName(self, tmp_path):
    +        from plugins.memory.honcho.client import HonchoClientConfig
    +        import json
    +        config_file = tmp_path / "honcho.json"
    +        config_file.write_text(json.dumps({
    +            "apiKey": "***",
    +            "peerName": "eri",
    +            "pinPeerName": False,
    +            "hosts": {"hermes": {"pinUserPeer": True}},
    +        }))
    +        config = HonchoClientConfig.from_global_config(config_path=config_file)
    +        assert config.pin_peer_name is True
    +
    +    def test_host_pinUserPeer_false_disables_root_pinPeerName(self, tmp_path):
    +        from plugins.memory.honcho.client import HonchoClientConfig
    +        import json
    +        config_file = tmp_path / "honcho.json"
    +        config_file.write_text(json.dumps({
    +            "apiKey": "***",
    +            "peerName": "eri",
    +            "pinPeerName": True,
    +            "hosts": {"hermes": {"pinUserPeer": False}},
    +        }))
    +        config = HonchoClientConfig.from_global_config(config_path=config_file)
    +        assert config.pin_peer_name is False, (
    +            "Host-level pinUserPeer=false must override root-level "
    +            "pinPeerName=true so a host can unpin a globally-pinned profile."
    +        )
    +
    +    def test_pinPeerName_still_works_unchanged(self, tmp_path):
    +        from plugins.memory.honcho.client import HonchoClientConfig
    +        import json
    +        config_file = tmp_path / "honcho.json"
    +        config_file.write_text(json.dumps({
    +            "apiKey": "***",
    +            "peerName": "eri",
    +            "hosts": {"hermes": {"pinPeerName": True}},
    +        }))
    +        config = HonchoClientConfig.from_global_config(config_path=config_file)
    +        assert config.pin_peer_name is True
    +
    +
    +class TestPinTransition:
    +    """Behavior when honcho.json flips ``pinPeerName`` true → false.
    +
    +    Covers two contracts:
    +      1. A freshly-built manager picks up the flipped config and resolves
    +         the same runtime ID to a new peer (no resolver staleness).
    +      2. The gateway's agent-cache signature reflects honcho identity-mapping
    +         changes, so a config edit busts the cached AIAgent on the next turn.
    +    """
    +
    +    def _pinned(self) -> HonchoClientConfig:
    +        return HonchoClientConfig(
    +            api_key="k",
    +            peer_name="Igor",
    +            pin_peer_name=True,
    +            enabled=False,
    +            write_frequency="turn",
    +        )
    +
    +    def _unpinned(self) -> HonchoClientConfig:
    +        return HonchoClientConfig(
    +            api_key="k",
    +            peer_name="Igor",
    +            pin_peer_name=False,
    +            enabled=False,
    +            write_frequency="turn",
    +        )
    +
    +    def test_fresh_manager_after_flip_resolves_to_runtime(self):
    +        pinned_mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._pinned(),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(pinned_mgr)
    +        before = pinned_mgr.get_or_create("telegram:86701400")
    +        assert before.user_peer_id == "Igor"
    +
    +        unpinned_mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._unpinned(),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(unpinned_mgr)
    +        after = unpinned_mgr.get_or_create("telegram:86701400")
    +        assert after.user_peer_id == "86701400", (
    +            "After flipping pinPeerName off, the same runtime ID must resolve "
    +            "to its own peer — otherwise multi-user mode silently merges users."
    +        )
    +
    +    def test_cached_session_survives_config_flip_in_same_manager(self):
    +        mgr = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._pinned(),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(mgr)
    +        first = mgr.get_or_create("telegram:86701400")
    +        assert first.user_peer_id == "Igor"
    +
    +        mgr._config = self._unpinned()
    +        second = mgr.get_or_create("telegram:86701400")
    +        assert second.user_peer_id == "Igor", (
    +            "The per-key session cache is keyed by session-key, not by "
    +            "resolved peer.  In-process flips don't invalidate it — the "
    +            "gateway cache must bust the whole manager instead."
    +        )
    +
    +    def test_cache_busting_signature_reflects_pin_peer_name(self, tmp_path, monkeypatch):
    +        """Gateway agent cache must bust when honcho.json's pinPeerName flips."""
    +        from gateway.run import GatewayRunner
    +
    +        cfg_path = tmp_path / "honcho.json"
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +
    +        cfg_path.write_text(json.dumps({"apiKey": "k", "peerName": "Igor", "pinPeerName": True}))
    +        sig_pinned = GatewayRunner._extract_cache_busting_config({})
    +
    +        cfg_path.write_text(json.dumps({"apiKey": "k", "peerName": "Igor", "pinPeerName": False}))
    +        sig_unpinned = GatewayRunner._extract_cache_busting_config({})
    +
    +        assert sig_pinned["honcho.pin_peer_name"] != sig_unpinned["honcho.pin_peer_name"]
    +
    +    def test_cache_busting_signature_reflects_user_peer_aliases(self, tmp_path, monkeypatch):
    +        from gateway.run import GatewayRunner
    +
    +        cfg_path = tmp_path / "honcho.json"
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +
    +        cfg_path.write_text(json.dumps({"apiKey": "k", "peerName": "Igor"}))
    +        sig_no_aliases = GatewayRunner._extract_cache_busting_config({})
    +
    +        cfg_path.write_text(json.dumps({
    +            "apiKey": "k",
    +            "peerName": "Igor",
    +            "userPeerAliases": {"86701400": "Igor"},
    +        }))
    +        sig_with_aliases = GatewayRunner._extract_cache_busting_config({})
    +
    +        assert sig_no_aliases["honcho.user_peer_aliases"] != sig_with_aliases["honcho.user_peer_aliases"]
    +
    +    def test_cache_busting_signature_reflects_runtime_peer_prefix(self, tmp_path, monkeypatch):
    +        from gateway.run import GatewayRunner
    +
    +        cfg_path = tmp_path / "honcho.json"
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +
    +        cfg_path.write_text(json.dumps({"apiKey": "k", "peerName": "Igor"}))
    +        sig_no_prefix = GatewayRunner._extract_cache_busting_config({})
    +
    +        cfg_path.write_text(json.dumps({
    +            "apiKey": "k",
    +            "peerName": "Igor",
    +            "runtimePeerPrefix": "telegram_",
    +        }))
    +        sig_with_prefix = GatewayRunner._extract_cache_busting_config({})
    +
    +        assert sig_no_prefix["honcho.runtime_peer_prefix"] != sig_with_prefix["honcho.runtime_peer_prefix"]
    +
    +    def test_cache_busting_signature_reflects_ai_peer(self, tmp_path, monkeypatch):
    +        """Editing ``aiPeer`` mid-flight must invalidate the cached agent.
    +
    +        ``HonchoSessionManager`` freezes ``cfg.ai_peer`` at construction —
    +        without busting here, assistant writes keep landing on the old
    +        peer until an unrelated cache eviction.
    +        """
    +        from gateway.run import GatewayRunner
    +
    +        cfg_path = tmp_path / "honcho.json"
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +
    +        cfg_path.write_text(json.dumps({
    +            "apiKey": "k",
    +            "peerName": "Igor",
    +            "aiPeer": "hermes",
    +        }))
    +        sig_before = GatewayRunner._extract_cache_busting_config({})
    +
    +        cfg_path.write_text(json.dumps({
    +            "apiKey": "k",
    +            "peerName": "Igor",
    +            "aiPeer": "hermetika",
    +        }))
    +        sig_after = GatewayRunner._extract_cache_busting_config({})
    +
    +        assert sig_before["honcho.ai_peer"] != sig_after["honcho.ai_peer"]
    +
    +
    +class TestProfilePeerUniqueness:
    +    """Each Hermes profile can pin to its own unique peerName.
    +
    +    Profile cloning copies host blocks, but operators routinely diverge them
    +    afterwards (e.g. `hermes -p partner` pinned to a different person's peer).
    +    The resolver must honor host-level ``peerName`` so two profiles in the
    +    same workspace stay scoped to different Honcho peers.
    +    """
    +
    +    def _pinned_to(self, name: str) -> HonchoClientConfig:
    +        return HonchoClientConfig(
    +            api_key="k",
    +            peer_name=name,
    +            pin_peer_name=True,
    +            enabled=False,
    +            write_frequency="turn",
    +        )
    +
    +    def test_two_profiles_pinned_to_different_peer_names_resolve_distinctly(self):
    +        mgr_a = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._pinned_to("alice"),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(mgr_a)
    +        sess_a = mgr_a.get_or_create("telegram:86701400")
    +
    +        mgr_b = HonchoSessionManager(
    +            honcho=MagicMock(),
    +            config=self._pinned_to("bob"),
    +            runtime_user_peer_name="86701400",
    +        )
    +        _patch_manager_for_resolution_test(mgr_b)
    +        sess_b = mgr_b.get_or_create("telegram:86701400")
    +
    +        assert sess_a.user_peer_id == "alice"
    +        assert sess_b.user_peer_id == "bob"
    +        assert sess_a.user_peer_id != sess_b.user_peer_id, (
    +            "Profiles pinned to distinct peer names must not collapse to "
    +            "the same Honcho peer — otherwise profile isolation is fictional."
    +        )
    +
    +    def test_host_peer_name_overrides_root_when_pinned(self, tmp_path, monkeypatch):
    +        """Host-level peerName wins so each profile can pin uniquely while
    +        sharing a single root-level apiKey and workspace.
    +        """
    +        config_file = tmp_path / "honcho.json"
    +        config_file.write_text(json.dumps({
    +            "apiKey": "k",
    +            "peerName": "default-user",
    +            "hosts": {
    +                "hermes.partner": {
    +                    "peerName": "partner-user",
    +                    "pinPeerName": True,
    +                },
    +            },
    +        }))
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "isolated"))
    +
    +        cfg = HonchoClientConfig.from_global_config(
    +            host="hermes.partner", config_path=config_file,
    +        )
    +        assert cfg.peer_name == "partner-user"
    +        assert cfg.pin_peer_name is True

  • tests/honcho_plugin/test_session.py

    @@ -212,6 +212,39 @@ def test_get_peer_card_uses_direct_peer_lookup(self):
             assert mgr.get_peer_card(session.key) == ["Name: Robert"]
             assistant_peer.get_card.assert_called_once_with(target=session.user_peer_id)
     
    +    def test_get_peer_card_falls_back_to_target_peer_own_card(self):
    +        # When the observer-target card slot is empty (returns None/[]), fall
    +        # back to the target peer's own card. Self-hosted Honcho v3 stores the
    +        # peer card on the peer itself; the observer-target slot is only
    +        # populated when writes also go through that path.
    +        mgr, session = self._make_cached_manager()
    +        assistant_peer = MagicMock()
    +        assistant_peer.get_card.return_value = None  # observer-target slot empty
    +        user_peer = MagicMock()
    +        user_peer.get_card.return_value = ["Prefers: dark mode"]
    +
    +        def _peer(peer_id: str) -> MagicMock:
    +            return assistant_peer if peer_id == session.assistant_peer_id else user_peer
    +
    +        mgr._get_or_create_peer = MagicMock(side_effect=_peer)
    +
    +        assert mgr.get_peer_card(session.key) == ["Prefers: dark mode"]
    +        assistant_peer.get_card.assert_called_once_with(target=session.user_peer_id)
    +        user_peer.get_card.assert_called_once_with()
    +
    +    def test_set_peer_card_uses_observer_target_in_ai_observe_others_mode(self):
    +        # Writes must go to the same observer-target slot that reads check,
    +        # so that a subsequent honcho_profile read returns what was written.
    +        mgr, session = self._make_cached_manager()
    +        assistant_peer = MagicMock()
    +        assistant_peer.set_card.return_value = ["Role: user"]
    +        mgr._get_or_create_peer = MagicMock(return_value=assistant_peer)
    +
    +        result = mgr.set_peer_card(session.key, ["Role: user"])
    +
    +        assert result == ["Role: user"]
    +        assistant_peer.set_card.assert_called_once_with(["Role: user"], target=session.user_peer_id)
    +
         def test_search_context_uses_assistant_perspective_with_target(self):
             mgr, session = self._make_cached_manager()
             assistant_peer = MagicMock()
    @@ -573,7 +606,7 @@ class TestToolsModeInitBehavior:
         """Verify initOnSessionStart controls session init timing in tools mode."""
     
         def _make_provider_with_config(self, recall_mode="tools", init_on_session_start=False,
    -                                    peer_name=None, user_id=None):
    +                                    peer_name=None, user_id=None, user_id_alt=None):
             """Create a HonchoMemoryProvider with mocked config and dependencies."""
             from plugins.memory.honcho.client import HonchoClientConfig
     
    @@ -598,6 +631,8 @@ def _make_provider_with_config(self, recall_mode="tools", init_on_session_start=
             init_kwargs = {}
             if user_id:
                 init_kwargs["user_id"] = user_id
    +        if user_id_alt:
    +            init_kwargs["user_id_alt"] = user_id_alt
     
             with patch("plugins.memory.honcho.client.HonchoClientConfig.from_global_config", return_value=cfg), \
                  patch("plugins.memory.honcho.client.get_honcho_client", return_value=MagicMock()), \
    @@ -655,6 +690,15 @@ def test_user_id_used_when_no_peer_name(self):
             assert cfg.peer_name is None
             assert mock_manager_cls.call_args.kwargs["runtime_user_peer_name"] == "8439114563"
     
    +    def test_user_id_alt_is_passed_to_session_manager(self):
    +        """Gateway alternate user IDs are available for Honcho alias matching."""
    +        _, _, mock_manager_cls = self._make_provider_with_config(
    +            recall_mode="tools", init_on_session_start=True,
    +            peer_name=None, user_id="open-id", user_id_alt="union-id",
    +        )
    +        assert mock_manager_cls.call_args.kwargs["runtime_user_peer_name"] == "open-id"
    +        assert mock_manager_cls.call_args.kwargs["runtime_user_peer_name_alt"] == "union-id"
    +
     
     class TestPerSessionMigrateGuard:
         """Verify migrate_memory_files is skipped under per-session strategy.

  • tests/run_agent/test_memory_provider_init.py

    @@ -4,6 +4,27 @@
     from unittest.mock import patch
     
     
    +class RecordingMemoryProvider:
    +    name = "recording"
    +
    +    def __init__(self):
    +        self.init_kwargs = None
    +        self.init_session_id = None
    +
    +    def is_available(self):
    +        return True
    +
    +    def initialize(self, session_id, **kwargs):
    +        self.init_session_id = session_id
    +        self.init_kwargs = dict(kwargs)
    +
    +    def get_tool_schemas(self):
    +        return []
    +
    +    def shutdown(self):
    +        pass
    +
    +
     def test_blank_memory_provider_does_not_auto_enable_honcho():
         """Blank memory.provider should remain opt-out even if Honcho fallback looks configured."""
         cfg = {"memory": {"provider": ""}, "agent": {}}
    @@ -37,3 +58,35 @@ def test_blank_memory_provider_does_not_auto_enable_honcho():
         load_memory_provider.assert_not_called()
         save_config.assert_not_called()
     
    +
    +def test_aiagent_forwards_user_id_alt_to_memory_provider():
    +    provider = RecordingMemoryProvider()
    +    cfg = {"memory": {"provider": "recording"}, "agent": {}}
    +
    +    with (
    +        patch("hermes_cli.config.load_config", return_value=cfg),
    +        patch("plugins.memory.load_memory_provider", return_value=provider),
    +        patch("agent.model_metadata.get_model_context_length", return_value=204_800),
    +        patch("run_agent.get_tool_definitions", return_value=[]),
    +        patch("run_agent.check_toolset_requirements", return_value={}),
    +        patch("run_agent.OpenAI"),
    +    ):
    +        from run_agent import AIAgent
    +
    +        agent = AIAgent(
    +            api_key="[redacted:secret-assignment]",
    +            base_url="https://openrouter.ai/api/v1",
    +            quiet_mode=True,
    +            skip_context_files=True,
    +            skip_memory=False,
    +            session_id="sess-alt",
    +            platform="feishu",
    +            user_id="open-id",
    +            user_id_alt="union-id",
    +        )
    +
    +    assert agent._memory_manager is not None
    +    assert provider.init_session_id == "sess-alt"
    +    assert provider.init_kwargs["user_id"] == "open-id"
    +    assert provider.init_kwargs["user_id_alt"] == "union-id"
    +    assert provider.init_kwargs["platform"] == "feishu"

  • tests/test_honcho_session_context.py

    @@ -0,0 +1,95 @@
    +"""Tests for Honcho session context peer resolution."""
    +
    +from types import SimpleNamespace
    +
    +from plugins.memory.honcho.session import HonchoSession, HonchoSessionManager
    +
    +
    +class _FakeSummary:
    +    content = "summary"
    +
    +
    +class _FakeContext:
    +    summary = _FakeSummary()
    +    peer_representation = "representation"
    +    peer_card = ["fact"]
    +    messages = []
    +
    +
    +class _RecordingHonchoSession:
    +    def __init__(self):
    +        self.calls = []
    +
    +    def context(self, **kwargs):
    +        self.calls.append(kwargs)
    +        return _FakeContext()
    +
    +
    +def _manager_with_cached_session(*, ai_observe_others=True):
    +    cfg = SimpleNamespace(
    +        write_frequency="turn",
    +        dialectic_reasoning_level="low",
    +        dialectic_dynamic=True,
    +        dialectic_max_chars=600,
    +        observation_mode="directional",
    +        user_observe_me=True,
    +        user_observe_others=True,
    +        ai_observe_me=True,
    +        ai_observe_others=ai_observe_others,
    +        message_max_chars=25000,
    +        dialectic_max_input_chars=10000,
    +    )
    +    mgr = HonchoSessionManager(honcho=SimpleNamespace(), config=cfg)
    +    session = HonchoSession(
    +        key="test-session",
    +        user_peer_id="chris",
    +        assistant_peer_id="hermes",
    +        honcho_session_id="test-session",
    +    )
    +    fake_honcho_session = _RecordingHonchoSession()
    +    mgr._cache[session.key] = session
    +    mgr._sessions_cache[session.honcho_session_id] = fake_honcho_session
    +    return mgr, fake_honcho_session
    +
    +
    +def test_session_context_user_alias_uses_assistant_observer_when_ai_can_observe_others():
    +    mgr, fake = _manager_with_cached_session(ai_observe_others=True)
    +
    +    result = mgr.get_session_context("test-session", peer="user")
    +
    +    assert result["summary"] == "summary"
    +    assert fake.calls == [
    +        {
    +            "summary": True,
    +            "peer_target": "chris",
    +            "peer_perspective": "hermes",
    +        }
    +    ]
    +
    +
    +def test_session_context_explicit_user_peer_matches_user_alias():
    +    mgr, fake = _manager_with_cached_session(ai_observe_others=True)
    +
    +    mgr.get_session_context("test-session", peer="chris")
    +
    +    assert fake.calls == [
    +        {
    +            "summary": True,
    +            "peer_target": "chris",
    +            "peer_perspective": "hermes",
    +        }
    +    ]
    +
    +
    +def test_session_context_user_alias_uses_user_self_observer_when_ai_cannot_observe_others():
    +    mgr, fake = _manager_with_cached_session(ai_observe_others=False)
    +
    +    mgr.get_session_context("test-session", peer="user")
    +
    +    assert fake.calls == [
    +        {
    +            "summary": True,
    +            "peer_target": "chris",
    +            "peer_perspective": "chris",
    +        }
    +    ]