feat(memory): unify owner identity across channels in Honcho

cron/scheduler.py

@@ -14,6 +14,7 @@
 import json
 import logging
 import os
+import re
 import subprocess
 import sys
 
@@ -730,6 +731,11 @@ def _build_job_prompt(job: dict, prerun_script: Optional[tuple] = None) -> str:
     return "\n".join(parts)
 
 
+def _slug(raw: str) -> str:
+    """Lowercase, strip non-[a-z0-9-], collapse runs of hyphens."""
+    return re.sub(r'-+', '-', re.sub(r'[^a-z0-9-]', '-', raw.lower())).strip('-')
+
+
 def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
     """
     Execute a single cron job.
@@ -921,7 +927,11 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
             disabled_toolsets=["cronjob", "messaging", "clarify"],
             quiet_mode=True,
             skip_context_files=True,  # Don't inject SOUL.md/AGENTS.md from scheduler cwd
-            skip_memory=True,  # Cron system prompts would corrupt user representations
+            # Each cron job gets its own Honcho peer (cron-{name}) so Honcho builds a
+            # coherent representation of what each scheduled behavior does over time. The
+            # assistant's own responses still attribute to the aiPeer, so the agent's
+            # self-model accumulates cron actions as its own.
+            user_id=f"cron-{_slug(job.get('name') or job['id'][:8])}",
             platform="cron",
             session_id=_cron_session_id,
             session_db=_session_db,
@@ -1059,7 +1069,14 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
         return False, output, "", error_msg
 
     finally:
-        # Clean up ContextVar session/delivery state for this job.
+        # Flush memory providers so the last sync_turn / aretain_batch aren't
+        # dropped when the thread pool shuts down.
+        try:
+            agent.shutdown_memory_provider()
+        except Exception as e:
+            logger.debug("Job '%s': memory provider shutdown failed: %s", job_id, e)
+        # Clean up ContextVar session/delivery state for this job. Upstream
+        # consolidated the manual env-var unset into clear_session_vars().
         clear_session_vars(_ctx_tokens)
         if _session_db:
             try:

gateway/run.py

@@ -1047,6 +1047,55 @@ def _session_key_for_source(self, source: SessionSource) -> str:
             thread_sessions_per_user=getattr(config, "thread_sessions_per_user", False),
         )
 
+    # Platforms where identity is not enforced by the transport layer.
+    # Owner detection is skipped for these to avoid false positives from
+    # spoofed user_ids.
+    _UNTRUSTED_IDENTITY_PLATFORMS = frozenset({
+        Platform.WEBHOOK,
+        Platform.API_SERVER,
+    })
+    _UNTRUSTED_IDENTITY_PLATFORM_VALUES = frozenset(
+        str(getattr(platform, "value", platform)).lower()
+        for platform in _UNTRUSTED_IDENTITY_PLATFORMS
+    )
+
+    def _is_owner_source(self, source) -> bool:
+        """Return True if this inbound source is the provisioned owner.
+
+        Strict check: only recognizes DM-shaped home channels where the
+        chat_id matches, the platform enforces identity, and chat_type was
+        explicitly set to "dm". Group-chat home channels and untrusted
+        platforms fall through to False.
+
+        On False, callers should continue to pass source.user_id to AIAgent
+        so non-owner callers land on their own transport-level peer.
+        """
+        if not source or not source.platform:
+            return False
+        platform_value = str(getattr(source.platform, "value", source.platform)).lower()
+        if platform_value in self._UNTRUSTED_IDENTITY_PLATFORM_VALUES:
+            return False
+        # Defensive: test fixtures sometimes skip self.config entirely or
+        # stub it with a SimpleNamespace that lacks get_home_channel. Treat
+        # missing config surface as "no home channel known → not the owner".
+        _cfg = getattr(self, "config", None)
+        _get_hc = getattr(_cfg, "get_home_channel", None) if _cfg is not None else None
+        if not callable(_get_hc):
+            return False
+        hc = _get_hc(source.platform)
+        if not hc:
+            return False
+        if str(source.chat_id) != str(hc.chat_id):
+            return False
+        # Strict DM check. SessionSource.chat_type defaults to "dm"
+        # (gateway/session.py:78), so an adapter that forgets to set it for a
+        # group chat would pass this check — a false-positive vector. Adapters
+        # MUST set chat_type explicitly for non-DM sources. This is already
+        # the case for all shipped adapters (Discord, Telegram, Signal, etc.).
+        if getattr(source, "chat_type", None) != "dm":
+            return False
+        return True
+
     def _resolve_session_agent_runtime(
         self,
         *,
@@ -6550,6 +6599,12 @@ async def _run_background_task(
             self._service_tier = self._load_service_tier()
             turn_route = self._resolve_turn_agent_config(prompt, model, runtime_kwargs)
 
+            _is_owner = self._is_owner_source(source)
+            _legacy_peers = (
+                [str(hc.chat_id) for p in Platform
+                 if (hc := self.config.get_home_channel(p))]
+            ) if _is_owner else []
+
             def run_sync():
                 agent = AIAgent(
                     model=turn_route["model"],
@@ -6569,12 +6624,13 @@ def run_sync():
                     provider_data_collection=pr.get("data_collection"),
                     session_id=task_id,
                     platform=platform_key,
-                    user_id=source.user_id,
+                    user_id=None if _is_owner else source.user_id,
                     user_name=source.user_name,
                     chat_id=source.chat_id,
                     chat_name=source.chat_name,
                     chat_type=source.chat_type,
                     thread_id=source.thread_id,
+                    legacy_peer_ids=_legacy_peers,
                     session_db=self._session_db,
                     fallback_model=self._fallback_model,
                 )
@@ -8569,6 +8625,11 @@ async def _run_process_watcher(self, watcher: dict) -> None:
                             break
                     if adapter and source.chat_id:
                         try:
+                            import dataclasses as _dc
+                            from gateway.platforms.base import MessageEvent, MessageType
+                            # Force chat_type="synthetic" so owner-detection doesn't
+                            # treat the synthetic bg notification as a real DM turn.
+                            source = _dc.replace(source, chat_type="synthetic")
                             synth_event = MessageEvent(
                                 text=synth_text,
                                 message_type=MessageType.TEXT,
@@ -9806,6 +9867,12 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
 
             turn_route = self._resolve_turn_agent_config(message, model, runtime_kwargs)
 
+            _is_owner = self._is_owner_source(source)
+            _legacy_peers = (
+                [str(hc.chat_id) for p in Platform
+                 if (hc := self.config.get_home_channel(p))]
+            ) if _is_owner else []
+
             # Check agent cache — reuse the AIAgent from the previous message
             # in this session to preserve the frozen system prompt and tool
             # schemas for prompt cache hits.
@@ -9860,13 +9927,14 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
                     provider_data_collection=pr.get("data_collection"),
                     session_id=session_id,
                     platform=platform_key,
-                    user_id=source.user_id,
+                    user_id=None if _is_owner else source.user_id,
                     user_name=source.user_name,
                     chat_id=source.chat_id,
                     chat_name=source.chat_name,
                     chat_type=source.chat_type,
                     thread_id=source.thread_id,
                     gateway_session_key=session_key,
+                    legacy_peer_ids=_legacy_peers,
                     session_db=self._session_db,
                     fallback_model=self._fallback_model,
                 )

plugins/memory/honcho/__init__.py

@@ -225,8 +225,6 @@ def __init__(self):
         self._lazy_init_kwargs: Optional[dict] = None
         self._lazy_init_session_id: Optional[str] = None
 
-        # Port #4053: cron guard — when True, plugin is fully inactive
-        self._cron_skipped = False
 
     @property
     def name(self) -> str:
@@ -271,20 +269,11 @@ def post_setup(self, hermes_home: str, config: dict) -> None:
     def initialize(self, session_id: str, **kwargs) -> None:
         """Initialize Honcho session manager.
 
