From 2b47a5e024be9245243f479a16222b0e206bb16d Mon Sep 17 00:00:00 2001 From: handsdiff <239876380+handsdiff@users.noreply.github.com> Date: Mon, 13 Apr 2026 20:52:55 -0400 Subject: [PATCH 1/3] feat(memory): give cron jobs per-job Honcho peers instead of skipping memory MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cron-triggered agents had zero memory — skip_memory=True meant Honcho and Hindsight were both disabled, so self-reflection crons couldn't read prior self-reflections and hub-discovery crons couldn't remember what they'd already found. Route each cron job to its own Honcho peer (cron-{job_name}) so Honcho builds a coherent representation per scheduled behavior. The assistant peer still accumulates all actions into the agent's unified self-model. Changes: - Remove skip_memory=True from cron AIAgent construction, pass user_id="cron-{slug(name)}" to route to per-job peers - Add agent.shutdown_memory_provider() to cron teardown so final sync_turn/aretain_batch writes aren't dropped - Delete the _cron_skipped guard from HonchoMemoryProvider (flag init, cron/flush check in initialize(), and 9 early-return checks) - Add skip_memory=True to the hygiene compress agent, which was accidentally protected by the cron guard but should never write to memory providers - Revert #6995 peerName guard (not cfg.peer_name) that blocked ALL user_id overrides on provisioned agents — cron peers, stranger isolation, and future owner-unification all depend on the override being unconditional Co-Authored-By: Claude Opus 4.6 (1M context) --- cron/scheduler.py | 21 ++++++++++++++-- plugins/memory/honcho/__init__.py | 42 ++++++++----------------------- 2 files changed, 30 insertions(+), 33 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index d051a7ab36ed9..f45cd2abfdf28 100644 --- a/cron/scheduler.py +++ b/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: diff --git a/plugins/memory/honcho/__init__.py b/plugins/memory/honcho/__init__.py index 6ca32c1dcbb5c..dc2ba83924899 100644 --- a/plugins/memory/honcho/__init__.py +++ b/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,6 +282,16 @@ 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 # ----- B1: recall_mode from config ----- @@ -442,8 +441,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 @@ -497,8 +494,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 +546,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 "" @@ -702,8 +694,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 +1052,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 +1079,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 +1093,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 +1108,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: From e6f978ebda79e088d543a1afb27a9c5dbae91fe3 Mon Sep 17 00:00:00 2001 From: handsdiff <239876380+handsdiff@users.noreply.github.com> Date: Mon, 13 Apr 2026 21:59:03 -0400 Subject: [PATCH 2/3] feat(memory): unify owner identity across channels in Honcho memory MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Owner messages from different channels (Telegram, Discord, CLI, etc.) were each creating separate Honcho peers, fragmenting the owner's representation. The agent couldn't remember cross-channel conversations. Add gateway-side owner detection so the configured peerName ("user") survives for owner sources while strangers still get transport-level peer isolation: - Add _is_owner_source helper with strict DM-only check, untrusted platform exclusion (webhook, API server), and chat_id match against home channels - At both memory-participating AIAgent construction sites, pass user_id=None for owner sources so the honcho plugin's unconditional override doesn't fire, letting peerName: "user" survive - Set chat_type="synthetic" on the background process notification's synthetic SessionSource to prevent false-positive owner detection - Thread legacy_peer_ids through AIAgent → memory init → honcho plugin for dual-read of pre-unification transport-level peer representations - Extend get_prefetch_context to fetch legacy peer contexts and merge into first-turn context as "Prior Channel History" so existing owner history isn't orphaned after unification Co-Authored-By: Claude Opus 4.6 (1M context) --- gateway/run.py | 67 ++++++++++++++++++++++++++++++- plugins/memory/honcho/__init__.py | 12 +++++- plugins/memory/honcho/session.py | 29 ++++++++++++- run_agent.py | 4 ++ 4 files changed, 107 insertions(+), 5 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index db3f8b00d5ed3..c42b715dba538 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1047,6 +1047,50 @@ 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, + }) + + 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 + if source.platform in self._UNTRUSTED_IDENTITY_PLATFORMS: + 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 +6594,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 +6619,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 +8620,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 +9862,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 +9922,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, ) diff --git a/plugins/memory/honcho/__init__.py b/plugins/memory/honcho/__init__.py index dc2ba83924899..05fb52d053997 100644 --- a/plugins/memory/honcho/__init__.py +++ b/plugins/memory/honcho/__init__.py @@ -293,6 +293,7 @@ def initialize(self, session_id: str, **kwargs) -> None: 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" @@ -471,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}") @@ -568,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: diff --git a/plugins/memory/honcho/session.py b/plugins/memory/honcho/session.py index 79625b5cd5800..1596bc6189071 100644 --- a/plugins/memory/honcho/session.py +++ b/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: diff --git a/run_agent.py b/run_agent.py index affcbbd7218dc..333ca9f498f9a 100644 --- a/run_agent.py +++ b/run_agent.py @@ -747,6 +747,7 @@ def __init__( chat_type: str = None, thread_id: str = None, gateway_session_key: str = None, + legacy_peer_ids: list[str] = None, skip_context_files: bool = False, skip_memory: bool = False, session_db=None, @@ -821,6 +822,7 @@ def __init__( self._chat_type = chat_type self._thread_id = thread_id self._gateway_session_key = gateway_session_key # Stable per-chat key (e.g. agent:main:telegram:dm:123) + self._legacy_peer_ids = legacy_peer_ids or [] # Historical transport-level peer IDs for owner dual-read # Pluggable print function — CLI replaces this with _cprint so that # raw ANSI status lines are routed through prompt_toolkit's renderer # instead of going directly to stdout where patch_stdout's StdoutProxy @@ -1495,6 +1497,8 @@ def __init__( # Thread gateway session key for stable per-chat Honcho session isolation if self._gateway_session_key: _init_kwargs["gateway_session_key"] = self._gateway_session_key + if self._legacy_peer_ids: + _init_kwargs["legacy_peer_ids"] = self._legacy_peer_ids # Profile identity for per-profile provider scoping try: from hermes_cli.profiles import get_active_profile_name From a0b88ad72b09fb67515679c6e6f641a0af1adef3 Mon Sep 17 00:00:00 2001 From: handsdiff <239876380+handsdiff@users.noreply.github.com> Date: Thu, 23 Apr 2026 23:33:57 -0400 Subject: [PATCH 3/3] fix: tolerate non-enum owner source platforms --- gateway/run.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/gateway/run.py b/gateway/run.py index c42b715dba538..4face94d445cc 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1054,6 +1054,10 @@ def _session_key_for_source(self, source: SessionSource) -> str: 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. @@ -1068,7 +1072,8 @@ def _is_owner_source(self, source) -> bool: """ if not source or not source.platform: return False - if source.platform in self._UNTRUSTED_IDENTITY_PLATFORMS: + 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