diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 79352e2fe3a26..9a68982c7905b 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -1010,6 +1010,7 @@ class AsyncCodexAuxiliaryClient: """Async-compatible wrapper matching AsyncOpenAI.chat.completions.create().""" def __init__(self, sync_wrapper: "CodexAuxiliaryClient"): + self._sync_wrapper = sync_wrapper sync_adapter = sync_wrapper.chat.completions async_adapter = _AsyncCodexCompletionsAdapter(sync_adapter) self.chat = _AsyncCodexChatShim(async_adapter) @@ -4487,6 +4488,73 @@ def _force_close_async_httpx(client: Any) -> None: pass +def _is_client_closed(client: Any) -> bool: + """Return True when a cached auxiliary client is no longer usable. + + The OpenAI SDK reports requests through a closed underlying httpx client as + ``APIConnectionError('Connection error.')``. The Codex Responses timeout + watchdog intentionally closes the underlying client to unblock a stuck + stream, so the cache must not hand that wrapper back to later auxiliary + calls. + + This is deliberately best-effort and side-effect free: if a provider + wrapper does not expose a closed flag, treat it as open. + """ + if client is None: + return False + + seen: set[int] = set() + + def _check(obj: Any) -> bool: + if obj is None: + return False + oid = id(obj) + if oid in seen: + return False + seen.add(oid) + + try: + closed_flag = getattr(obj, "is_closed", False) + if isinstance(closed_flag, bool) and closed_flag: + return True + if callable(closed_flag): + try: + result = closed_flag() + except TypeError: + result = False + if isinstance(result, bool) and result: + return True + except Exception: + pass + + try: + state = getattr(obj, "_state", None) + if state is not None and str(state).endswith(".CLOSED"): + return True + except Exception: + pass + + # Wrapper chain examples: + # CodexAuxiliaryClient._real_client -> OpenAI._client -> httpx.Client + # AsyncCodexAuxiliaryClient._sync_wrapper -> CodexAuxiliaryClient + # Anthropic/other adapters may expose _client directly. + # Only inspect explicitly stored attributes. unittest.mock.MagicMock + # fabricates arbitrary attributes on access; recursing into those would + # make every mock look like a nested client and can produce false + # closed detections on cache-hit tests. + try: + obj_vars = vars(obj) + except TypeError: + obj_vars = {} + for attr in ("_sync_wrapper", "_real_client", "_client", "client"): + inner = obj_vars.get(attr) + if inner is not None and inner is not obj and _check(inner): + return True + return False + + return _check(client) + + def shutdown_cached_clients() -> None: """Close all cached clients (sync and async) to prevent event-loop errors. @@ -4607,7 +4675,18 @@ def _get_cached_client( with _client_cache_lock: if cache_key in _client_cache: cached_client, cached_default, cached_loop = _client_cache[cache_key] - if async_mode: + if _is_client_closed(cached_client): + # The Codex Responses timeout watchdog closes the underlying + # OpenAI/httpx client to unblock a stuck stream. Do not reuse + # that wrapper: the next request would surface only as the + # misleading OpenAI SDK message "Connection error.". + logger.debug( + "Auxiliary %s client cache entry is closed; evicting and rebuilding", + provider, + ) + _force_close_async_httpx(cached_client) + del _client_cache[cache_key] + elif async_mode: # Validate: the cached client must be bound to the CURRENT, # OPEN loop. If the loop changed or was closed, the httpx # transport inside is dead — force-close and replace. @@ -4662,7 +4741,30 @@ def _get_cached_client( del _client_cache[evict_key] _client_cache[cache_key] = (client, default_model, bound_loop) else: - client, default_model, _ = _client_cache[cache_key] + cached_client, cached_default, cached_loop = _client_cache[cache_key] + reuse_cached = not _is_client_closed(cached_client) + if reuse_cached and async_mode: + reuse_cached = ( + cached_loop is not None + and cached_loop is current_loop + and not cached_loop.is_closed() + ) + + if reuse_cached: + # Another thread populated an equivalent usable entry while + # we were building outside the lock. Use the winner and + # close the loser so we do not leak connection pools. + _force_close_async_httpx(client) + try: + close_fn = getattr(client, "close", None) + if callable(close_fn): + close_fn() + except Exception: + pass + client, default_model = cached_client, cached_default + else: + _force_close_async_httpx(cached_client) + _client_cache[cache_key] = (client, default_model, bound_loop) return client, model or default_model diff --git a/agent/codex_responses_adapter.py b/agent/codex_responses_adapter.py index 943131f55924d..8ccc876232146 100644 --- a/agent/codex_responses_adapter.py +++ b/agent/codex_responses_adapter.py @@ -1074,8 +1074,8 @@ def _normalize_codex_response( message_items_raw: List[Dict[str, Any]] = [] tool_calls: List[Any] = [] has_incomplete_items = response_status in {"queued", "in_progress", "incomplete"} - saw_commentary_phase = False - saw_final_answer_phase = False + saw_hidden_phase_message_text = False + saw_unknown_phase_message_text = False saw_reasoning_item = False for item in output: @@ -1094,13 +1094,20 @@ def _normalize_codex_response( normalized_phase = None if isinstance(item_phase, str): normalized_phase = item_phase.strip().lower() - if normalized_phase in {"commentary", "analysis"}: - saw_commentary_phase = True - elif normalized_phase in {"final_answer", "final"}: - saw_final_answer_phase = True message_text = _extract_responses_message_text(item) if message_text: - content_parts.append(message_text) + is_hidden_phase = normalized_phase in {"commentary", "analysis"} + is_visible_phase = normalized_phase in {None, "", "final_answer", "final"} + if is_hidden_phase: + saw_hidden_phase_message_text = True + elif not is_visible_phase: + saw_unknown_phase_message_text = True + logger.warning( + "Codex response message has unknown phase %r; hiding it and treating response as incomplete.", + normalized_phase, + ) + if is_visible_phase: + content_parts.append(message_text) raw_message_item: Dict[str, Any] = { "type": "message", "role": "assistant", @@ -1195,7 +1202,7 @@ def _normalize_codex_response( )) final_text = "\n".join([p for p in content_parts if p]).strip() - if not final_text and hasattr(response, "output_text"): + if not final_text and not (saw_hidden_phase_message_text or saw_unknown_phase_message_text) and hasattr(response, "output_text"): out_text = getattr(response, "output_text", "") if isinstance(out_text, str): final_text = out_text.strip() @@ -1245,7 +1252,7 @@ def _normalize_codex_response( finish_reason = "tool_calls" elif leaked_tool_call_text: finish_reason = "incomplete" - elif has_incomplete_items or (saw_commentary_phase and not saw_final_answer_phase): + elif has_incomplete_items or ((saw_hidden_phase_message_text or saw_unknown_phase_message_text) and not final_text): finish_reason = "incomplete" elif (reasoning_items_raw or reasoning_parts or saw_reasoning_item) and not final_text: # Response contains only reasoning (encrypted thinking state and/or diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 398deed3c166f..d14a7faaea69f 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -25,6 +25,33 @@ logger = logging.getLogger(__name__) +try: + import openai.lib._parsing._responses as _openai_parse_responses + + if not getattr(_openai_parse_responses.parse_response, "_hermes_none_guard", False): + _hermes_orig_parse_response = _openai_parse_responses.parse_response + + def _hermes_patched_parse_response(*, text_format, input_tools, response): + # The ChatGPT Codex backend can close after response.output_item.done + # without response.completed, leaving SDK snapshots with output=None. + if getattr(response, "output", None) is None: + response.output = [] + return _hermes_orig_parse_response( + text_format=text_format, + input_tools=input_tools, + response=response, + ) + + _hermes_patched_parse_response._hermes_none_guard = True + _openai_parse_responses.parse_response = _hermes_patched_parse_response +except Exception as _e: + logger.warning( + "Hermes: failed to apply openai SDK parse_response None-guard (%s); " + "Codex streams that close without response.completed may crash with TypeError.", + _e, + ) + + def run_codex_app_server_turn( agent, *, diff --git a/cli.py b/cli.py index 000778b750f85..cdadb70d918a8 100644 --- a/cli.py +++ b/cli.py @@ -5052,7 +5052,11 @@ def _resolve_turn_agent_config(self, user_message: str) -> dict: return route try: - overrides = resolve_fast_mode_overrides(route["model"]) + overrides = resolve_fast_mode_overrides( + route["model"], + provider=runtime.get("provider"), + api_mode=runtime.get("api_mode"), + ) except Exception: overrides = None route["request_overrides"] = overrides @@ -6307,12 +6311,15 @@ def _show_session_status(self): def _fast_command_available(self) -> bool: try: - from hermes_cli.models import model_supports_fast_mode + from hermes_cli.models import model_supports_fast_mode, runtime_supports_priority_processing except Exception: return False agent = getattr(self, "agent", None) model = getattr(agent, "model", None) or getattr(self, "model", None) - return model_supports_fast_mode(model) + return model_supports_fast_mode(model) or runtime_supports_priority_processing( + getattr(self, "provider", None) or getattr(self, "requested_provider", None), + getattr(self, "api_mode", None), + ) def _command_available(self, slash_command: str) -> bool: if slash_command == "/fast": diff --git a/cron/scheduler.py b/cron/scheduler.py index f5c71ceed4f09..294c99a6ea67e 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -721,6 +721,20 @@ def _send_media_via_adapter( logger.warning("Job '%s': failed to send media %s: %s", job.get("id", "?"), media_path, e) +def _raw_response_get(raw_response, key: str): + """Safely read optional metadata from adapter raw_response objects. + + Some gateway adapters return SDK response objects as ``SendResult.raw_response`` + (for example Feishu/Lark ``ReplyMessageResponse``) rather than dictionaries. + Cron delivery uses this metadata only for best-effort diagnostics, so a + non-mapping raw response must not turn an already-successful live-adapter + send into a fallback send that duplicates the message. + """ + if isinstance(raw_response, dict): + return raw_response.get(key) + return None + + def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Optional[str]: """ Deliver job output to the configured target(s) (origin chat, specific platform, etc.). @@ -852,9 +866,9 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option send_result and thread_id and getattr(send_result, "raw_response", None) - and send_result.raw_response.get("thread_fallback") + and _raw_response_get(send_result.raw_response, "thread_fallback") ): - requested_thread_id = send_result.raw_response.get("requested_thread_id") or thread_id + requested_thread_id = _raw_response_get(send_result.raw_response, "requested_thread_id") or thread_id msg = ( f"configured thread_id {requested_thread_id} for " f"{platform_name}:{chat_id} was not found; delivered without thread_id" diff --git a/gateway/config.py b/gateway/config.py index f130fa7da81a9..9112049025251 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -912,7 +912,7 @@ def _merge_platform_map(source_platforms: Any) -> None: bridged["group_allow_admin_from"] = platform_cfg["group_allow_admin_from"] if "group_user_allowed_commands" in platform_cfg: bridged["group_user_allowed_commands"] = platform_cfg["group_user_allowed_commands"] - if plat in {Platform.DISCORD, Platform.SLACK} and "channel_skill_bindings" in platform_cfg: + if plat in {Platform.DISCORD, Platform.SLACK, Platform.FEISHU} and "channel_skill_bindings" in platform_cfg: bridged["channel_skill_bindings"] = platform_cfg["channel_skill_bindings"] if "channel_prompts" in platform_cfg: channel_prompts = platform_cfg["channel_prompts"] diff --git a/gateway/platforms/feishu.py b/gateway/platforms/feishu.py index b361ebc8cfcc1..e125e4a49a7fa 100644 --- a/gateway/platforms/feishu.py +++ b/gateway/platforms/feishu.py @@ -155,7 +155,8 @@ re.MULTILINE, ) # Detect markdown tables: a line starting with | followed by a separator line. -# Feishu post-type 'md' elements do not render tables, so we force text mode. +# Feishu post-type 'md' elements do not render tables reliably, so outbound +# markdown tables are converted into fenced code blocks before sending as post. _MARKDOWN_TABLE_RE = re.compile(r"^\|.*\|\n\|[-|: ]+\|", re.MULTILINE) _MARKDOWN_LINK_RE = re.compile(r"\[([^\]]+)\]\(([^)]+)\)") _MARKDOWN_FENCE_OPEN_RE = re.compile(r"^```([^\n`]*)\s*$") @@ -369,6 +370,7 @@ class FeishuAdapterSettings: encrypt_key: str verification_token: str group_policy: str + require_mention: bool allowed_group_users: frozenset[str] # Bot's own open_id (app-scoped) — returned by /bot/v3/info. Used only for # @mention matching: Feishu puts this value in mentions[].id.open_id when @@ -394,7 +396,6 @@ class FeishuAdapterSettings: default_group_policy: str = "" group_rules: Dict[str, FeishuGroupRule] = field(default_factory=dict) allow_bots: str = "none" # "none" | "mentions" | "all" - require_mention: bool = True @dataclass @@ -543,6 +544,19 @@ def _coerce_required_int(value: Any, default: int, min_value: int = 0) -> int: return default if parsed is None else parsed +def _coerce_bool(value: Any, default: bool = True) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + normalized = str(value).strip().lower() + if normalized in {"1", "true", "yes", "y", "on"}: + return True + if normalized in {"0", "false", "no", "n", "off"}: + return False + return default + + # --------------------------------------------------------------------------- # Post payload builders and parsers # --------------------------------------------------------------------------- @@ -560,6 +574,69 @@ def _build_markdown_post_payload(content: str) -> str: ) +def _is_markdown_table_separator(line: str) -> bool: + stripped = line.strip() + if not (stripped.startswith("|") and stripped.endswith("|")): + return False + cells = [cell.strip() for cell in stripped.strip("|").split("|")] + if not cells or any(not cell for cell in cells): + return False + return all(re.fullmatch(r":?-{3,}:?", cell) for cell in cells) + + +def _is_markdown_table_row(line: str) -> bool: + stripped = line.strip() + return stripped.startswith("|") and stripped.endswith("|") and stripped.count("|") >= 2 + + +def _convert_markdown_tables_to_code_blocks(content: str) -> str: + """Wrap markdown tables in fenced code blocks for Feishu md rendering. + + Feishu's post ``md`` element supports common markdown but not GitHub-style + pipe tables. Sending raw tables as ``md`` can render blank content in the + client, while sending the whole reply as plain text loses all markdown + formatting. Converting only table blocks to ``text`` fences preserves the + rest of the reply as rich markdown and keeps the table readable. + """ + if not content or not _MARKDOWN_TABLE_RE.search(content): + return content + + lines = content.splitlines() + converted: List[str] = [] + index = 0 + in_code_block = False + + while index < len(lines): + line = lines[index] + stripped_line = line.strip() + + if _MARKDOWN_FENCE_OPEN_RE.match(stripped_line) or _MARKDOWN_FENCE_CLOSE_RE.match(stripped_line): + in_code_block = not in_code_block + converted.append(line) + index += 1 + continue + + if ( + not in_code_block + and index + 1 < len(lines) + and _is_markdown_table_row(line) + and _is_markdown_table_separator(lines[index + 1]) + ): + table_lines = [line, lines[index + 1]] + index += 2 + while index < len(lines) and _is_markdown_table_row(lines[index]): + table_lines.append(lines[index]) + index += 1 + converted.extend(["```text", *table_lines, "```"]) + continue + + converted.append(line) + index += 1 + + trailing_newline = "\n" if content.endswith("\n") else "" + return "\n".join(converted) + trailing_newline + + def _build_markdown_post_rows(content: str) -> List[List[Dict[str, str]]]: """Build Feishu post rows while isolating fenced code blocks. @@ -1522,7 +1599,11 @@ def _load_settings(extra: Dict[str, Any]) -> FeishuAdapterSettings: verification_token=str( extra.get("verification_token") or os.getenv("FEISHU_VERIFICATION_TOKEN", "") ).strip(), - group_policy=os.getenv("FEISHU_GROUP_POLICY", "allowlist").strip().lower(), + group_policy=str(extra.get("group_policy") or os.getenv("FEISHU_GROUP_POLICY", "allowlist")).strip().lower(), + require_mention=_coerce_bool( + extra.get("require_mention", os.getenv("FEISHU_REQUIRE_MENTION", "true")), + default=True, + ), allowed_group_users=frozenset( item.strip() for item in os.getenv("FEISHU_ALLOWED_USERS", "").split(",") @@ -1570,9 +1651,6 @@ def _load_settings(extra: Dict[str, Any]) -> FeishuAdapterSettings: default_group_policy=default_group_policy, group_rules=group_rules, allow_bots=allow_bots, - require_mention=_to_boolean( - extra.get("require_mention", os.getenv("FEISHU_REQUIRE_MENTION", "true")) - ), ) def _apply_settings(self, settings: FeishuAdapterSettings) -> None: @@ -1583,6 +1661,7 @@ def _apply_settings(self, settings: FeishuAdapterSettings) -> None: self._encrypt_key = settings.encrypt_key self._verification_token = settings.verification_token self._group_policy = settings.group_policy + self._require_mention = settings.require_mention self._allowed_group_users = set(settings.allowed_group_users) self._admins = set(settings.admins) self._default_group_policy = settings.default_group_policy or settings.group_policy @@ -3093,6 +3172,129 @@ def _clear_webhook_anomaly(self, remote_ip: str) -> None: # Inbound processing pipeline # ========================================================================= + @staticmethod + def _normalize_root_message_id(value: Any) -> Optional[str]: + candidate = str(value or "").strip() + if not candidate or candidate.startswith("omt_"): + return None + return candidate + + @staticmethod + def _extract_topic_thread_id(message: Any) -> Optional[str]: + candidate = str(getattr(message, "thread_id", "") or "").strip() + return candidate if candidate.startswith("omt_") else None + + @classmethod + def _extract_thread_context_id(cls, message: Any) -> Optional[str]: + return ( + cls._normalize_root_message_id(getattr(message, "root_id", None)) + or cls._normalize_root_message_id(getattr(message, "upper_message_id", None)) + or None + ) + + async def _fetch_message_item(self, message_id: str) -> Optional[Any]: + if not self._client or not message_id: + return None + try: + request = self._build_get_message_request(message_id) + response = await asyncio.to_thread(self._client.im.v1.message.get, request) + if not response or getattr(response, "success", lambda: False)() is False: + code = getattr(response, "code", "unknown") + msg = getattr(response, "msg", "message lookup failed") + logger.warning("[Feishu] Failed to fetch message %s: [%s] %s", message_id, code, msg) + return None + items = getattr(getattr(response, "data", None), "items", None) or [] + return items[0] if items else None + except Exception: + logger.warning("[Feishu] Failed to fetch message %s", message_id, exc_info=True) + return None + + async def _resolve_thread_context_id( + self, + *, + message: Any = None, + message_id: Optional[str] = None, + fetch_current_message: bool = False, + ) -> Optional[str]: + thread_context_id = self._extract_thread_context_id(message) if message is not None else None + if thread_context_id: + return thread_context_id + + lookup_message_id = str(message_id or getattr(message, "message_id", None) or "").strip() + topic_thread_id = self._extract_topic_thread_id(message) if message is not None else None + resolved_message = message + + if lookup_message_id and (message is None or (fetch_current_message and topic_thread_id)): + resolved_message = await self._fetch_message_item(lookup_message_id) + thread_context_id = self._extract_thread_context_id(resolved_message) if resolved_message is not None else None + if thread_context_id: + return thread_context_id + topic_thread_id = topic_thread_id or self._extract_topic_thread_id(resolved_message) + + if not topic_thread_id: + return None + + parent_message_id = self._normalize_root_message_id( + getattr(resolved_message, "parent_id", None) + if resolved_message is not None + else None + ) + if parent_message_id: + parent_message = await self._fetch_message_item(parent_message_id) + if parent_message is not None: + return ( + self._extract_thread_context_id(parent_message) + or self._normalize_root_message_id(getattr(parent_message, "message_id", None)) + ) + + current_message_id = self._normalize_root_message_id( + getattr(resolved_message, "message_id", None) + if resolved_message is not None + else lookup_message_id + ) + if current_message_id: + logger.info( + "[Feishu] Treating message %s as thread root for topic %s because Feishu omitted root/upper/parent ids", + current_message_id, + topic_thread_id, + ) + return current_message_id + + logger.info( + "[Feishu] Unable to recover root message id for topic %s from message %s", + topic_thread_id, + lookup_message_id or "", + ) + return None + + def _resolve_channel_skills(self, channel_id: str, parent_id: str | None = None) -> list[str] | None: + config_extra = getattr(getattr(self, "config", None), "extra", {}) or {} + bindings = config_extra.get("channel_skill_bindings", []) + if not bindings: + return None + ids_to_check = [channel_id] + if parent_id: + ids_to_check.append(parent_id) + for binding_id in ids_to_check: + for entry in bindings: + if not isinstance(entry, dict): + continue + if str(entry.get("id", "")).strip() != binding_id: + continue + skills = entry.get("skills") or entry.get("skill") + if isinstance(skills, str): + skills = [skills] + if isinstance(skills, list): + normalized = [str(skill).strip() for skill in skills if str(skill).strip()] + return list(dict.fromkeys(normalized)) or None + return None + + def _resolve_channel_prompt(self, channel_id: str, parent_id: str | None = None) -> str | None: + from gateway.platforms.base import resolve_channel_prompt + + config_extra = getattr(getattr(self, "config", None), "extra", {}) or {} + return resolve_channel_prompt(config_extra, channel_id, parent_id) + async def _process_inbound_message( self, *, @@ -3148,6 +3350,15 @@ async def _process_inbound_message( ) chat_id = getattr(message, "chat_id", "") or "" + thread_context_id = await self._resolve_thread_context_id( + message=message, + message_id=message_id, + fetch_current_message=True, + ) + channel_binding_id = thread_context_id or chat_id + parent_binding_id = chat_id if thread_context_id else None + auto_skill = self._resolve_channel_skills(channel_binding_id, parent_binding_id) + channel_prompt = self._resolve_channel_prompt(channel_binding_id, parent_binding_id) chat_info = await self.get_chat_info(chat_id) sender_profile = await self._resolve_sender_profile(sender_id, is_bot=is_bot) source = self.build_source( @@ -3156,7 +3367,7 @@ async def _process_inbound_message( chat_type=self._resolve_source_chat_type(chat_info=chat_info, event_chat_type=chat_type), user_id=sender_profile["user_id"], user_name=sender_profile["user_name"], - thread_id=thread_id, + thread_id=thread_context_id, user_id_alt=sender_profile["user_id_alt"], is_bot=is_bot, ) @@ -3170,6 +3381,8 @@ async def _process_inbound_message( media_types=media_types, reply_to_message_id=reply_to_message_id, reply_to_text=reply_to_text, + auto_skill=auto_skill, + channel_prompt=channel_prompt, timestamp=datetime.now(), ) await self._dispatch_inbound_event(normalized) @@ -4176,6 +4389,21 @@ def _allow_group_message( return bool(sender_ids and (sender_ids & self._allowed_group_users)) + def _resolve_group_policy_rule(self, chat_id: str = "") -> tuple[str, set[str], set[str]]: + rule = self._group_rules.get(chat_id) if chat_id else None + if rule: + return rule.policy, rule.allowlist, rule.blacklist + return self._default_group_policy or self._group_policy, self._allowed_group_users, set() + + def _should_accept_group_message(self, message: Any, sender_id: Any, chat_id: str = "") -> bool: + """Gate group messages by policy first, then mention requirement when needed.""" + if not self._allow_group_message(sender_id, chat_id): + return False + policy, _, _ = self._resolve_group_policy_rule(chat_id) + if policy == "open" or not self._require_mention: + return True + return self._mentions_self(message) + # --- Mention detection ---------------------------------------------------- def _mentions_self(self, message: Any) -> bool: @@ -4372,12 +4600,13 @@ def _is_duplicate(self, message_id: str) -> bool: # ========================================================================= def _build_outbound_payload(self, content: str) -> tuple[str, str]: - # Feishu post-type 'md' elements do not render markdown tables; sending - # table content as post causes the message to appear blank on the client. - # Force plain text for anything that looks like a markdown table. + # Feishu post-type 'md' elements do not render markdown tables reliably. + # Convert just the table blocks to fenced code blocks so the rest of the + # reply still renders as rich markdown instead of downgrading everything + # to plain text. if _MARKDOWN_TABLE_RE.search(content): - text_payload = {"text": content} - return "text", json.dumps(text_payload, ensure_ascii=False) + content = _convert_markdown_tables_to_code_blocks(content) + return "post", _build_markdown_post_payload(content) if _MARKDOWN_HINT_RE.search(content): return "post", _build_markdown_post_payload(content) text_payload = {"text": content} @@ -4456,10 +4685,11 @@ async def _send_raw_message( reply_to: Optional[str], metadata: Optional[Dict[str, Any]], ) -> Any: + metadata = metadata or {} effective_reply_to = reply_to - if not effective_reply_to and metadata and metadata.get("thread_id"): - effective_reply_to = metadata.get("reply_to_message_id") - reply_in_thread = bool((metadata or {}).get("thread_id")) + if not effective_reply_to and metadata.get("thread_id"): + effective_reply_to = metadata.get("reply_to_message_id") or metadata.get("thread_id") + reply_in_thread = bool(metadata.get("thread_id")) if effective_reply_to: body = self._build_reply_message_body( content=payload, diff --git a/gateway/run.py b/gateway/run.py index 14dc362a4da6e..9f1d0375f21c1 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1701,6 +1701,30 @@ def _format_gateway_process_notification(evt: dict) -> "str | None": return None +def _history_has_conversation_messages(history: List[Dict[str, Any]]) -> bool: + """Return True when transcript history already contains real conversation.""" + for msg in history or []: + role = msg.get("role") + if not role or role in ("session_meta", "system"): + continue + if role == "tool": + return True + if msg.get("tool_calls") or msg.get("tool_call_id"): + return True + if msg.get("content"): + return True + return False + + +def _should_bootstrap_auto_skills(session_entry, history: List[Dict[str, Any]]) -> bool: + """Return True when bound auto-skills still need first-turn injection.""" + is_new_session = ( + session_entry.created_at == session_entry.updated_at + or getattr(session_entry, "was_auto_reset", False) + ) + return is_new_session or not _history_has_conversation_messages(history) + + # Module-level weak reference to the active GatewayRunner instance. # Used by tools (e.g. send_message) that need to route through a live # adapter for plugin platforms. Set in GatewayRunner.__init__(). @@ -2766,7 +2790,11 @@ def _resolve_turn_agent_config(self, user_message: str, model: str, runtime_kwar return route try: - overrides = resolve_fast_mode_overrides(route["model"]) + overrides = resolve_fast_mode_overrides( + route["model"], + provider=runtime.get("provider"), + api_mode=runtime.get("api_mode"), + ) except Exception: overrides = None route["request_overrides"] = overrides or {} @@ -8986,6 +9014,7 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g self._set_session_reasoning_override(session_key, None) if hasattr(self, "_pending_model_notes"): self._pending_model_notes.pop(session_key, None) + history = self.session_store.load_transcript(session_entry.session_id) # Emit session:start for new or auto-reset sessions _is_new_session = ( @@ -9088,10 +9117,12 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g # Auto-load skill(s) for topic/channel bindings (Telegram DM Topics, # Discord channel_skill_bindings). Supports a single name or ordered list. - # Only inject on NEW sessions — ongoing conversations already have the - # skill content in their conversation history from the first message. + # Inject on session bootstrap. Most sessions are detected via timestamp + # metadata, but restarts can leave behind an existing session entry with + # no conversation history yet. _auto = getattr(event, "auto_skill", None) - if _is_new_session and _auto: + _needs_auto_skill_bootstrap = _should_bootstrap_auto_skills(session_entry, history) + if _needs_auto_skill_bootstrap and _auto: _skill_names = [_auto] if isinstance(_auto, str) else list(_auto) try: from agent.skill_commands import _load_skill_payload, _build_skill_message @@ -9122,9 +9153,6 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g except Exception as e: logger.warning("[Gateway] Failed to auto-load skill(s) %s: %s", _skill_names, e) - # Load conversation history from transcript - history = self.session_store.load_transcript(session_entry.session_id) - # ----------------------------------------------------------------- # Session hygiene: auto-compress pathologically large transcripts # @@ -13020,9 +13048,9 @@ def _save_config_key(key_path: str, value): return t("gateway.reasoning.set_session", effort=effort) async def _handle_fast_command(self, event: MessageEvent) -> str: - """Handle /fast — mirror the CLI Priority Processing toggle in gateway chats.""" + """Handle /fast — mirror the CLI fast-mode toggle in gateway chats.""" import yaml - from hermes_cli.models import model_supports_fast_mode + from hermes_cli.models import _is_anthropic_fast_model, model_supports_fast_mode, runtime_supports_priority_processing args = event.get_command_args().strip().lower() config_path = _hermes_home / "config.yaml" @@ -13030,8 +13058,16 @@ async def _handle_fast_command(self, event: MessageEvent) -> str: user_config = _load_gateway_config() model = _resolve_gateway_model(user_config) - if not model_supports_fast_mode(model): + try: + runtime_kwargs = _resolve_runtime_agent_kwargs() + except Exception: + runtime_kwargs = {} + if not model_supports_fast_mode(model) and not runtime_supports_priority_processing( + runtime_kwargs.get("provider"), + runtime_kwargs.get("api_mode"), + ): return t("gateway.fast.not_supported") + feature_name = "Anthropic Fast Mode" if _is_anthropic_fast_model(model) else "Priority Processing" def _save_config_key(key_path: str, value): """Save a dot-separated key to config.yaml.""" @@ -13055,7 +13091,7 @@ def _save_config_key(key_path: str, value): if not args or args == "status": status = t("gateway.fast.status_fast") if self._service_tier == "priority" else t("gateway.fast.status_normal") - return t("gateway.fast.status", mode=status) + return t("gateway.fast.status", feature=feature_name, mode=status) if args in {"fast", "on"}: self._service_tier = "priority" @@ -13069,8 +13105,8 @@ def _save_config_key(key_path: str, value): return t("gateway.fast.unknown_arg", arg=args) if _save_config_key("agent.service_tier", saved_value): - return t("gateway.fast.saved", label=label) - return t("gateway.fast.session_only", label=label) + return t("gateway.fast.saved", feature=feature_name, label=label) + return t("gateway.fast.session_only", feature=feature_name, label=label) async def _handle_yolo_command(self, event: MessageEvent) -> Union[str, EphemeralReply]: """Handle /yolo — toggle dangerous command approval bypass for this session only.""" diff --git a/hermes_cli/models.py b/hermes_cli/models.py index b9b3c819c16ca..e64e5166f9c39 100644 --- a/hermes_cli/models.py +++ b/hermes_cli/models.py @@ -1919,7 +1919,30 @@ def _is_anthropic_fast_model(model_id: Optional[str]) -> bool: return "opus-4-6" in base or "opus-4.6" in base -def resolve_fast_mode_overrides(model_id: Optional[str]) -> dict[str, Any] | None: +def runtime_supports_priority_processing( + provider: Optional[str] = None, + api_mode: Optional[str] = None, +) -> bool: + """Return True for OpenAI/Codex-compatible runtimes that accept service_tier. + + This covers saved custom providers that point at an OpenAI-compatible or + Codex Responses-compatible endpoint even when their model name is not in + the public OpenAI Priority Processing model table. + """ + provider_key = str(provider or "").strip().lower() + api_mode_key = str(api_mode or "").strip().lower() + if provider_key in {"openai", "openai-codex"}: + return True + if provider_key.startswith("custom"): + return api_mode_key in {"chat_completions", "responses", "codex_responses"} + return False + + +def resolve_fast_mode_overrides( + model_id: Optional[str], + provider: Optional[str] = None, + api_mode: Optional[str] = None, +) -> dict[str, Any] | None: """Return request_overrides for fast/priority mode, or None if unsupported. Returns provider-appropriate overrides: @@ -1930,10 +1953,10 @@ def resolve_fast_mode_overrides(model_id: Optional[str]) -> dict[str, Any] | Non ``_build_api_kwargs`` in run_agent.py — each API path handles its own keys (service_tier for OpenAI/Codex, speed for Anthropic Messages). """ - if not model_supports_fast_mode(model_id): - return None if _is_anthropic_fast_model(model_id): return {"speed": "fast"} + if not model_supports_fast_mode(model_id) and not runtime_supports_priority_processing(provider, api_mode): + return None return {"service_tier": "priority"} diff --git a/locales/en.yaml b/locales/en.yaml index 1516977ccb68b..577fb7f975a56 100644 --- a/locales/en.yaml +++ b/locales/en.yaml @@ -123,11 +123,11 @@ gateway: denied_plural: "❌ Commands denied ({count} commands)." fast: - not_supported: "⚡ /fast is only available for OpenAI models that support Priority Processing." - status: "⚡ Priority Processing\n\nCurrent mode: `{mode}`\n\n_Usage:_ `/fast `" + not_supported: "⚡ /fast is only available for models that support fast mode (OpenAI Priority Processing, OpenAI/Codex-compatible custom providers, or Anthropic Fast Mode)." + status: "⚡ {feature}\n\nCurrent mode: `{mode}`\n\n_Usage:_ `/fast `" unknown_arg: "⚠️ Unknown argument: `{arg}`\n\n**Valid options:** normal, fast, status" - saved: "⚡ ✓ Priority Processing: **{label}** (saved to config)\n_(takes effect on next message)_" - session_only: "⚡ ✓ Priority Processing: **{label}** (this session only)" + saved: "⚡ ✓ {feature}: **{label}** (saved to config)\n_(takes effect on next message)_" + session_only: "⚡ ✓ {feature}: **{label}** (this session only)" label_fast: "FAST" label_normal: "NORMAL" status_fast: "fast" diff --git a/tests/agent/test_auxiliary_client.py b/tests/agent/test_auxiliary_client.py index 074372d1c6d47..0ce7905bbeb5e 100644 --- a/tests/agent/test_auxiliary_client.py +++ b/tests/agent/test_auxiliary_client.py @@ -29,6 +29,11 @@ _resolve_auto, _resolve_xai_oauth_for_aux, _CodexCompletionsAdapter, + _get_cached_client, + _client_cache, + _client_cache_key, + CodexAuxiliaryClient, + AsyncCodexAuxiliaryClient, ) @@ -2871,6 +2876,174 @@ def create(self, **kwargs): assert time.monotonic() - started < 0.14 + def test_get_cached_client_rebuilds_closed_codex_wrapper(self): + class FakeHTTPXClient: + def __init__(self, is_closed=False): + self.is_closed = is_closed + + class FakeOpenAIClient: + def __init__(self, is_closed=False): + self._client = FakeHTTPXClient(is_closed=is_closed) + + closed_client = SimpleNamespace( + _real_client=FakeOpenAIClient(is_closed=True), + base_url="https://example.test/v1", + ) + fresh_client = SimpleNamespace( + _real_client=FakeOpenAIClient(is_closed=False), + base_url="https://example.test/v1", + ) + + _client_cache.clear() + try: + with patch("agent.auxiliary_client.resolve_provider_client", return_value=(closed_client, "old-model")): + first_client, first_model = _get_cached_client( + "custom", "gpt-5.5", base_url="https://example.test/v1", api_key="key" + ) + assert first_client is closed_client + assert first_model == "gpt-5.5" + + with patch("agent.auxiliary_client.resolve_provider_client", return_value=(fresh_client, "new-model")) as resolver: + second_client, second_model = _get_cached_client( + "custom", "gpt-5.5", base_url="https://example.test/v1", api_key="key" + ) + + resolver.assert_called_once() + assert second_client is fresh_client + assert second_model == "gpt-5.5" + finally: + _client_cache.clear() + + def test_get_cached_client_reuses_open_codex_wrapper(self): + class FakeHTTPXClient: + is_closed = False + + cached_client = SimpleNamespace( + _real_client=SimpleNamespace(_client=FakeHTTPXClient()), + base_url="https://example.test/v1", + ) + + _client_cache.clear() + try: + with patch("agent.auxiliary_client.resolve_provider_client", return_value=(cached_client, "cached-model")): + first_client, _ = _get_cached_client( + "custom", "gpt-5.5", base_url="https://example.test/v1", api_key="key" + ) + with patch("agent.auxiliary_client.resolve_provider_client") as resolver: + second_client, _ = _get_cached_client( + "custom", "gpt-5.5", base_url="https://example.test/v1", api_key="key" + ) + + resolver.assert_not_called() + assert first_client is cached_client + assert second_client is cached_client + finally: + _client_cache.clear() + + def test_async_store_race_replaces_wrong_loop_cached_client(self): + class FakeLoop: + def __init__(self, closed=False): + self._closed = closed + + def is_closed(self): + return self._closed + + class FakeClient: + def __init__(self, name): + self.name = name + self.close_called = False + + def close(self): + self.close_called = True + + provider = "custom" + model = "gpt-5.5" + base_url = "https://example.test/v1" + api_key = "key" + current_loop = FakeLoop() + other_loop = FakeLoop() + stale_client = FakeClient("stale") + fresh_client = FakeClient("fresh") + cache_key = _client_cache_key( + provider, + async_mode=True, + base_url=base_url, + api_key=api_key, + ) + + def populate_racing_entry(*args, **kwargs): + _client_cache[cache_key] = (stale_client, "stale-model", other_loop) + return fresh_client, "fresh-model" + + _client_cache.clear() + try: + with patch("asyncio.get_event_loop", return_value=current_loop), \ + patch("agent.auxiliary_client.resolve_provider_client", side_effect=populate_racing_entry): + client, resolved_model = _get_cached_client( + provider, + model, + async_mode=True, + base_url=base_url, + api_key=api_key, + ) + + assert client is fresh_client + assert resolved_model == model + assert _client_cache[cache_key] == (fresh_client, "fresh-model", current_loop) + finally: + _client_cache.clear() + + def test_get_cached_client_rebuilds_closed_async_codex_wrapper(self): + class FakeHTTPXClient: + is_closed = True + + class FakeOpenAIClient: + api_key = "key" + base_url = "https://example.test/v1" + + def __init__(self): + self._client = FakeHTTPXClient() + + def close(self): + pass + + provider = "custom" + model = "gpt-5.5" + base_url = "https://example.test/v1" + api_key = "key" + current_loop = SimpleNamespace(is_closed=lambda: False) + closed_async = AsyncCodexAuxiliaryClient(CodexAuxiliaryClient(FakeOpenAIClient(), model)) + fresh_client = SimpleNamespace(base_url=base_url) + + _client_cache.clear() + try: + with patch("asyncio.get_event_loop", return_value=current_loop), \ + patch("agent.auxiliary_client.resolve_provider_client", return_value=(closed_async, "old-model")): + first_client, _ = _get_cached_client( + provider, + model, + async_mode=True, + base_url=base_url, + api_key=api_key, + ) + assert first_client is closed_async + + with patch("asyncio.get_event_loop", return_value=current_loop), \ + patch("agent.auxiliary_client.resolve_provider_client", return_value=(fresh_client, "fresh-model")) as resolver: + second_client, resolved_model = _get_cached_client( + provider, + model, + async_mode=True, + base_url=base_url, + api_key=api_key, + ) + + resolver.assert_called_once() + assert second_client is fresh_client + assert resolved_model == model + finally: + _client_cache.clear() + class TestCodexAuxiliaryAdapterNullOutputRecovery: def test_recovers_output_item_when_terminal_event_has_null_output(self): diff --git a/tests/agent/transports/test_codex_transport.py b/tests/agent/transports/test_codex_transport.py index 5d8aa6ba12b88..893024a7669c7 100644 --- a/tests/agent/transports/test_codex_transport.py +++ b/tests/agent/transports/test_codex_transport.py @@ -377,6 +377,30 @@ def test_unknown(self, transport): class TestCodexNormalizeResponse: + def _response_with_message_items(self, *items, output_text=None): + kwargs = { + "output": list(items), + "status": "completed", + "incomplete_details": None, + "usage": SimpleNamespace(input_tokens=10, output_tokens=5, + input_tokens_details=None, output_tokens_details=None), + } + if output_text is not None: + kwargs["output_text"] = output_text + return SimpleNamespace(**kwargs) + + def _message_item(self, text, *, phase=None, item_id="msg_abc"): + kwargs = { + "type": "message", + "role": "assistant", + "id": item_id, + "content": [SimpleNamespace(type="output_text", text=text)], + "status": "completed", + } + if phase is not None: + kwargs["phase"] = phase + return SimpleNamespace(**kwargs) + def test_text_response(self, transport): """Normalize a simple text Codex response.""" r = SimpleNamespace( @@ -398,6 +422,100 @@ def test_text_response(self, transport): assert nr.content == "Hello world" assert nr.finish_reason == "stop" + @pytest.mark.parametrize("phase", ["commentary", "analysis"]) + def test_commentary_or_analysis_only_message_is_not_visible_and_incomplete( + self, transport, phase + ): + r = self._response_with_message_items( + self._message_item("Internal commentary", phase=phase), + output_text="Internal commentary", + ) + + nr = transport.normalize_response(r) + + assert nr.content == "" + assert nr.finish_reason == "incomplete" + assert nr.provider_data["codex_message_items"] == [ + { + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "Internal commentary"}], + "id": "msg_abc", + "phase": phase, + } + ] + + @pytest.mark.parametrize("phase", ["final_answer", "final", " Final "]) + def test_final_answer_message_remains_visible_and_stops(self, transport, phase): + r = self._response_with_message_items( + self._message_item("Visible answer", phase=phase), + ) + + nr = transport.normalize_response(r) + + assert nr.content == "Visible answer" + assert nr.finish_reason == "stop" + + def test_unknown_phase_message_is_hidden_and_incomplete(self, transport): + r = self._response_with_message_items( + self._message_item("Unknown phase text", phase="future_phase"), + output_text="Unknown phase text", + ) + + nr = transport.normalize_response(r) + + assert nr.content == "" + assert nr.finish_reason == "incomplete" + assert nr.codex_message_items[0]["phase"] == "future_phase" + + def test_mixed_commentary_and_final_answer_only_shows_final(self, transport): + r = self._response_with_message_items( + self._message_item( + "Internal commentary", + phase="commentary", + item_id="msg_commentary", + ), + self._message_item( + "Visible answer", + phase="final_answer", + item_id="msg_final", + ), + ) + + nr = transport.normalize_response(r) + + assert nr.content == "Visible answer" + assert nr.finish_reason == "stop" + assert nr.codex_message_items == [ + { + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "Internal commentary"}], + "id": "msg_commentary", + "phase": "commentary", + }, + { + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "Visible answer"}], + "id": "msg_final", + "phase": "final_answer", + }, + ] + + def test_no_phase_message_remains_visible_for_compatibility(self, transport): + r = self._response_with_message_items( + self._message_item("Legacy visible answer", phase=None), + ) + + nr = transport.normalize_response(r) + + assert nr.content == "Legacy visible answer" + assert nr.finish_reason == "stop" + def test_message_items_preserved_in_provider_data(self, transport): """Codex assistant message item ids/phases must survive transport normalization.""" r = SimpleNamespace( diff --git a/tests/cli/test_fast_command.py b/tests/cli/test_fast_command.py index 7745737c45416..b1583248e432e 100644 --- a/tests/cli/test_fast_command.py +++ b/tests/cli/test_fast_command.py @@ -334,6 +334,17 @@ def test_resolve_overrides_returns_service_tier_for_openai(self): result = resolve_fast_mode_overrides("gpt-5.4") assert result == {"service_tier": "priority"} + def test_resolve_overrides_returns_service_tier_for_custom_codex_runtime(self): + from hermes_cli.models import resolve_fast_mode_overrides, runtime_supports_priority_processing + + assert runtime_supports_priority_processing("custom", "codex_responses") is True + result = resolve_fast_mode_overrides( + "gpt-5.3-codex", + provider="custom", + api_mode="codex_responses", + ) + assert result == {"service_tier": "priority"} + def test_is_anthropic_fast_model(self): """The speed=fast parameter is Opus 4.6 only — other Claude excluded.""" from hermes_cli.models import _is_anthropic_fast_model diff --git a/tests/cron/test_cron_delivery_live_adapter.py b/tests/cron/test_cron_delivery_live_adapter.py new file mode 100644 index 0000000000000..5d479c703048a --- /dev/null +++ b/tests/cron/test_cron_delivery_live_adapter.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +import asyncio +import threading +from types import SimpleNamespace + + +class FakeReplyMessageResponse: + """SDK-style response object: truthy, but deliberately no .get().""" + + pass + + +def _start_background_loop(): + loop = asyncio.new_event_loop() + ready = threading.Event() + + def run_loop(): + asyncio.set_event_loop(loop) + ready.set() + loop.run_forever() + + thread = threading.Thread(target=run_loop, daemon=True) + thread.start() + ready.wait(timeout=5) + return loop, thread + + +def _stop_background_loop(loop, thread): + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=5) + loop.close() + + +def test_live_adapter_sdk_raw_response_does_not_fallback(monkeypatch): + """A successful Feishu live-adapter send must not fallback just because + raw_response is an SDK object rather than a dict. + + Regression coverage for duplicate cron deliveries: the message.reply call + can succeed, then scheduler diagnostics used raw_response.get(...), raised + AttributeError, and sent the same content again via standalone fallback. + """ + from cron import scheduler + from gateway.config import Platform + import gateway.config as gateway_config + import tools.send_message_tool as send_message_tool + + monkeypatch.setattr(scheduler, "load_config", lambda: {"cron": {"wrap_response": False}}) + monkeypatch.setattr( + scheduler, + "_resolve_delivery_targets", + lambda job: [{"platform": "feishu", "chat_id": "oc_test", "thread_id": "om_test"}], + ) + monkeypatch.setattr( + gateway_config, + "load_gateway_config", + lambda: SimpleNamespace(platforms={Platform.FEISHU: SimpleNamespace(enabled=True)}), + ) + + fallback_calls = [] + + async def fake_send_to_platform(*args, **kwargs): + fallback_calls.append((args, kwargs)) + return {"error": "fallback should not be called"} + + monkeypatch.setattr(send_message_tool, "_send_to_platform", fake_send_to_platform) + + live_calls = [] + + class FakeLiveAdapter: + async def send(self, chat_id, content, metadata=None): + live_calls.append((chat_id, content, metadata)) + return SimpleNamespace(success=True, raw_response=FakeReplyMessageResponse()) + + loop, thread = _start_background_loop() + try: + error = scheduler._deliver_result( + {"id": "job1", "deliver": "origin", "name": "test"}, + "hello", + adapters={Platform.FEISHU: FakeLiveAdapter()}, + loop=loop, + ) + finally: + _stop_background_loop(loop, thread) + + assert error is None + assert len(live_calls) == 1 + assert live_calls[0][2] == {"thread_id": "om_test"} + assert fallback_calls == [] + + +def test_raw_response_get_handles_dict_and_sdk_object(): + from cron.scheduler import _raw_response_get + + assert _raw_response_get({"thread_fallback": True}, "thread_fallback") is True + assert _raw_response_get(FakeReplyMessageResponse(), "thread_fallback") is None diff --git a/tests/gateway/test_auto_skill_bootstrap.py b/tests/gateway/test_auto_skill_bootstrap.py new file mode 100644 index 0000000000000..e93844bdb3c33 --- /dev/null +++ b/tests/gateway/test_auto_skill_bootstrap.py @@ -0,0 +1,41 @@ +from datetime import datetime, timedelta +from types import SimpleNamespace + +from gateway.run import _history_has_conversation_messages, _should_bootstrap_auto_skills + + +def test_history_has_conversation_messages_ignores_metadata_only_entries(): + history = [ + {"role": "session_meta", "content": "metadata"}, + {"role": "system", "content": "system prompt"}, + ] + + assert not _history_has_conversation_messages(history) + + +def test_history_has_conversation_messages_detects_real_user_content(): + assert _history_has_conversation_messages([{"role": "user", "content": "hello"}]) + + +def test_should_bootstrap_auto_skills_for_existing_metadata_only_session(): + now = datetime.now() + session_entry = SimpleNamespace( + created_at=now - timedelta(minutes=1), + updated_at=now, + was_auto_reset=False, + ) + history = [{"role": "session_meta", "content": "created"}] + + assert _should_bootstrap_auto_skills(session_entry, history) + + +def test_should_not_bootstrap_auto_skills_when_conversation_exists(): + now = datetime.now() + session_entry = SimpleNamespace( + created_at=now - timedelta(minutes=1), + updated_at=now, + was_auto_reset=False, + ) + history = [{"role": "assistant", "content": "loaded"}] + + assert not _should_bootstrap_auto_skills(session_entry, history) diff --git a/tests/gateway/test_config.py b/tests/gateway/test_config.py index 4f2756dba3b16..549c5f7b7f8f4 100644 --- a/tests/gateway/test_config.py +++ b/tests/gateway/test_config.py @@ -652,6 +652,31 @@ def test_bridges_slack_channel_prompts_from_config_yaml(self, tmp_path, monkeypa "C01ABC": "Code review mode", } + def test_bridges_feishu_channel_skill_bindings_from_config_yaml(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + config_path = hermes_home / "config.yaml" + config_path.write_text( + "feishu:\n" + " channel_skill_bindings:\n" + " - id: oc_chat\n" + " skill: code\n" + " - id: omt_thread\n" + " skills:\n" + " - review\n" + " - test\n", + encoding="utf-8", + ) + + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + config = load_gateway_config() + + assert config.platforms[Platform.FEISHU].extra["channel_skill_bindings"] == [ + {"id": "oc_chat", "skill": "code"}, + {"id": "omt_thread", "skills": ["review", "test"]}, + ] + def test_bridges_feishu_allow_bots_from_config_yaml_to_env(self, tmp_path, monkeypatch): hermes_home = tmp_path / ".hermes" hermes_home.mkdir() diff --git a/tests/gateway/test_fast_command.py b/tests/gateway/test_fast_command.py index 58db9faf05e34..df22866958e04 100644 --- a/tests/gateway/test_fast_command.py +++ b/tests/gateway/test_fast_command.py @@ -121,6 +121,54 @@ def test_turn_route_skips_priority_processing_for_unsupported_models(): assert route["request_overrides"] == {} +def test_turn_route_injects_speed_for_anthropic_fast_mode(): + runner = _make_runner() + runner._service_tier = "priority" + runtime_kwargs = { + "api_key": "sk-ant-test", + "base_url": "https://api.anthropic.com", + "provider": "anthropic", + "api_mode": "anthropic_messages", + "command": None, + "args": [], + "credential_pool": None, + } + + route = gateway_run.GatewayRunner._resolve_turn_agent_config( + runner, + "hi", + "claude-opus-4-6", + runtime_kwargs, + ) + + assert route["runtime"]["provider"] == "anthropic" + assert route["request_overrides"] == {"speed": "fast"} + + +def test_turn_route_injects_priority_for_custom_codex_runtime(): + runner = _make_runner() + runner._service_tier = "priority" + runtime_kwargs = { + "api_key": "***", + "base_url": "https://codex.example/v1", + "provider": "custom", + "api_mode": "codex_responses", + "command": None, + "args": [], + "credential_pool": None, + } + + route = gateway_run.GatewayRunner._resolve_turn_agent_config( + runner, + "hi", + "gpt-5.3-codex", + runtime_kwargs, + ) + + assert route["runtime"]["provider"] == "custom" + assert route["request_overrides"] == {"service_tier": "priority"} + + @pytest.mark.asyncio async def test_handle_fast_command_persists_config(monkeypatch, tmp_path): runner = _make_runner() @@ -138,6 +186,55 @@ async def test_handle_fast_command_persists_config(monkeypatch, tmp_path): assert saved["agent"]["service_tier"] == "fast" +@pytest.mark.asyncio +async def test_handle_fast_command_accepts_anthropic_fast_mode(monkeypatch, tmp_path): + runner = _make_runner() + + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {}) + monkeypatch.setattr(gateway_run, "_resolve_gateway_model", lambda config=None: "claude-opus-4-6") + + response = await runner._handle_fast_command(_make_event("/fast fast")) + + assert "Anthropic Fast Mode" in response + assert "FAST" in response + assert runner._service_tier == "priority" + + +@pytest.mark.asyncio +async def test_handle_fast_command_accepts_custom_codex_provider(monkeypatch, tmp_path): + runner = _make_runner() + + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {}) + monkeypatch.setattr(gateway_run, "_resolve_gateway_model", lambda config=None: "gpt-5.3-codex") + monkeypatch.setattr( + gateway_run, + "_resolve_runtime_agent_kwargs", + lambda: {"provider": "custom", "api_mode": "codex_responses", "base_url": "https://codex.example/v1"}, + ) + + response = await runner._handle_fast_command(_make_event("/fast fast")) + + assert "Priority Processing" in response + assert "FAST" in response + assert runner._service_tier == "priority" + + +@pytest.mark.asyncio +async def test_handle_fast_command_unsupported_message_mentions_both_fast_modes(monkeypatch, tmp_path): + runner = _make_runner() + + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {}) + monkeypatch.setattr(gateway_run, "_resolve_gateway_model", lambda config=None: "deepseek-chat") + + response = await runner._handle_fast_command(_make_event("/fast fast")) + + assert "OpenAI Priority Processing" in response + assert "Anthropic Fast Mode" in response + + @pytest.mark.asyncio async def test_run_agent_passes_priority_processing_to_gateway_agent(monkeypatch, tmp_path): _install_fake_agent(monkeypatch) diff --git a/tests/gateway/test_feishu.py b/tests/gateway/test_feishu.py index 4d78b454b0ca7..b95f57fef5c64 100644 --- a/tests/gateway/test_feishu.py +++ b/tests/gateway/test_feishu.py @@ -161,6 +161,36 @@ def test_normalize_interactive_card_preserves_title_body_and_actions(self): class TestFeishuAdapterMessaging(unittest.TestCase): + def test_markdown_table_reply_stays_rich_post_with_table_as_code_block(self): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + + adapter = FeishuAdapter(PlatformConfig()) + content = "**结果**\n\n| 模型 | GSM8K |\n|---|---:|\n| GPT-4 | 95.5 |\n\n- 结论：下降" + + msg_type, payload = adapter._build_outbound_payload(content) + data = json.loads(payload) + rows = data["zh_cn"]["content"] + rendered = "\n".join(part["text"] for row in rows for part in row if part.get("tag") == "md") + + self.assertEqual(msg_type, "post") + self.assertIn("**结果**", rendered) + self.assertIn("```text", rendered) + self.assertIn("| 模型 | GSM8K |", rendered) + self.assertIn("```", rendered) + self.assertIn("- 结论：下降", rendered) + + def test_plain_text_reply_remains_text_message(self): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + + adapter = FeishuAdapter(PlatformConfig()) + + msg_type, payload = adapter._build_outbound_payload("hello world") + + self.assertEqual(msg_type, "text") + self.assertEqual(json.loads(payload), {"text": "hello world"}) + @patch.dict(os.environ, { "FEISHU_APP_ID": "cli_app", "FEISHU_APP_SECRET": "secret_app", @@ -765,7 +795,7 @@ def test_reaction_on_our_own_bot_message_is_routed(self): adapter._handle_message_with_guards.assert_awaited_once() @patch.dict(os.environ, {"FEISHU_GROUP_POLICY": "open"}, clear=True) - def test_group_message_requires_mentions_even_when_policy_open(self): + def test_group_message_policy_open_accepts_without_mentions(self): from gateway.config import PlatformConfig from gateway.platforms.feishu import FeishuAdapter @@ -778,7 +808,7 @@ def test_group_message_requires_mentions_even_when_policy_open(self): self.assertFalse(_admits_group(adapter, message_with_mention, sender_id, "")) @patch.dict(os.environ, {"FEISHU_GROUP_POLICY": "open"}, clear=True) - def test_group_message_with_other_user_mention_is_rejected_when_bot_identity_unknown(self): + def test_group_message_policy_open_accepts_other_user_mention_when_bot_identity_unknown(self): from gateway.config import PlatformConfig from gateway.platforms.feishu import FeishuAdapter @@ -1030,7 +1060,7 @@ def test_default_group_policy_fallback_for_chats_without_explicit_rule(self): ) ) - @patch.dict(os.environ, {"FEISHU_GROUP_POLICY": "open"}, clear=True) + @patch.dict(os.environ, {"FEISHU_GROUP_POLICY": "allowlist", "FEISHU_ALLOWED_USERS": "ou_any"}, clear=True) def test_group_message_matches_bot_open_id_when_configured(self): from gateway.config import PlatformConfig from gateway.platforms.feishu import FeishuAdapter @@ -1055,7 +1085,7 @@ def test_group_message_matches_bot_open_id_when_configured(self): _admits_group(adapter, SimpleNamespace(mentions=[other_mention]), sender_id, "") ) - @patch.dict(os.environ, {"FEISHU_GROUP_POLICY": "open"}, clear=True) + @patch.dict(os.environ, {"FEISHU_GROUP_POLICY": "allowlist", "FEISHU_ALLOWED_USERS": "ou_any"}, clear=True) def test_group_message_matches_bot_name_when_only_name_available(self): """Name fallback engages when either side lacks an open_id. When BOTH the mention and the bot carry open_ids, IDs are authoritative — a @@ -1952,6 +1982,271 @@ def test_process_inbound_message_fetches_reply_to_text(self): self.assertEqual(event.reply_to_message_id, "om_parent") self.assertEqual(event.reply_to_text, "父消息内容") + @patch.dict(os.environ, {}, clear=True) + def test_process_inbound_message_uses_upper_message_id_as_thread_context(self): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + + adapter = FeishuAdapter(PlatformConfig()) + adapter._dispatch_inbound_event = AsyncMock() + adapter.get_chat_info = AsyncMock( + return_value={"chat_id": "oc_group", "name": "Feishu Group", "type": "group"} + ) + adapter._resolve_sender_profile = AsyncMock( + return_value={"user_id": "ou_user", "user_name": "张三", "user_id_alt": None} + ) + adapter._fetch_message_text = AsyncMock(return_value="根消息内容") + message = SimpleNamespace( + chat_id="oc_group", + thread_id=None, + parent_id="om_parent_reply", + upper_message_id="om_thread_root", + message_type="text", + content='{"text":"thread reply"}', + message_id="om_thread_child", + ) + + asyncio.run( + adapter._process_inbound_message( + data=SimpleNamespace(event=SimpleNamespace(message=message)), + message=message, + sender_id=SimpleNamespace(open_id="ou_user", user_id=None, union_id=None), + chat_type="group", + message_id="om_thread_child", + ) + ) + + event = adapter._dispatch_inbound_event.await_args.args[0] + self.assertEqual(event.source.thread_id, "om_thread_root") + self.assertEqual(event.reply_to_message_id, "om_parent_reply") + self.assertEqual(event.reply_to_text, "根消息内容") + + @patch.dict(os.environ, {}, clear=True) + def test_process_inbound_message_uses_message_id_as_thread_root_when_feishu_omits_all_root_fields(self): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + + adapter = FeishuAdapter(PlatformConfig()) + adapter._dispatch_inbound_event = AsyncMock() + adapter.get_chat_info = AsyncMock( + return_value={"chat_id": "oc_group", "name": "Feishu Group", "type": "group"} + ) + adapter._resolve_sender_profile = AsyncMock( + return_value={"user_id": "ou_user", "user_name": "张三", "user_id_alt": None} + ) + captured = {} + + class _MessageAPI: + def get(self, request): + captured["request"] = request + return SimpleNamespace( + success=lambda: True, + data=SimpleNamespace( + items=[ + SimpleNamespace( + message_id="om_topic_root", + chat_id="oc_group", + thread_id="omt_topic_123", + root_id=None, + upper_message_id=None, + parent_id=None, + ) + ] + ), + ) + + adapter._client = SimpleNamespace(im=SimpleNamespace(v1=SimpleNamespace(message=_MessageAPI()))) + message = SimpleNamespace( + chat_id="oc_group", + root_id=None, + thread_id="omt_topic_123", + parent_id=None, + upper_message_id=None, + message_type="post", + content='{"text":"topic root"}', + message_id="om_topic_root", + ) + + async def _direct(func, *args, **kwargs): + return func(*args, **kwargs) + + with patch("gateway.platforms.feishu.asyncio.to_thread", side_effect=_direct): + asyncio.run( + adapter._process_inbound_message( + data=SimpleNamespace(event=SimpleNamespace(message=message)), + message=message, + sender_id=SimpleNamespace(open_id="ou_user", user_id=None, union_id=None), + chat_type="group", + message_id="om_topic_root", + ) + ) + + event = adapter._dispatch_inbound_event.await_args.args[0] + self.assertEqual(captured["request"].message_id, "om_topic_root") + self.assertEqual(event.source.thread_id, "om_topic_root") + + @patch.dict(os.environ, {}, clear=True) + def test_process_inbound_message_recovers_upper_message_id_when_event_only_has_omt_thread_id(self): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + + adapter = FeishuAdapter(PlatformConfig()) + adapter._dispatch_inbound_event = AsyncMock() + adapter.get_chat_info = AsyncMock( + return_value={"chat_id": "oc_group", "name": "Feishu Group", "type": "group"} + ) + adapter._resolve_sender_profile = AsyncMock( + return_value={"user_id": "ou_user", "user_name": "张三", "user_id_alt": None} + ) + captured = {} + + class _MessageAPI: + def get(self, request): + captured["request"] = request + return SimpleNamespace( + success=lambda: True, + data=SimpleNamespace( + items=[ + SimpleNamespace( + message_id="om_thread_child", + chat_id="oc_group", + thread_id="omt_topic_123", + root_id=None, + upper_message_id="om_thread_root", + parent_id="om_parent_reply", + ) + ] + ), + ) + + adapter._client = SimpleNamespace(im=SimpleNamespace(v1=SimpleNamespace(message=_MessageAPI()))) + message = SimpleNamespace( + chat_id="oc_group", + root_id=None, + thread_id="omt_topic_123", + parent_id=None, + upper_message_id=None, + message_type="text", + content='{"text":"thread reply"}', + message_id="om_thread_child", + ) + + async def _direct(func, *args, **kwargs): + return func(*args, **kwargs) + + with patch("gateway.platforms.feishu.asyncio.to_thread", side_effect=_direct): + asyncio.run( + adapter._process_inbound_message( + data=SimpleNamespace(event=SimpleNamespace(message=message)), + message=message, + sender_id=SimpleNamespace(open_id="ou_user", user_id=None, union_id=None), + chat_type="group", + message_id="om_thread_child", + ) + ) + + event = adapter._dispatch_inbound_event.await_args.args[0] + self.assertEqual(captured["request"].message_id, "om_thread_child") + self.assertEqual(event.source.thread_id, "om_thread_root") + + @patch.dict(os.environ, {}, clear=True) + def test_process_inbound_message_resolves_thread_skill_and_prompt_bindings(self): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + + adapter = FeishuAdapter( + PlatformConfig( + extra={ + "channel_skill_bindings": [ + {"id": "om_thread_root", "skills": ["code", "plan"]}, + {"id": "oc_group", "skills": ["fallback-skill"]}, + ], + "channel_prompts": { + "om_thread_root": "Use the engineering workflow.", + "oc_group": "Fallback chat prompt.", + }, + } + ) + ) + adapter._dispatch_inbound_event = AsyncMock() + adapter.get_chat_info = AsyncMock( + return_value={"chat_id": "oc_group", "name": "Hermes 群", "type": "group"} + ) + adapter._resolve_sender_profile = AsyncMock( + return_value={"user_id": "ou_user", "user_name": "张三", "user_id_alt": None} + ) + adapter._fetch_message_text = AsyncMock(return_value="根消息内容") + message = SimpleNamespace( + chat_id="oc_group", + root_id="om_thread_root", + thread_id=None, + parent_id="om_parent_reply", + upper_message_id=None, + message_type="text", + content='{"text":"请检查这个问题"}', + message_id="om_group_thread_msg", + ) + + asyncio.run( + adapter._process_inbound_message( + data=SimpleNamespace(event=SimpleNamespace(message=message)), + message=message, + sender_id=SimpleNamespace(open_id="ou_user", user_id=None, union_id=None), + chat_type="group", + message_id="om_group_thread_msg", + ) + ) + + event = adapter._dispatch_inbound_event.await_args.args[0] + self.assertEqual(event.source.thread_id, "om_thread_root") + self.assertEqual(event.auto_skill, ["code", "plan"]) + self.assertEqual(event.channel_prompt, "Use the engineering workflow.") + + @patch.dict(os.environ, {}, clear=True) + def test_process_inbound_message_falls_back_to_chat_binding_without_thread(self): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + + adapter = FeishuAdapter( + PlatformConfig( + extra={ + "channel_skill_bindings": [{"id": "oc_chat", "skill": "code"}], + "channel_prompts": {"oc_chat": "Chat-level prompt."}, + } + ) + ) + adapter._dispatch_inbound_event = AsyncMock() + adapter.get_chat_info = AsyncMock( + return_value={"chat_id": "oc_chat", "name": "Feishu DM", "type": "dm"} + ) + adapter._resolve_sender_profile = AsyncMock( + return_value={"user_id": "ou_user", "user_name": "张三", "user_id_alt": None} + ) + message = SimpleNamespace( + chat_id="oc_chat", + thread_id=None, + parent_id=None, + upper_message_id=None, + message_type="text", + content='{"text":"hello"}', + message_id="om_chat_text", + ) + + asyncio.run( + adapter._process_inbound_message( + data=SimpleNamespace(event=SimpleNamespace(message=message)), + message=message, + sender_id=SimpleNamespace(open_id="ou_user", user_id=None, union_id=None), + chat_type="p2p", + message_id="om_chat_text", + ) + ) + + event = adapter._dispatch_inbound_event.await_args.args[0] + self.assertIsNone(event.source.thread_id) + self.assertEqual(event.auto_skill, ["code"]) + self.assertEqual(event.channel_prompt, "Chat-level prompt.") + @patch.dict(os.environ, {}, clear=True) def test_send_replies_in_thread_when_thread_metadata_present(self): from gateway.config import PlatformConfig @@ -1993,6 +2288,43 @@ async def _direct(func, *args, **kwargs): self.assertEqual(result.message_id, "om_reply") self.assertTrue(captured["request"].request_body.reply_in_thread) + @patch.dict(os.environ, {}, clear=True) + def test_send_plain_reply_does_not_force_thread_reply_without_thread_metadata(self): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + + adapter = FeishuAdapter(PlatformConfig()) + captured = {} + + class _ReplyAPI: + def reply(self, request): + captured["request"] = request + return SimpleNamespace( + success=lambda: True, + data=SimpleNamespace(message_id="om_plain_reply"), + ) + + adapter._client = SimpleNamespace( + im=SimpleNamespace(v1=SimpleNamespace(message=_ReplyAPI())) + ) + + async def _direct(func, *args, **kwargs): + return func(*args, **kwargs) + + with patch("gateway.platforms.feishu.asyncio.to_thread", side_effect=_direct): + result = asyncio.run( + adapter.send( + chat_id="oc_chat", + content="plain reply", + reply_to="om_parent", + metadata=None, + ) + ) + + self.assertTrue(result.success) + self.assertEqual(result.message_id, "om_plain_reply") + self.assertFalse(captured["request"].request_body.reply_in_thread) + @patch.dict(os.environ, {}, clear=True) def test_send_uses_metadata_reply_target_for_threaded_feishu_topic(self): from gateway.config import PlatformConfig diff --git a/tests/gateway/test_tts_media_routing.py b/tests/gateway/test_tts_media_routing.py index eaf9c59280894..c669275e7d54a 100644 --- a/tests/gateway/test_tts_media_routing.py +++ b/tests/gateway/test_tts_media_routing.py @@ -121,6 +121,22 @@ async def test_base_adapter_routes_voice_tagged_telegram_ogg_media_tag_to_voice_ adapter.send_document.assert_not_awaited() +@pytest.mark.asyncio +async def test_base_adapter_preserves_unsupported_media_tag_in_visible_text(): + adapter = _MediaRoutingAdapter() + event = _event() + adapter._message_handler = AsyncMock(return_value="Review MEDIA:/tmp/build.unknownext") + adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="text")) + adapter.send_document = AsyncMock(return_value=SendResult(success=True, message_id="doc")) + + await adapter._process_message_background(event, build_session_key(event.source)) + + adapter.send.assert_awaited_once() + sent_text = adapter.send.await_args.kwargs["content"] + assert "MEDIA:/tmp/build.unknownext" in sent_text + adapter.send_document.assert_not_awaited() + + def _fake_runner(thread_meta): """Build a fake GatewayRunner-like object with the helper methods needed by _deliver_media_from_response.""" diff --git a/tests/run_agent/test_run_agent_codex_responses.py b/tests/run_agent/test_run_agent_codex_responses.py index 7f899c601d19e..aaf35cc1d748d 100644 --- a/tests/run_agent/test_run_agent_codex_responses.py +++ b/tests/run_agent/test_run_agent_codex_responses.py @@ -1348,7 +1348,12 @@ def test_normalize_codex_response_marks_commentary_only_message_as_incomplete(mo ) assert finish_reason == "incomplete" - assert "inspect the repository" in (assistant_message.content or "") + assert assistant_message.content == "" + assert assistant_message.codex_message_items[0]["phase"] == "commentary" + assert ( + "inspect the repository" + in assistant_message.codex_message_items[0]["content"][0]["text"] + ) def test_normalize_codex_response_preserves_message_status_for_replay(monkeypatch): @@ -1683,13 +1688,22 @@ def _fake_execute_tool_calls(assistant_message, messages, effective_task_id): assert result["completed"] is True assert result["final_response"] == "Architecture summary complete." + + def _has_hidden_commentary_item(msg): + items = msg.get("codex_message_items") or [{}] + text = items[0].get("content", [{}])[0].get("text", "") + return ( + msg.get("role") == "assistant" + and msg.get("finish_reason") == "incomplete" + and not (msg.get("content") or "") + and "inspect the repo structure" in text + ) + + assert any(_has_hidden_commentary_item(msg) for msg in result["messages"]) assert any( - msg.get("role") == "assistant" - and msg.get("finish_reason") == "incomplete" - and "inspect the repo structure" in (msg.get("content") or "") + msg.get("role") == "tool" and msg.get("tool_call_id") == "call_1" for msg in result["messages"] ) - assert any(msg.get("role") == "tool" and msg.get("tool_call_id") == "call_1" for msg in result["messages"]) def test_run_conversation_codex_continues_after_ack_stop_message(monkeypatch): diff --git a/tests/tools/test_send_message_tool.py b/tests/tools/test_send_message_tool.py index 63f45e1e75c6e..f737c5199c31e 100644 --- a/tests/tools/test_send_message_tool.py +++ b/tests/tools/test_send_message_tool.py @@ -29,6 +29,7 @@ def _reset_signal_scheduler(): from tools.send_message_tool import ( _is_telegram_thread_not_found, _parse_target_ref, + _send_feishu, _send_matrix_via_adapter, _send_signal, _send_telegram, @@ -240,6 +241,94 @@ def test_cron_duplicate_target_is_skipped_and_explained(self): send_mock.assert_not_awaited() mirror_mock.assert_not_called() + def test_send_feishu_routes_non_image_media_to_document(self, tmp_path, monkeypatch): + file_path = tmp_path / "report.md" + file_path.write_text("# Report\n", encoding="utf-8") + calls = [] + + class FakeFeishuAdapter: + def __init__(self, _config): + self._domain_name = "feishu" + + def _build_lark_client(self, domain): + calls.append(("build_client", domain)) + return object() + + async def send(self, chat_id, message, metadata=None): + calls.append(("send", chat_id, message, metadata)) + return SimpleNamespace(success=True, message_id="text") + + async def send_image_file(self, chat_id, media_path, metadata=None): + calls.append(("send_image_file", chat_id, media_path, metadata)) + return SimpleNamespace(success=True, message_id="image") + + async def send_document(self, chat_id, media_path, metadata=None): + calls.append(("send_document", chat_id, media_path, metadata)) + return SimpleNamespace(success=True, message_id="doc") + + fake_module = SimpleNamespace( + FEISHU_AVAILABLE=True, + FEISHU_DOMAIN="https://open.feishu.cn", + LARK_DOMAIN="https://open.larksuite.com", + FeishuAdapter=FakeFeishuAdapter, + ) + monkeypatch.setitem(sys.modules, "gateway.platforms.feishu", fake_module) + + result = asyncio.run( + _send_feishu( + SimpleNamespace(enabled=True, token="tok", extra={}), + "oc_chat", + "attached", + media_files=[(str(file_path), False)], + ) + ) + + assert result["success"] is True + assert ("send_document", "oc_chat", str(file_path), None) in calls + assert not any(call[0] == "send_image_file" for call in calls) + + def test_send_feishu_routes_image_media_to_image_upload(self, tmp_path, monkeypatch): + image_path = tmp_path / "screenshot.png" + image_path.write_bytes(b"\x89PNG\r\n\x1a\n") + calls = [] + + class FakeFeishuAdapter: + def __init__(self, _config): + self._domain_name = "feishu" + + def _build_lark_client(self, domain): + calls.append(("build_client", domain)) + return object() + + async def send_image_file(self, chat_id, media_path, metadata=None): + calls.append(("send_image_file", chat_id, media_path, metadata)) + return SimpleNamespace(success=True, message_id="image") + + async def send_document(self, chat_id, media_path, metadata=None): + calls.append(("send_document", chat_id, media_path, metadata)) + return SimpleNamespace(success=True, message_id="doc") + + fake_module = SimpleNamespace( + FEISHU_AVAILABLE=True, + FEISHU_DOMAIN="https://open.feishu.cn", + LARK_DOMAIN="https://open.larksuite.com", + FeishuAdapter=FakeFeishuAdapter, + ) + monkeypatch.setitem(sys.modules, "gateway.platforms.feishu", fake_module) + + result = asyncio.run( + _send_feishu( + SimpleNamespace(enabled=True, token="tok", extra={}), + "oc_chat", + "", + media_files=[(str(image_path), False)], + ) + ) + + assert result["success"] is True + assert ("send_image_file", "oc_chat", str(image_path), None) in calls + assert not any(call[0] == "send_document" for call in calls) + def test_resolved_telegram_topic_name_preserves_thread_id(self): config, telegram_cfg = _make_config() diff --git a/tinker-atropos b/tinker-atropos new file mode 160000 index 0000000000000..65f084ee8054a --- /dev/null +++ b/tinker-atropos @@ -0,0 +1 @@ +Subproject commit 65f084ee8054a5d02aeac76e24ed60388511c82b diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index 53a9fc6003753..00d2d676e5c62 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -147,7 +147,7 @@ async def _send_telegram_message_with_retry(bot, *, attempts: int = 3, **kwargs) }, "message": { "type": "string", - "description": "The message text to send. To send an image or file, include MEDIA: (e.g. 'MEDIA:/tmp/report.pdf') in the message — the platform will deliver it as a native media attachment." + "description": "The message text to send. To attach a local file, include MEDIA: (e.g. 'MEDIA:/tmp/hermes/cache/report.md') in the message. Hermes extracts these tags and uploads attachments natively: images use image upload, video/audio use their native routes where supported, and Feishu/Lark non-image files (md/json/html/yaml/xml/tsv/pdf/csv/xlsx/etc.) are uploaded with native document upload (send_document). Do not send a Feishu/Lark file by pasting the path as ordinary text." } }, "required": []