-        Handles: cron guard, recall_mode, session name resolution,
+        Handles: recall_mode, session name resolution,
         peer memory mode, SOUL.md ai_peer sync, memory file migration,
         and pre-warming context at init.
         """
         try:
-            # ----- Port #4053: cron guard -----
-            agent_context = kwargs.get("agent_context", "")
-            platform = kwargs.get("platform", "cli")
-            if agent_context in ("cron", "flush") or platform == "cron":
-                logger.debug("Honcho skipped: cron/flush context (agent_context=%s, platform=%s)",
-                             agent_context, platform)
-                self._cron_skipped = True
-                return
-
             from plugins.memory.honcho.client import HonchoClientConfig, get_honcho_client
             from plugins.memory.honcho.session import HonchoSessionManager
 
@@ -293,7 +282,18 @@ def initialize(self, session_id: str, **kwargs) -> None:
                 logger.debug("Honcho not configured — plugin inactive")
                 return
 
+            # Override peer_name with the caller-supplied user_id so each caller
+            # gets their own Honcho peer.  For owner sessions the gateway will
+            # stop passing user_id once user-unify ships — until then, owner
+            # messages fragment across transport-level peers (pre-#6995 behavior,
+            # preferable to the active stranger/cron pollution #6995 introduced).
+            _gw_user_id = kwargs.get("user_id")
+            if _gw_user_id:
+                cfg.peer_name = _gw_user_id
+
+
             self._config = cfg
+            self._legacy_peer_ids = kwargs.get("legacy_peer_ids") or []
 
             # ----- B1: recall_mode from config -----
             self._recall_mode = cfg.recall_mode  # "context", "tools", or "hybrid"
@@ -442,8 +442,6 @@ def _ensure_session(self) -> bool:
         """
         if self._manager and self._session_initialized:
             return True
