diff --git a/gateway/run.py b/gateway/run.py index d8a28f491a2a..da87018ce831 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1207,6 +1207,13 @@ def __init__(self, config: Optional[GatewayConfig] = None): # Key: session_key, Value: AIAgent instance self._running_agents: Dict[str, Any] = {} self._running_agents_ts: Dict[str, float] = {} # start timestamp per session + # Per-origin conversation locks serialize every transcript-mutating + # conversation event for a chat/session. Native platform turns already + # have adapter-level active-session guards; this lock extends the same + # ordering to out-of-band events such as Kanban completion synthesis so + # they cannot run_conversation()/mirror into the same session in + # parallel with a user turn. + self._conversation_locks: Dict[str, asyncio.Lock] = {} self._pending_messages: Dict[str, str] = {} # Queued messages during interrupt # Overflow buffer for explicit /queue commands. The adapter-level # _pending_messages dict is a single slot per session (designed for @@ -1993,6 +2000,27 @@ def _queue_during_drain_enabled(self) -> bool: # process to pick up. "interrupt" mode drops them (current behaviour). return self._restart_requested and self._busy_input_mode in ("queue", "steer") + def _conversation_lock_for_key(self, session_key: str) -> asyncio.Lock: + """Return the per-session conversation serialization lock. + + Native user turns, queued follow-ups, and synthetic completion replies + all acquire this lock before mutating session history or mirroring an + assistant message. Locks are per session key, so unrelated chats keep + running concurrently. + """ + locks = getattr(self, "_conversation_locks", None) + if locks is None: + locks = {} + self._conversation_locks = locks + lock = locks.get(session_key) + if lock is None: + lock = asyncio.Lock() + locks[session_key] = lock + return lock + + def _conversation_lock_for_source(self, source: "SessionSource") -> asyncio.Lock: + return self._conversation_lock_for_key(self._session_key_for_source(source)) + # -------- /queue FIFO helpers -------------------------------------- # /queue must produce one full agent turn per invocation, in FIFO # order, with no merging. The adapter's _pending_messages dict is a @@ -3825,6 +3853,34 @@ async def _session_expiry_watcher(self, interval: int = 300): break await asyncio.sleep(1) + def _kanban_notify_in_gateway_enabled(self) -> bool: + """Return whether this gateway should consume Kanban notify rows. + + Multiple profile gateways can be alive on the same host. Notification + rows are a shared queue; whichever watcher consumes a row deletes it. + A secondary profile with its own bot can therefore steal the default + profile's subscription and fail delivery to an unknown chat. + + Default ownership follows ``kanban.dispatch_in_gateway`` because the + dispatcher-owning gateway is the board owner in the current deployment + model. Operators that run an external dispatcher but still want gateway + notifications can explicitly set ``kanban.notify_in_gateway: true``. + """ + env_override = os.environ.get("HERMES_KANBAN_NOTIFY_IN_GATEWAY", "").strip().lower() + if env_override in ("0", "false", "no", "off"): + return False + if env_override in ("1", "true", "yes", "on"): + return True + try: + from hermes_cli.config import load_config as _load_config + cfg = _load_config() + except Exception: + return False + kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {} + if "notify_in_gateway" in kanban_cfg: + return bool(kanban_cfg.get("notify_in_gateway")) + return bool(kanban_cfg.get("dispatch_in_gateway", True)) + async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None: """Poll ``kanban_notify_subs`` and deliver terminal events to users. @@ -3845,6 +3901,12 @@ async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None: purely a fan-out of the single-DB poll. """ from gateway.config import Platform as _Platform + if not self._kanban_notify_in_gateway_enabled(): + logger.info( + "kanban notifier: disabled via config/env " + "(kanban.notify_in_gateway=false or dispatch_in_gateway=false)" + ) + return try: from hermes_cli import kanban_db as _kb except Exception: @@ -3944,6 +4006,9 @@ def _collect(): # chat subscribes to many tasks) legible at a glance. who = (task.assignee if task and task.assignee else None) tag = f"@{who} " if who else "" + public_mode = str( + sub.get("notification_mode") or "direct" + ).strip().lower() == "synthesize" if kind == "completed": # Prefer the run's summary (the worker's # intentional human-facing handoff, carried @@ -3967,29 +4032,49 @@ def _collect(): elif kind == "blocked": reason = "" if ev.payload and ev.payload.get("reason"): - reason = f": {str(ev.payload['reason'])[:160]}" - msg = f"⏸ {tag}Kanban {sub['task_id']} blocked{reason}" + reason = str(ev.payload["reason"])[:160] + if public_mode: + msg = ( + f"I need one clarification before I can continue: {reason}" + if reason else + "I need one clarification before I can continue." + ) + else: + suffix = f": {reason}" if reason else "" + msg = f"⏸ {tag}Kanban {sub['task_id']} blocked{suffix}" elif kind == "gave_up": err = "" if ev.payload and ev.payload.get("error"): err = f"\n{str(ev.payload['error'])[:200]}" - msg = ( - f"✖ {tag}Kanban {sub['task_id']} gave up " - f"after repeated spawn failures{err}" - ) + if public_mode: + msg = ( + "I hit a backend issue and couldn’t finish this after retries. " + "Ask for internal run details if you want me to inspect the failure." + ) + else: + msg = ( + f"✖ {tag}Kanban {sub['task_id']} gave up " + f"after repeated spawn failures{err}" + ) elif kind == "crashed": - msg = ( - f"✖ {tag}Kanban {sub['task_id']} worker crashed " - f"(pid gone); dispatcher will retry" - ) + if public_mode: + msg = "I hit a backend issue while working on this. I’ll retry automatically." + else: + msg = ( + f"✖ {tag}Kanban {sub['task_id']} worker crashed " + f"(pid gone); dispatcher will retry" + ) elif kind == "timed_out": limit = 0 if ev.payload and ev.payload.get("limit_seconds"): limit = int(ev.payload["limit_seconds"]) - msg = ( - f"⏱ {tag}Kanban {sub['task_id']} timed out " - f"(max_runtime={limit}s); will retry" - ) + if public_mode: + msg = "This is taking longer than expected, so I’m retrying it." + else: + msg = ( + f"⏱ {tag}Kanban {sub['task_id']} timed out " + f"(max_runtime={limit}s); will retry" + ) else: continue metadata: dict[str, Any] = {} @@ -4000,8 +4085,9 @@ def _collect(): sub["chat_id"], sub.get("thread_id") or "", ) try: - await adapter.send( - sub["chat_id"], msg, metadata=metadata, + await self._send_kanban_notification( + adapter, sub, msg, metadata, + event=ev, task=task, board=board_slug, ) # Reset the failure counter on success. sub_fail_counts.pop(sub_key, None) @@ -4049,6 +4135,232 @@ def _collect(): return await asyncio.sleep(1) + async def _send_kanban_notification( + self, + adapter: Any, + sub: dict, + msg: str, + metadata: dict[str, Any], + *, + event: Any = None, + task: Any = None, + board: Optional[str] = None, + ) -> None: + """Send one Kanban notification and raise if the adapter reports failure. + + Platform adapters normally return ``SendResult`` instead of raising on + delivery failure. The notifier must treat ``success=False`` as failure; + otherwise it advances/unsubscribes and silently loses the completion + ping even though nothing reached the user. ``notification_mode`` can + opt into synthesized user-facing text or suppress delivery entirely. + """ + mode = str(sub.get("notification_mode") or "direct").strip().lower() + if mode == "silent": + logger.info( + "kanban notifier: silent notification for %s on %s:%s", + sub["task_id"], sub["platform"], sub["chat_id"], + ) + return + send_msg = msg + + async def _send_and_mirror() -> None: + nonlocal send_msg + if mode == "synthesize" and getattr(event, "kind", None) == "completed": + try: + synthesized = await self._synthesize_kanban_notification( + sub=sub, + event=event, + task=task, + board=board, + direct_message=msg, + ) + if synthesized and str(synthesized).strip(): + send_msg = str(synthesized).strip() + except Exception as exc: + logger.warning( + "kanban notifier: synthesis failed for %s; falling back direct: %s", + sub["task_id"], exc, + ) + send_msg = msg + result = await adapter.send(sub["chat_id"], send_msg, metadata=metadata) + if getattr(result, "success", True) is False: + error = getattr(result, "error", None) or "adapter returned success=False" + raise RuntimeError(str(error)) + + try: + from gateway.mirror import mirror_to_session + + mirrored = mirror_to_session( + str(sub.get("platform") or ""), + str(sub.get("chat_id") or ""), + send_msg, + source_label="kanban", + thread_id=(str(sub.get("thread_id") or "") or None), + user_id=(str(sub.get("user_id") or "") or None), + ) + except Exception: + mirrored = False + + logger.info( + "kanban notifier: sent %s event to %s:%s message_id=%s mirrored=%s", + sub["task_id"], sub["platform"], sub["chat_id"], + getattr(result, "message_id", None), mirrored, + ) + + try: + platform_str = str(sub.get("platform") or "telegram").lower() + platform = Platform(platform_str) + source = SessionSource( + platform=platform, + chat_id=str(sub.get("chat_id") or ""), + user_id=str(sub.get("user_id") or "") or None, + thread_id=str(sub.get("thread_id") or "") or None, + ) + lock = self._conversation_lock_for_source(source) + except Exception: + lock = None + + if lock is None: + await _send_and_mirror() + else: + async with lock: + await _send_and_mirror() + + @staticmethod + def _build_kanban_synthesis_prompt( + *, + sub: dict, + event: Any, + task: Any, + board: Optional[str], + worker_summary: Any, + worker_metadata: Any, + direct_message: str, + ) -> str: + """Build the user-facing completion synthesis prompt.""" + origin_context = str(sub.get("origin_context") or "").strip() + return ( + "You are the origin/default Hermes profile writing the final user-facing reply.\n" + "Use only the provided handoff, metadata, task title/body, and origin context.\n" + "Do NOT browse, fetch, run commands, create tasks, or redo data collection.\n" + "Default UX rule: hide internal workflow plumbing. Do NOT mention Kanban, task ids, assignees, " + "workers, worker/dispatcher flow, notification subscriptions, run ids, process ids, or board names " + "unless the user explicitly asked for internal mechanics, debugging, task status, or audit details.\n" + "Return the consolidated answer/result directly and concisely.\n" + "If the handoff is insufficient, say what is missing and give the best available concise status; " + "still avoid internal plumbing unless it is necessary to resolve the problem.\n\n" + f"Original user/context excerpt:\n{origin_context[:2000]}\n\n" + f"Task title/context:\n{getattr(task, 'title', '') if task else ''}\n" + f"{str(getattr(task, 'body', '') or '')[:1200]}\n\n" + f"Worker summary / durable handoff:\n{str(worker_summary or '')[:4000]}\n\n" + f"Worker metadata JSON:\n{json.dumps(worker_metadata, ensure_ascii=False, default=str)[:3000]}\n\n" + "Internal debug context (use only if the user's request is explicitly about internals/debugging):\n" + f"Task id: {sub.get('task_id')}\n" + f"Board: {board or 'default'}\n" + f"Event kind: {getattr(event, 'kind', '')}\n" + f"Run id: {getattr(event, 'run_id', '')}\n" + f"Origin session id: {sub.get('origin_session_id') or ''}\n" + f"Origin profile: {sub.get('origin_profile') or ''}\n" + f"Direct fallback message: {direct_message}" + ) + + async def _synthesize_kanban_notification( + self, + *, + sub: dict, + event: Any, + task: Any, + board: Optional[str], + direct_message: str, + ) -> str: + """Run a lightweight no-tools origin-profile turn for a completion. + + The synthesis agent receives only the durable worker handoff and stored + origin context. It intentionally has no toolsets, so it cannot re-run + heavy collection or recursively create Kanban tasks/subscriptions. + """ + from gateway.config import Platform as _Platform + from gateway.session import SessionSource + from run_agent import AIAgent + + platform_str = str(sub.get("platform") or "telegram").lower() + try: + platform = _Platform(platform_str) + except Exception: + platform = _Platform.TELEGRAM + source = SessionSource( + platform=platform, + chat_id=str(sub.get("chat_id") or ""), + user_id=str(sub.get("user_id") or "") or None, + thread_id=str(sub.get("thread_id") or "") or None, + ) + user_config = _load_gateway_config() + model, runtime_kwargs = self._resolve_session_agent_runtime( + source=source, + user_config=user_config, + ) + if not runtime_kwargs.get("api_key"): + raise RuntimeError("no provider credentials configured for synthesis") + + event_payload = getattr(event, "payload", None) or {} + run = None + if getattr(event, "run_id", None): + try: + from hermes_cli import kanban_db as _kb + conn = _kb.connect(board=board) + try: + run = _kb.get_run(conn, int(event.run_id)) + finally: + conn.close() + except Exception: + run = None + worker_summary = ( + (run.summary if run and run.summary else None) + or event_payload.get("summary") + or (task.result if task and getattr(task, "result", None) else None) + or direct_message + ) + worker_metadata = run.metadata if run and run.metadata is not None else None + prompt = self._build_kanban_synthesis_prompt( + sub=sub, + event=event, + task=task, + board=board, + worker_summary=worker_summary, + worker_metadata=worker_metadata, + direct_message=direct_message, + ) + + def run_sync() -> str: + agent = AIAgent( + model=model, + **runtime_kwargs, + max_iterations=1, + quiet_mode=True, + verbose_logging=False, + enabled_toolsets=[], + disabled_toolsets=None, + session_id=f"kanban_synth_{sub.get('task_id')}_{getattr(event, 'id', int(time.time()))}", + platform=platform_str, + user_id=source.user_id, + chat_id=source.chat_id, + thread_id=source.thread_id, + session_db=getattr(self, "_session_db", None), + fallback_model=getattr(self, "_fallback_model", None), + skip_context_files=True, + ) + try: + result = agent.run_conversation(user_message=prompt) + return str((result or {}).get("final_response") or "").strip() + finally: + self._cleanup_agent_resources(agent) + + timeout_s = int((user_config.get("kanban") or {}).get("synthesis_timeout_seconds", 45)) + return await asyncio.wait_for( + self._run_in_executor_with_context(run_sync), + timeout=max(5, timeout_s), + ) + def _kanban_advance( self, sub: dict, cursor: int, board: Optional[str] = None, ) -> None: @@ -6163,61 +6475,63 @@ async def _do_undo(): return self._telegram_topic_root_lobby_message() return None - # ── Claim this session before any await ─────────────────────── - # Between here and _run_agent registering the real AIAgent, there - # are numerous await points (hooks, vision enrichment, STT, - # session hygiene compression). Without this sentinel a second - # message arriving during any of those yields would pass the - # "already running" guard and spin up a duplicate agent for the - # same session — corrupting the transcript. - self._running_agents[_quick_key] = _AGENT_PENDING_SENTINEL - self._running_agents_ts[_quick_key] = time.time() - _run_generation = self._begin_session_run_generation(_quick_key) - - try: - _agent_result = await self._handle_message_with_agent(event, source, _quick_key, _run_generation) - # Goal continuation: after the agent returns a final response - # for this turn, check any standing /goal — the judge will - # either mark it done, pause it (budget), or enqueue a - # continuation prompt back through the adapter FIFO so the - # next turn makes more progress. Wrapped in try/except so a - # broken judge never breaks normal message handling. + conversation_lock = self._conversation_lock_for_key(_quick_key) + async with conversation_lock: + # ── Claim this session before any await ─────────────────────── + # Between here and _run_agent registering the real AIAgent, there + # are numerous await points (hooks, vision enrichment, STT, + # session hygiene compression). Without this sentinel a second + # message arriving during any of those yields would pass the + # "already running" guard and spin up a duplicate agent for the + # same session — corrupting the transcript. + self._running_agents[_quick_key] = _AGENT_PENDING_SENTINEL + self._running_agents_ts[_quick_key] = time.time() + _run_generation = self._begin_session_run_generation(_quick_key) + try: - _final_text = "" - if isinstance(_agent_result, dict): - _final_text = str(_agent_result.get("final_response") or "") - elif isinstance(_agent_result, str): - _final_text = _agent_result - # Skip for empty responses (interrupted / errored) — the - # judge would almost always say "continue" and we'd loop - # on error. Let the user drive the next turn. - if _final_text.strip(): - try: - session_entry = self.session_store.get_or_create_session(source) - except Exception: - session_entry = None - if session_entry is not None: - await self._post_turn_goal_continuation( - session_entry=session_entry, - source=source, - final_response=_final_text, - ) - except Exception as _goal_exc: - logger.debug("goal continuation hook failed: %s", _goal_exc) - return _agent_result - finally: - # If _run_agent replaced the sentinel with a real agent and - # then cleaned it up, this is a no-op. If we exited early - # (exception, command fallthrough, etc.) the sentinel must - # not linger or the session would be permanently locked out. - if self._running_agents.get(_quick_key) is _AGENT_PENDING_SENTINEL: - self._release_running_agent_state(_quick_key) - else: - # Agent path already cleaned _running_agents; make sure - # the paired metadata dicts are gone too. - self._running_agents_ts.pop(_quick_key, None) - if hasattr(self, "_busy_ack_ts"): - self._busy_ack_ts.pop(_quick_key, None) + _agent_result = await self._handle_message_with_agent(event, source, _quick_key, _run_generation) + # Goal continuation: after the agent returns a final response + # for this turn, check any standing /goal — the judge will + # either mark it done, pause it (budget), or enqueue a + # continuation prompt back through the adapter FIFO so the + # next turn makes more progress. Wrapped in try/except so a + # broken judge never breaks normal message handling. + try: + _final_text = "" + if isinstance(_agent_result, dict): + _final_text = str(_agent_result.get("final_response") or "") + elif isinstance(_agent_result, str): + _final_text = _agent_result + # Skip for empty responses (interrupted / errored) — the + # judge would almost always say "continue" and we'd loop + # on error. Let the user drive the next turn. + if _final_text.strip(): + try: + session_entry = self.session_store.get_or_create_session(source) + except Exception: + session_entry = None + if session_entry is not None: + await self._post_turn_goal_continuation( + session_entry=session_entry, + source=source, + final_response=_final_text, + ) + except Exception as _goal_exc: + logger.debug("goal continuation hook failed: %s", _goal_exc) + return _agent_result + finally: + # If _run_agent replaced the sentinel with a real agent and + # then cleaned it up, this is a no-op. If we exited early + # (exception, command fallthrough, etc.) the sentinel must + # not linger or the session would be permanently locked out. + if self._running_agents.get(_quick_key) is _AGENT_PENDING_SENTINEL: + self._release_running_agent_state(_quick_key) + else: + # Agent path already cleaned _running_agents; make sure + # the paired metadata dicts are gone too. + self._running_agents_ts.pop(_quick_key, None) + if hasattr(self, "_busy_ack_ts"): + self._busy_ack_ts.pop(_quick_key, None) async def _prepare_inbound_message_text( self, diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index aa3655b17629..989ac4268652 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -858,6 +858,10 @@ class Event: chat_id TEXT NOT NULL, thread_id TEXT NOT NULL DEFAULT '', user_id TEXT, + notification_mode TEXT NOT NULL DEFAULT 'direct', + origin_session_id TEXT, + origin_profile TEXT, + origin_context TEXT, created_at INTEGER NOT NULL, last_event_id INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (task_id, platform, chat_id, thread_id) @@ -1082,6 +1086,21 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: "ON task_events(run_id, id)" ) + notify_cols = { + row["name"] for row in conn.execute("PRAGMA table_info(kanban_notify_subs)") + } + if "notification_mode" not in notify_cols: + conn.execute( + "ALTER TABLE kanban_notify_subs ADD COLUMN " + "notification_mode TEXT NOT NULL DEFAULT 'direct'" + ) + if "origin_session_id" not in notify_cols: + conn.execute("ALTER TABLE kanban_notify_subs ADD COLUMN origin_session_id TEXT") + if "origin_profile" not in notify_cols: + conn.execute("ALTER TABLE kanban_notify_subs ADD COLUMN origin_profile TEXT") + if "origin_context" not in notify_cols: + conn.execute("ALTER TABLE kanban_notify_subs ADD COLUMN origin_context TEXT") + # One-shot backfill: any task that is 'running' before runs existed # had its claim_lock / claim_expires / worker_pid on the task row. # Synthesize a matching task_runs row so subsequent end-run / heartbeat @@ -4183,6 +4202,11 @@ def task_age(task: Task) -> dict: # Notification subscriptions (used by the gateway kanban-notifier) # --------------------------------------------------------------------------- +def _normalize_notification_mode(mode: Optional[str]) -> str: + value = str(mode or "direct").strip().lower() + return value if value in {"direct", "synthesize", "silent"} else "direct" + + def add_notify_sub( conn: sqlite3.Connection, *, @@ -4191,18 +4215,34 @@ def add_notify_sub( chat_id: str, thread_id: Optional[str] = None, user_id: Optional[str] = None, + notification_mode: Optional[str] = "direct", + origin_session_id: Optional[str] = None, + origin_profile: Optional[str] = None, + origin_context: Optional[str] = None, ) -> None: """Register a gateway source that wants terminal-state notifications for ``task_id``. Idempotent on (task, platform, chat, thread).""" now = int(time.time()) + mode = _normalize_notification_mode(notification_mode) with write_txn(conn): conn.execute( """ - INSERT OR IGNORE INTO kanban_notify_subs - (task_id, platform, chat_id, thread_id, user_id, created_at) - VALUES (?, ?, ?, ?, ?, ?) + INSERT INTO kanban_notify_subs + (task_id, platform, chat_id, thread_id, user_id, + notification_mode, origin_session_id, origin_profile, + origin_context, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(task_id, platform, chat_id, thread_id) DO UPDATE SET + notification_mode = excluded.notification_mode, + user_id = COALESCE(excluded.user_id, user_id), + origin_session_id = COALESCE(excluded.origin_session_id, origin_session_id), + origin_profile = COALESCE(excluded.origin_profile, origin_profile), + origin_context = COALESCE(excluded.origin_context, origin_context) """, - (task_id, platform, chat_id, thread_id or "", user_id, now), + ( + task_id, platform, chat_id, thread_id or "", user_id, + mode, origin_session_id, origin_profile, origin_context, now, + ), ) diff --git a/hermes_cli/tools_config.py b/hermes_cli/tools_config.py index 74fc29247d26..a67ac531a278 100644 --- a/hermes_cli/tools_config.py +++ b/hermes_cli/tools_config.py @@ -71,6 +71,7 @@ ("messaging", "📨 Cross-Platform Messaging", "send_message"), ("rl", "🧪 RL Training", "Tinker-Atropos training tools"), ("homeassistant", "🏠 Home Assistant", "smart home device control"), + ("kanban", "🗂️ Kanban Coordination", "show, create, complete, block, heartbeat"), ("spotify", "🎵 Spotify", "playback, search, playlists, library"), ("discord", "💬 Discord (read/participate)", "fetch messages, search members, create thread"), ("discord_admin", "🛡️ Discord Server Admin", "list channels/roles, pin, assign roles"), diff --git a/model_tools.py b/model_tools.py index 253cf02fe8d2..0b490cd54f32 100644 --- a/model_tools.py +++ b/model_tools.py @@ -702,6 +702,10 @@ def handle_function_call( task_id: Optional[str] = None, tool_call_id: Optional[str] = None, session_id: Optional[str] = None, + platform: Optional[str] = None, + chat_id: Optional[str] = None, + thread_id: Optional[str] = None, + user_id: Optional[str] = None, user_task: Optional[str] = None, enabled_tools: Optional[List[str]] = None, skip_pre_tool_call_hook: bool = False, @@ -713,6 +717,8 @@ def handle_function_call( function_name: Name of the function to call. function_args: Arguments for the function. task_id: Unique identifier for terminal/browser session isolation. + session_id: Hermes conversation/session id for provenance-aware tools. + platform/chat_id/thread_id/user_id: Optional origin surface metadata. user_task: The user's original task (for browser_snapshot context). enabled_tools: Tool names enabled for this session. When provided, execute_code uses this list to determine which sandbox @@ -781,12 +787,22 @@ def handle_function_call( function_name, function_args, task_id=task_id, enabled_tools=sandbox_enabled, + session_id=session_id, + platform=platform, + chat_id=chat_id, + thread_id=thread_id, + user_id=user_id, ) else: result = registry.dispatch( function_name, function_args, task_id=task_id, user_task=user_task, + session_id=session_id, + platform=platform, + chat_id=chat_id, + thread_id=thread_id, + user_id=user_id, ) duration_ms = int((time.monotonic() - _dispatch_start) * 1000) diff --git a/run_agent.py b/run_agent.py index 8ae39c6faf05..dbfc3752b3c9 100644 --- a/run_agent.py +++ b/run_agent.py @@ -10006,6 +10006,11 @@ def _invoke_tool(self, function_name: str, function_args: dict, effective_task_i function_name, function_args, effective_task_id, tool_call_id=tool_call_id, session_id=self.session_id or "", + platform=self.platform or "", + chat_id=self._chat_id or "", + thread_id=self._thread_id or "", + user_id=self._user_id or "", + user_task=getattr(self, "_current_user_message_for_tools", ""), enabled_tools=list(self.valid_tool_names) if self.valid_tool_names else None, skip_pre_tool_call_hook=True, ) @@ -10717,6 +10722,11 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe function_name, function_args, effective_task_id, tool_call_id=tool_call.id, session_id=self.session_id or "", + platform=self.platform or "", + chat_id=self._chat_id or "", + thread_id=self._thread_id or "", + user_id=self._user_id or "", + user_task=getattr(self, "_current_user_message_for_tools", ""), enabled_tools=list(self.valid_tool_names) if self.valid_tool_names else None, skip_pre_tool_call_hook=True, ) @@ -10737,6 +10747,11 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe function_name, function_args, effective_task_id, tool_call_id=tool_call.id, session_id=self.session_id or "", + platform=self.platform or "", + chat_id=self._chat_id or "", + thread_id=self._thread_id or "", + user_id=self._user_id or "", + user_task=getattr(self, "_current_user_message_for_tools", ""), enabled_tools=list(self.valid_tool_names) if self.valid_tool_names else None, skip_pre_tool_call_hook=True, ) @@ -11145,6 +11160,7 @@ def run_conversation( user_message = _sanitize_surrogates(user_message) if isinstance(persist_user_message, str): persist_user_message = _sanitize_surrogates(persist_user_message) + self._current_user_message_for_tools = str(user_message or "")[:4000] # Store stream callback for _interruptible_api_call to pick up self._stream_callback = stream_callback diff --git a/tests/gateway/test_kanban_notifier.py b/tests/gateway/test_kanban_notifier.py new file mode 100644 index 000000000000..fa6b070dea61 --- /dev/null +++ b/tests/gateway/test_kanban_notifier.py @@ -0,0 +1,309 @@ +import asyncio +import pytest +from types import SimpleNamespace + +from gateway.config import Platform +from gateway.platforms.base import SendResult +from gateway.run import GatewayRunner +from gateway.session import SessionSource + + +class _Adapter: + def __init__(self, result): + self.result = result + self.calls = [] + + async def send(self, chat_id, content, metadata=None): + self.calls.append((chat_id, content, metadata)) + return self.result + + +@pytest.mark.asyncio +async def test_kanban_notification_sendresult_failure_raises(monkeypatch): + runner = object.__new__(GatewayRunner) + adapter = _Adapter(SendResult(success=False, error="Not connected")) + sub = {"task_id": "t_fail", "platform": "telegram", "chat_id": "123"} + mirror_calls = [] + monkeypatch.setattr( + "gateway.mirror.mirror_to_session", + lambda *args, **kwargs: mirror_calls.append((args, kwargs)) or True, + ) + + with pytest.raises(RuntimeError, match="Not connected"): + await runner._send_kanban_notification(adapter, sub, "done", {}) + + assert adapter.calls == [("123", "done", {})] + assert mirror_calls == [] + + +@pytest.mark.asyncio +async def test_kanban_notification_sendresult_success_does_not_raise(): + runner = object.__new__(GatewayRunner) + adapter = _Adapter(SendResult(success=True, message_id="m1")) + sub = {"task_id": "t_ok", "platform": "telegram", "chat_id": "123"} + + await runner._send_kanban_notification(adapter, sub, "done", {"thread_id": "7"}) + + assert adapter.calls == [("123", "done", {"thread_id": "7"})] + + +@pytest.mark.asyncio +async def test_kanban_notification_synthesize_mode_sends_synthesized_text(monkeypatch): + runner = object.__new__(GatewayRunner) + adapter = _Adapter(SendResult(success=True, message_id="m1")) + sub = { + "task_id": "t_syn", + "platform": "telegram", + "chat_id": "123", + "notification_mode": "synthesize", + "origin_context": "user asked for a concise result", + } + ev = SimpleNamespace(kind="completed", payload={"summary": "worker handoff"}, run_id=7) + + async def fake_synthesize(**kwargs): + assert kwargs["sub"] is sub + assert kwargs["event"] is ev + assert kwargs["direct_message"] == "direct fallback" + return "synthesized reply" + + monkeypatch.setattr(runner, "_synthesize_kanban_notification", fake_synthesize) + + await runner._send_kanban_notification( + adapter, sub, "direct fallback", {}, event=ev, task=None, board="default" + ) + + assert adapter.calls == [("123", "synthesized reply", {})] + + +@pytest.mark.asyncio +async def test_kanban_notification_synthesis_failure_falls_back_direct(monkeypatch): + runner = object.__new__(GatewayRunner) + adapter = _Adapter(SendResult(success=True, message_id="m1")) + sub = { + "task_id": "t_syn_fail", + "platform": "telegram", + "chat_id": "123", + "notification_mode": "synthesize", + } + ev = SimpleNamespace(kind="completed", payload={"summary": "worker handoff"}, run_id=7) + + async def fake_synthesize(**kwargs): + raise RuntimeError("synthesis unavailable") + + monkeypatch.setattr(runner, "_synthesize_kanban_notification", fake_synthesize) + + await runner._send_kanban_notification( + adapter, sub, "direct fallback", {"thread_id": "9"}, event=ev, task=None + ) + + assert adapter.calls == [("123", "direct fallback", {"thread_id": "9"})] + + +@pytest.mark.asyncio +async def test_kanban_notification_silent_mode_noops(): + runner = object.__new__(GatewayRunner) + adapter = _Adapter(SendResult(success=True, message_id="m1")) + sub = { + "task_id": "t_silent", + "platform": "telegram", + "chat_id": "123", + "notification_mode": "silent", + } + + await runner._send_kanban_notification(adapter, sub, "direct fallback", {}) + + assert adapter.calls == [] + + +@pytest.mark.asyncio +async def test_kanban_notification_success_mirrors_into_origin_session(monkeypatch): + runner = object.__new__(GatewayRunner) + adapter = _Adapter(SendResult(success=True, message_id="m1")) + sub = { + "task_id": "t_ctx", + "platform": "telegram", + "chat_id": "123", + "thread_id": "7", + "user_id": "u1", + } + mirror_calls = [] + + def fake_mirror(platform, chat_id, message_text, source_label="cli", thread_id=None, user_id=None): + mirror_calls.append({ + "platform": platform, + "chat_id": chat_id, + "message_text": message_text, + "source_label": source_label, + "thread_id": thread_id, + "user_id": user_id, + }) + return True + + monkeypatch.setattr("gateway.mirror.mirror_to_session", fake_mirror) + + await runner._send_kanban_notification( + adapter, sub, "done text", {"thread_id": "7"} + ) + + assert adapter.calls == [("123", "done text", {"thread_id": "7"})] + assert mirror_calls == [{ + "platform": "telegram", + "chat_id": "123", + "message_text": "done text", + "source_label": "kanban", + "thread_id": "7", + "user_id": "u1", + }] + + +@pytest.mark.asyncio +async def test_kanban_synthesis_waits_for_active_origin_session_lock(monkeypatch): + runner = object.__new__(GatewayRunner) + runner._conversation_locks = {} + adapter = _Adapter(SendResult(success=True, message_id="m1")) + sub = { + "task_id": "t_wait", + "platform": "telegram", + "chat_id": "123", + "thread_id": "7", + "user_id": "u1", + "notification_mode": "synthesize", + } + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="123", + user_id="u1", + thread_id="7", + ) + lock = runner._conversation_lock_for_source(source) + await lock.acquire() + synth_started = asyncio.Event() + + async def fake_synthesize(**kwargs): + synth_started.set() + return "synth after native" + + monkeypatch.setattr(runner, "_synthesize_kanban_notification", fake_synthesize) + monkeypatch.setattr("gateway.mirror.mirror_to_session", lambda *a, **k: True) + + task = asyncio.create_task( + runner._send_kanban_notification( + adapter, + sub, + "direct fallback", + {"thread_id": "7"}, + event=SimpleNamespace(kind="completed"), + ) + ) + await asyncio.sleep(0) + + assert adapter.calls == [] + assert not synth_started.is_set() + + lock.release() + await asyncio.wait_for(task, timeout=1) + + assert adapter.calls == [("123", "synth after native", {"thread_id": "7"})] + + +@pytest.mark.asyncio +async def test_kanban_notification_does_not_block_unrelated_chats(monkeypatch): + runner = object.__new__(GatewayRunner) + runner._conversation_locks = {} + blocked_adapter = _Adapter(SendResult(success=True, message_id="blocked")) + free_adapter = _Adapter(SendResult(success=True, message_id="free")) + blocked_sub = {"task_id": "t_blocked", "platform": "telegram", "chat_id": "123", "user_id": "u1"} + free_sub = {"task_id": "t_free", "platform": "telegram", "chat_id": "456", "user_id": "u2"} + blocked_source = SessionSource(platform=Platform.TELEGRAM, chat_id="123", user_id="u1") + lock = runner._conversation_lock_for_source(blocked_source) + await lock.acquire() + monkeypatch.setattr("gateway.mirror.mirror_to_session", lambda *a, **k: True) + + blocked_task = asyncio.create_task( + runner._send_kanban_notification(blocked_adapter, blocked_sub, "blocked", {}) + ) + await asyncio.sleep(0) + await runner._send_kanban_notification(free_adapter, free_sub, "free", {}) + + assert blocked_adapter.calls == [] + assert free_adapter.calls == [("456", "free", {})] + + lock.release() + await asyncio.wait_for(blocked_task, timeout=1) + + +def test_kanban_notifier_defaults_to_dispatch_owner(monkeypatch): + """Secondary gateways with dispatch disabled must not consume shared subs.""" + from hermes_cli import config as hermes_config + + runner = object.__new__(GatewayRunner) + monkeypatch.delenv("HERMES_KANBAN_NOTIFY_IN_GATEWAY", raising=False) + monkeypatch.setattr( + hermes_config, + "load_config", + lambda: {"kanban": {"dispatch_in_gateway": False}}, + ) + + assert runner._kanban_notify_in_gateway_enabled() is False + + +def test_kanban_notifier_explicit_config_overrides_dispatch(monkeypatch): + from hermes_cli import config as hermes_config + + runner = object.__new__(GatewayRunner) + monkeypatch.delenv("HERMES_KANBAN_NOTIFY_IN_GATEWAY", raising=False) + monkeypatch.setattr( + hermes_config, + "load_config", + lambda: {"kanban": {"dispatch_in_gateway": False, "notify_in_gateway": True}}, + ) + + assert runner._kanban_notify_in_gateway_enabled() is True + + +def test_kanban_notifier_env_override(monkeypatch): + from hermes_cli import config as hermes_config + + runner = object.__new__(GatewayRunner) + monkeypatch.setattr( + hermes_config, + "load_config", + lambda: {"kanban": {"dispatch_in_gateway": True}}, + ) + + monkeypatch.setenv("HERMES_KANBAN_NOTIFY_IN_GATEWAY", "false") + assert runner._kanban_notify_in_gateway_enabled() is False + + monkeypatch.setenv("HERMES_KANBAN_NOTIFY_IN_GATEWAY", "true") + assert runner._kanban_notify_in_gateway_enabled() is True + + +def test_kanban_synthesis_prompt_hides_internal_workflow_by_default(): + task = SimpleNamespace( + title="answer factual question", + body="User asked: how many airports does Guangzhou have?", + ) + event = SimpleNamespace(payload={"summary": "Guangzhou has one operating passenger airport."}) + sub = { + "task_id": "t_internal123", + "origin_context": "how many airports does Guangzhou have?", + "origin_session_id": "telegram:chat", + "origin_profile": "default", + } + + prompt = GatewayRunner._build_kanban_synthesis_prompt( + sub=sub, + event=event, + task=task, + board="default", + worker_summary="Guangzhou has one operating passenger airport.", + worker_metadata={"assignee": "worker-research"}, + direct_message="✔ @worker-research Kanban t_internal123 done — answer factual question", + ) + + assert "Do NOT mention Kanban" in prompt + assert "task ids" in prompt + assert "assignees" in prompt + assert "worker/dispatcher" in prompt + assert "Internal debug context" in prompt + assert "t_internal123" in prompt # available for debug context, not for normal prose \ No newline at end of file diff --git a/tests/tools/test_kanban_tools.py b/tests/tools/test_kanban_tools.py index f5c7094ee474..2e6ef8202cd9 100644 --- a/tests/tools/test_kanban_tools.py +++ b/tests/tools/test_kanban_tools.py @@ -71,6 +71,15 @@ def worker_env(monkeypatch, tmp_path): home.mkdir() monkeypatch.setenv("HERMES_HOME", str(home)) monkeypatch.setenv("HERMES_PROFILE", "test-worker") + for var in ( + "HERMES_KANBAN_DB", + "HERMES_KANBAN_HOME", + "HERMES_KANBAN_BOARD", + "HERMES_KANBAN_WORKSPACES_ROOT", + "HERMES_KANBAN_RUN_ID", + "HERMES_KANBAN_CLAIM_LOCK", + ): + monkeypatch.delenv(var, raising=False) from pathlib import Path as _Path monkeypatch.setattr(_Path, "home", lambda: tmp_path) @@ -784,3 +793,294 @@ def test_orchestrator_complete_any_task_allowed(monkeypatch, tmp_path): out = kt._handle_complete({"task_id": tid, "summary": "orchestrator close"}) d = json.loads(out) assert d.get("ok") is True and d.get("task_id") == tid + + +def test_notify_subscription_duplicate_updates_mode_and_preserves_context(worker_env): + """Duplicate origin subscriptions should honor the latest delivery mode. + + A manual/direct subscription can be upgraded by a later model-created task + request for synthesized delivery without losing non-null provenance fields. + """ + from hermes_cli import kanban_db as kb + + with kb.connect() as conn: + kb.add_notify_sub( + conn, + task_id=worker_env, + platform="telegram", + chat_id="chat-456", + thread_id="thread-789", + user_id="user-1", + notification_mode="direct", + origin_session_id="origin-session", + origin_profile="default", + origin_context="original user asked for a public reply", + ) + kb.add_notify_sub( + conn, + task_id=worker_env, + platform="telegram", + chat_id="chat-456", + thread_id="thread-789", + user_id=None, + notification_mode="synthesize", + origin_session_id=None, + origin_profile=None, + origin_context=None, + ) + subs = kb.list_notify_subs(conn, worker_env) + + assert len(subs) == 1 + assert subs[0]["notification_mode"] == "synthesize" + assert subs[0]["user_id"] == "user-1" + assert subs[0]["origin_session_id"] == "origin-session" + assert subs[0]["origin_profile"] == "default" + assert subs[0]["origin_context"] == "original user asked for a public reply" + + +def test_create_worker_root_task_inherits_current_origin_subscription(worker_env): + """Parentless worker-created follow-up tasks keep the interactive origin.""" + from hermes_cli import kanban_db as kb + from tools import kanban_tools as kt + + with kb.connect() as conn: + kb.add_notify_sub( + conn, + task_id=worker_env, + platform="telegram", + chat_id="chat-456", + thread_id="thread-789", + user_id="user-1", + notification_mode="synthesize", + origin_session_id="origin-session", + origin_profile="default", + origin_context="please recover this user-visible task", + ) + + out = kt._handle_create({ + "title": "root recovery follow-up", + "assignee": "worker", + }) + d = json.loads(out) + assert d["ok"] is True + assert d["notification_subscription"] == { + "platform": "telegram", + "chat_id": "chat-456", + "thread_id": "thread-789", + "notification_mode": "synthesize", + "inherited_from_task": worker_env, + } + + with kb.connect() as conn: + subs = kb.list_notify_subs(conn, d["task_id"]) + assert len(subs) == 1 + assert subs[0]["platform"] == "telegram" + assert subs[0]["chat_id"] == "chat-456" + assert subs[0]["thread_id"] == "thread-789" + assert subs[0]["user_id"] == "user-1" + assert subs[0]["notification_mode"] == "synthesize" + assert subs[0]["origin_session_id"] == "origin-session" + assert subs[0]["origin_profile"] == "default" + assert subs[0]["origin_context"] == "please recover this user-visible task" + + +def test_create_worker_child_does_not_inherit_current_origin_subscription(worker_env): + """Normal worker fan-out linked by parent remains silent.""" + from hermes_cli import kanban_db as kb + from tools import kanban_tools as kt + + with kb.connect() as conn: + kb.add_notify_sub( + conn, + task_id=worker_env, + platform="telegram", + chat_id="chat-456", + notification_mode="synthesize", + ) + + out = kt._handle_create({ + "title": "internal child", + "assignee": "worker", + "parents": [worker_env], + }) + d = json.loads(out) + assert d["ok"] is True + assert d["notification_subscription"] is None + + with kb.connect() as conn: + assert kb.list_notify_subs(conn, d["task_id"]) == [] + + +def test_create_worker_root_task_respects_silent_notification_mode(worker_env): + """Workers can still explicitly suppress inherited origin delivery.""" + from hermes_cli import kanban_db as kb + from tools import kanban_tools as kt + + with kb.connect() as conn: + kb.add_notify_sub( + conn, + task_id=worker_env, + platform="telegram", + chat_id="chat-456", + notification_mode="synthesize", + ) + + out = kt._handle_create({ + "title": "silent root follow-up", + "assignee": "worker", + "notification_mode": "silent", + }) + d = json.loads(out) + assert d["ok"] is True + assert d["notification_subscription"] is None + + with kb.connect() as conn: + assert kb.list_notify_subs(conn, d["task_id"]) == [] + + +def test_create_auto_subscribes_cli_origin(monkeypatch, tmp_path): + """Model-tool-created tasks subscribe the initiating CLI session.""" + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setenv("HERMES_PROFILE", "orchestrator") + monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False) + for var in ( + "HERMES_KANBAN_DB", + "HERMES_KANBAN_HOME", + "HERMES_KANBAN_BOARD", + "HERMES_KANBAN_WORKSPACES_ROOT", + ): + monkeypatch.delenv(var, raising=False) + from pathlib import Path as _Path + monkeypatch.setattr(_Path, "home", lambda: tmp_path) + + from hermes_cli import kanban_db as kb + kb._INITIALIZED_PATHS.clear() + kb.init_db() + + import model_tools + out = model_tools.handle_function_call( + "kanban_create", + {"title": "agent-created", "assignee": "worker"}, + task_id="tool-session", + platform="cli", + session_id="session-123", + skip_pre_tool_call_hook=True, + ) + d = json.loads(out) + assert d["ok"] is True + assert d["notification_subscription"] == { + "platform": "cli", + "chat_id": "session-123", + "thread_id": "", + "notification_mode": "direct", + } + + with kb.connect() as conn: + subs = kb.list_notify_subs(conn, d["task_id"]) + assert len(subs) == 1 + assert subs[0]["platform"] == "cli" + assert subs[0]["chat_id"] == "session-123" + assert subs[0]["thread_id"] == "" + + kb.complete_task(conn, d["task_id"], summary="done by synthetic worker") + new_cursor, events = kb.unseen_events_for_sub( + conn, + task_id=d["task_id"], + platform="cli", + chat_id="session-123", + kinds=("completed", "blocked", "gave_up", "crashed", "timed_out"), + ) + assert new_cursor > 0 + assert [event.kind for event in events] == ["completed"] + assert events[0].payload["summary"] == "done by synthetic worker" + + +def test_create_auto_subscribes_gateway_origin_with_chat_id(monkeypatch, tmp_path): + """Gateway-origin tool calls subscribe only when the routable chat id is present.""" + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setenv("HERMES_PROFILE", "orchestrator") + monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False) + for var in ( + "HERMES_KANBAN_DB", + "HERMES_KANBAN_HOME", + "HERMES_KANBAN_BOARD", + "HERMES_KANBAN_WORKSPACES_ROOT", + ): + monkeypatch.delenv(var, raising=False) + from pathlib import Path as _Path + monkeypatch.setattr(_Path, "home", lambda: tmp_path) + + from hermes_cli import kanban_db as kb + kb._INITIALIZED_PATHS.clear() + kb.init_db() + + from tools import kanban_tools as kt + out = kt._handle_create( + {"title": "gateway-agent-created", "assignee": "worker"}, + platform="telegram", + session_id="not-routable-session", + chat_id="chat-456", + thread_id="thread-789", + user_id="user-1", + user_task="please run this with context", + ) + d = json.loads(out) + assert d["ok"] is True + assert d["notification_subscription"] == { + "platform": "telegram", + "chat_id": "chat-456", + "thread_id": "thread-789", + "notification_mode": "synthesize", + } + assert d["user_facing_status"] == "I’ll look into it and report back here." + + with kb.connect() as conn: + subs = kb.list_notify_subs(conn, d["task_id"]) + assert len(subs) == 1 + assert subs[0]["platform"] == "telegram" + assert subs[0]["chat_id"] == "chat-456" + assert subs[0]["thread_id"] == "thread-789" + assert subs[0]["user_id"] == "user-1" + assert subs[0]["notification_mode"] == "synthesize" + assert subs[0]["origin_session_id"] == "not-routable-session" + assert subs[0]["origin_profile"] == "orchestrator" + assert subs[0]["origin_context"] == "please run this with context" + + +def test_create_does_not_subscribe_gateway_origin_without_chat_id(monkeypatch, tmp_path): + """A gateway session id alone is not a routable notification endpoint.""" + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setenv("HERMES_PROFILE", "orchestrator") + monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False) + for var in ( + "HERMES_KANBAN_DB", + "HERMES_KANBAN_HOME", + "HERMES_KANBAN_BOARD", + "HERMES_KANBAN_WORKSPACES_ROOT", + ): + monkeypatch.delenv(var, raising=False) + from pathlib import Path as _Path + monkeypatch.setattr(_Path, "home", lambda: tmp_path) + + from hermes_cli import kanban_db as kb + kb._INITIALIZED_PATHS.clear() + kb.init_db() + + from tools import kanban_tools as kt + out = kt._handle_create( + {"title": "gateway-no-chat", "assignee": "worker"}, + platform="telegram", + session_id="not-routable-session", + ) + d = json.loads(out) + assert d["ok"] is True + assert d["notification_subscription"] is None + + with kb.connect() as conn: + assert kb.list_notify_subs(conn, d["task_id"]) == [] diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index 366252e385e1..31f694bf8e0b 100644 --- a/tools/kanban_tools.py +++ b/tools/kanban_tools.py @@ -135,6 +135,127 @@ def _ok(**fields: Any) -> str: return json.dumps({"ok": True, **fields}) +def _auto_subscribe_origin_for_created_task(kb, conn, created_task_id: str, **kw) -> Optional[dict[str, str]]: + """Subscribe the originating user surface for agent-created tasks. + + Slash-command creates already subscribe in gateway/run.py because they have + adapter source metadata in hand. This helper covers model-tool-created + tasks from normal/orchestrator chats by reusing kanban_notify_subs. + + Dispatcher-spawned workers set HERMES_KANBAN_TASK; skip those by default so + worker fan-out child tasks don't spam the worker's private run session. + Parentless worker-created recovery/root follow-ups are handled separately by + inheriting an existing notification subscription from the current task. + """ + if os.environ.get("HERMES_KANBAN_TASK"): + return None + + session_id = str(kw.get("session_id") or "").strip() + platform = str( + kw.get("platform") or os.environ.get("HERMES_SESSION_SOURCE") or "" + ).strip().lower() + if not platform: + platform = "cli" if session_id else "" + if not platform: + return None + + raw_chat_id = str(kw.get("chat_id") or "").strip() + if platform == "cli": + chat_id = raw_chat_id or session_id + else: + # Gateway delivery uses adapter chat ids. A Hermes session id alone is + # useful provenance but is not a routable Telegram/Discord/etc target. + chat_id = raw_chat_id + if not chat_id: + return None + + thread_id = str(kw.get("thread_id") or "").strip() + user_id = str(kw.get("user_id") or "").strip() or None + requested_mode = str(kw.get("notification_mode") or "").strip().lower() + if requested_mode in {"direct", "synthesize", "silent"}: + notification_mode = requested_mode + elif platform == "telegram": + # Interactive Telegram model-tool creates should flow back through the + # origin/default profile for persona/context-aware synthesis. CLI and + # cron/batch remain direct to avoid surprise nested agent runs. + notification_mode = "synthesize" + else: + notification_mode = "direct" + origin_context = str(kw.get("user_task") or "").strip()[:2000] or None + origin_profile = os.environ.get("HERMES_PROFILE") or None + kb.add_notify_sub( + conn, + task_id=created_task_id, + platform=platform, + chat_id=chat_id, + thread_id=thread_id or None, + user_id=user_id, + notification_mode=notification_mode, + origin_session_id=session_id or None, + origin_profile=origin_profile, + origin_context=origin_context, + ) + return { + "platform": platform, + "chat_id": chat_id, + "thread_id": thread_id, + "notification_mode": notification_mode, + } + + +def _inherit_notify_sub_for_worker_root_task( + kb, conn, created_task_id: str, parents: list[str], requested_mode: Any = None, +) -> Optional[dict[str, str]]: + """Propagate the current task's origin only for parentless worker roots. + + Worker-created child/fan-out tasks should normally be silent and dependency + linked to their current task. The exception is a parentless recovery/root + follow-up created from inside a worker handling an interactive request; that + task is user-visible and should keep the origin subscription so completion + auto-returns to the initiating chat. + """ + current_task_id = os.environ.get("HERMES_KANBAN_TASK") + if not current_task_id or parents: + return None + if str(requested_mode or "").strip().lower() == "silent": + return None + + current_subs = kb.list_notify_subs(conn, current_task_id) + if not current_subs: + return None + + first: Optional[dict[str, str]] = None + for sub in current_subs: + notification_mode = str( + requested_mode or sub.get("notification_mode") or "direct" + ).strip().lower() + if notification_mode not in {"direct", "synthesize", "silent"}: + notification_mode = "direct" + if notification_mode == "silent": + continue + kb.add_notify_sub( + conn, + task_id=created_task_id, + platform=str(sub.get("platform") or ""), + chat_id=str(sub.get("chat_id") or ""), + thread_id=str(sub.get("thread_id") or "") or None, + user_id=sub.get("user_id"), + notification_mode=notification_mode, + origin_session_id=sub.get("origin_session_id"), + origin_profile=sub.get("origin_profile"), + origin_context=sub.get("origin_context"), + ) + if first is None: + first = { + "platform": str(sub.get("platform") or ""), + "chat_id": str(sub.get("chat_id") or ""), + "thread_id": str(sub.get("thread_id") or ""), + "notification_mode": notification_mode, + "inherited_from_task": current_task_id, + } + return first + + # --------------------------------------------------------------------------- # Handlers # --------------------------------------------------------------------------- @@ -456,9 +577,24 @@ def _handle_create(args: dict, **kw) -> str: created_by=os.environ.get("HERMES_PROFILE") or "worker", ) new_task = kb.get_task(conn, new_tid) + subscription = _auto_subscribe_origin_for_created_task( + kb, conn, new_tid, + notification_mode=args.get("notification_mode"), + **kw, + ) + if subscription is None: + subscription = _inherit_notify_sub_for_worker_root_task( + kb, conn, new_tid, list(parents), + requested_mode=args.get("notification_mode"), + ) + user_facing_status = None + if subscription and str(subscription.get("platform") or "").lower() == "telegram": + user_facing_status = "I’ll look into it and report back here." return _ok( task_id=new_tid, status=new_task.status if new_task else None, + notification_subscription=subscription, + user_facing_status=user_facing_status, ) finally: conn.close() @@ -782,6 +918,19 @@ def _handle_link(args: dict, **kw) -> str: "assignee's profile." ), }, + "notification_mode": { + "type": "string", + "enum": ["direct", "synthesize", "silent"], + "description": ( + "Optional completion notification behavior for the origin " + "subscription. 'direct' sends the worker handoff as-is; " + "'synthesize' asks the origin/default profile to craft the " + "user-facing reply from the worker handoff and stored origin " + "context; 'silent' subscribes but sends no terminal ping. " + "Defaults to synthesize for interactive Telegram " + "tool-created tasks and direct elsewhere." + ), + }, }, "required": ["title", "assignee"], }, diff --git a/toolsets.py b/toolsets.py index 11114908a486..9acd1082be54 100644 --- a/toolsets.py +++ b/toolsets.py @@ -227,9 +227,10 @@ "kanban": { "description": ( - "Kanban multi-agent coordination — only active when the agent " + "Kanban multi-agent coordination — active when the agent " "is spawned by the kanban dispatcher (HERMES_KANBAN_TASK env " - "set). The dispatcher runs inside the gateway by default; see " + "set) or when a profile explicitly enables the kanban toolset. " + "The dispatcher runs inside the gateway by default; see " "`kanban.dispatch_in_gateway` in config.yaml. Lets workers mark " "tasks done with structured handoffs, block for human input, " "heartbeat during long ops, comment on threads, and (for "