-        if self._cron_skipped:
-            return False
         if not self._config or not self._lazy_init_kwargs:
             return False
 
@@ -474,6 +472,12 @@ def _format_first_turn_context(self, ctx: dict) -> str:
         if rep:
             parts.append(f"## User Representation\n{rep}")
 
+        # Legacy representations from pre-unification transport-level peers.
+        # Merged alongside the primary so the model sees full owner history.
+        legacy = ctx.get("legacy_representations", "")
+        if legacy:
+            parts.append(f"## Prior Channel History\n{legacy}")
+
         card = ctx.get("card", "")
         if card:
             parts.append(f"## User Peer Card\n{card}")
@@ -497,8 +501,6 @@ def system_prompt_block(self) -> str:
         that doesn't change between turns (prompt-cache friendly).
         Live context (representation, card) is injected via prefetch().
         """
-        if self._cron_skipped:
-            return ""
         if not self._manager or not self._session_key:
             # tools-only mode without session yet still returns a minimal block
             if self._recall_mode == "tools" and self._config:
@@ -551,9 +553,6 @@ def prefetch(self, query: str, *, session_id: str = "") -> str:
         B5: Respects injection_frequency — "first-turn" returns cached/empty after turn 0.
         Port #3265: Truncates to context_tokens budget.
         """
-        if self._cron_skipped:
-            return ""
-
         # B1: tools-only mode — no auto-injection
         if self._recall_mode == "tools":
             return ""
@@ -576,7 +575,10 @@ def prefetch(self, query: str, *, session_id: str = "") -> str:
             if self._base_context_cache is None:
                 # First call — synchronous fetch
                 try:
-                    ctx = self._manager.get_prefetch_context(self._session_key)
+                    ctx = self._manager.get_prefetch_context(
+                        self._session_key,
+                        legacy_peer_ids=self._legacy_peer_ids,
+                    )
                     self._base_context_cache = self._format_first_turn_context(ctx) if ctx else ""
                     self._last_context_turn = self._turn_count
                 except Exception as e:
@@ -702,8 +704,6 @@ def queue_prefetch(self, query: str, *, session_id: str = "") -> None:
         Context refresh updates the base layer (representation + card).
         Dialectic fires the LLM reasoning supplement.
         """
-        if self._cron_skipped:
-            return
         if not self._manager or not self._session_key or not query:
             return
 
@@ -1062,8 +1062,6 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
         Messages exceeding the Honcho API limit (default 25k chars) are
         split into multiple messages with continuation markers.
         """
-        if self._cron_skipped:
-            return
         if not self._manager or not self._session_key:
             return
 
@@ -1091,8 +1089,6 @@ def on_memory_write(self, action: str, target: str, content: str) -> None:
         """Mirror built-in user profile writes as Honcho conclusions."""
         if action != "add" or target != "user" or not content:
             return
-        if self._cron_skipped:
-            return
         if not self._manager or not self._session_key:
             return
 
@@ -1107,8 +1103,6 @@ def _write():
 
     def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
         """Flush all pending messages to Honcho on session end."""
-        if self._cron_skipped:
-            return
         if not self._manager:
             return
         # Wait for pending sync
@@ -1124,16 +1118,12 @@ def get_tool_schemas(self) -> List[Dict[str, Any]]:
 
         B1: context-only mode hides all tools.
         """
-        if self._cron_skipped:
-            return []
         if self._recall_mode == "context":
             return []
         return list(ALL_TOOL_SCHEMAS)
 
     def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str:
         """Handle a Honcho tool call, with lazy session init for tools-only mode."""
-        if self._cron_skipped:
-            return tool_error("Honcho is not active (cron context).")
 
         # Port #1957: ensure session is initialized for tools-only mode
         if not self._session_initialized:

plugins/memory/honcho/session.py

@@ -593,7 +593,12 @@ def pop_context_result(self, session_key: str) -> dict[str, str]:
         with self._prefetch_cache_lock:
             return self._context_cache.pop(session_key, {})
 
-    def get_prefetch_context(self, session_key: str, user_message: str | None = None) -> dict[str, str]:
+    def get_prefetch_context(
+        self,
+        session_key: str,
+        user_message: str | None = None,
+        legacy_peer_ids: list[str] | None = None,
+    ) -> dict[str, str]:
         """
         Pre-fetch user and AI peer context from Honcho.
 
@@ -603,13 +608,18 @@ def get_prefetch_context(self, session_key: str, user_message: str | None = None
         consume, and passing the raw message exposes conversation content in
         server access logs.
 
+        When legacy_peer_ids is provided (owner unification), also fetches
+        context from historical transport-level peers and merges into
+        'legacy_representations'.
+
         Args:
             session_key: The session key to get context for.
             user_message: Unused; kept for call-site compatibility.
+            legacy_peer_ids: Historical transport-level peer IDs for dual-read.
 
         Returns:
             Dictionary with 'representation', 'card', 'ai_representation',
-            'ai_card', and optionally 'summary' keys.
+            'ai_card', and optionally 'summary' and 'legacy_representations' keys.
         """
         session = self._cache.get(session_key)
         if not session:
@@ -645,6 +655,21 @@ def get_prefetch_context(self, session_key: str, user_message: str | None = None
         except Exception as e:
             logger.debug("Failed to fetch AI peer context from Honcho: %s", e)
 
+        # Dual-read: fetch legacy transport-level peer representations so
+        # owner history from before unification is visible in context.
+        if legacy_peer_ids:
+            legacy_reps = []
+            for peer_id in legacy_peer_ids:
+                try:
+                    legacy_ctx = self._fetch_peer_context(peer_id)
+                    rep = legacy_ctx.get("representation", "")
+                    if rep:
+                        legacy_reps.append(f"[from {peer_id}]\n{rep}")
+                except Exception as e:
+                    logger.debug("Legacy peer '%s' context fetch failed: %s", peer_id, e)
+            if legacy_reps:
+                result["legacy_representations"] = "\n\n".join(legacy_reps)
+
         return result
 
     def migrate_local_history(self, session_key: str, messages: list[dict[str, Any]]) -> bool: