[codex] fix(zai): sanitize GLM system prompt at runtime

agent/chat_completion_helpers.py

@@ -30,6 +30,7 @@
 from agent.error_classifier import FailoverReason
 from agent.gemini_native_adapter import is_native_gemini_base_url
 from agent.model_metadata import is_local_endpoint
+from agent.zai_prompt_policy import apply_zai_special_prompt
 from agent.message_sanitization import (
     _sanitize_surrogates,
     _repair_tool_call_arguments,
@@ -1539,6 +1540,7 @@ def handle_max_iterations(agent, messages: list, api_call_count: int) -> str:
             effective_system = (effective_system + "\n\n" + agent.ephemeral_system_prompt).strip()
         if effective_system:
             api_messages = [{"role": "system", "content": effective_system}] + api_messages
+        api_messages = apply_zai_special_prompt(agent, api_messages)
         if agent.prefill_messages:
             sys_offset = 1 if effective_system else 0
             for idx, pfm in enumerate(agent.prefill_messages):

agent/conversation_loop.py

@@ -61,6 +61,7 @@
 from agent.retry_utils import adaptive_rate_limit_backoff, jittered_backoff
 from agent.trajectory import has_incomplete_scratchpad
 from agent.usage_pricing import estimate_usage_cost, normalize_usage
+from agent.zai_prompt_policy import apply_zai_special_prompt
 from hermes_constants import PARTIAL_STREAM_STUB_ID
 from hermes_logging import set_session_context
 from tools.skill_provenance import set_current_write_origin
@@ -844,6 +845,7 @@ def run_conversation(
             effective_system = (effective_system + "\n\n" + agent.ephemeral_system_prompt).strip()
         if effective_system:
             api_messages = [{"role": "system", "content": effective_system}] + api_messages
+        api_messages = apply_zai_special_prompt(agent, api_messages)
 
         if moa_config:
             try:

agent/zai_prompt_policy.py

@@ -0,0 +1,103 @@
+"""Z.AI / GLM prompt policy.
+
+Z.AI GLM-5.x has repeatedly returned provider-side 429/code-1305 style
+failures when the normal Hermes-branded system prompt is sent verbatim.  This
+module is an API-boundary sanitizer: it rewrites only the per-request copy of
+messages sent to Z.AI, never the cached prompt or conversation history.
+"""
+from __future__ import annotations
+
+import copy
+import re
+from typing import Any
+
+
+_ZAI_SYSTEM_PREFIX = (
+    "You are a precise local AI coding and operations assistant. "
+    "Answer the user's task directly. Use the provided tools and conversation "
+    "context when available. Do not mention internal platform branding."
+)
+
+_BRANDING_REPLACEMENTS: tuple[tuple[str, str], ...] = (
+    (r"You are Hermes Agent, an intelligent AI assistant created by Nous Research\.\s*", ""),
+    (r"You run on Hermes Agent \(by Nous Research\)\.\s*", ""),
+    (r"Hermes Agent", "the local assistant"),
+    (r"hermes-agent", "local-agent"),
+    (r"\bHermes\b", "the local assistant"),
+    (r"Nous Research", "the platform provider"),
+    (r"\bHERMES_[A-Z0-9_]+\b", "LOCAL_AGENT_ENV"),
+)
+
+
+def is_zai_request(agent: Any) -> bool:
+    """Return True for any request routed to direct Z.AI/GLM endpoints."""
+    provider = (getattr(agent, "provider", "") or "").lower()
+    model = (getattr(agent, "model", "") or "").lower()
+    base_url = (getattr(agent, "base_url", "") or getattr(agent, "_base_url_lower", "") or "").lower()
+    return (
+        provider in {"zai", "z-ai", "z.ai", "glm", "zhipu"}
+        or model.startswith("glm-")
+        or "api.z.ai" in base_url
+        or "open.bigmodel.cn" in base_url
+    )
+
+
+def sanitize_zai_system_prompt(text: str) -> str:
+    """Return the special Z.AI-safe system prompt variant.
+
+    Keep operational/tool instructions, but strip/rewrite Hermes/Nous branding
+    and prepend a stable neutral instruction. This preserves capability while
+    avoiding the provider-triggering brand strings.
+    """
+    if not isinstance(text, str):
+        return text
+    out = text
+    for pattern, repl in _BRANDING_REPLACEMENTS:
+        out = re.sub(pattern, repl, out, flags=re.IGNORECASE)
+    out = re.sub(r"\n{3,}", "\n\n", out).strip()
+    if not out:
+        return _ZAI_SYSTEM_PREFIX
+    if out.startswith(_ZAI_SYSTEM_PREFIX):
+        return out
+    return f"{_ZAI_SYSTEM_PREFIX}\n\n{out}"
+
+
+def _sanitize_content(content: Any) -> Any:
+    if isinstance(content, str):
+        return sanitize_zai_system_prompt(content)
+    if isinstance(content, list):
+        new_content = []
+        for part in content:
+            if isinstance(part, dict):
+                item = part.copy()
+                if isinstance(item.get("text"), str):
+                    item["text"] = sanitize_zai_system_prompt(item["text"])
+                elif isinstance(item.get("content"), str):
+                    item["content"] = sanitize_zai_system_prompt(item["content"])
+                new_content.append(item)
+            else:
+                new_content.append(part)
+        return new_content
+    return content
+
+
+def apply_zai_special_prompt(agent: Any, api_messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Apply Z.AI special prompt policy to an API-message copy.
+
+    The function returns a new list/dicts to avoid mutating cached prompt state
+    or conversation history. It intentionally affects all direct Z.AI/GLM
+    requests, not only council calls.
+    """
+    if not is_zai_request(agent):
+        return api_messages
+    new_messages: list[dict[str, Any]] = []
+    saw_system = False
+    for msg in api_messages:
+        cloned = copy.deepcopy(msg)
+        if cloned.get("role") == "system":
+            saw_system = True
+            cloned["content"] = _sanitize_content(cloned.get("content"))
+        new_messages.append(cloned)
+    if not saw_system:
+        return [{"role": "system", "content": _ZAI_SYSTEM_PREFIX}] + new_messages
+    return new_messages

tests/agent/test_zai_prompt_policy.py

@@ -0,0 +1,42 @@
+from types import SimpleNamespace
+
+from agent.zai_prompt_policy import apply_zai_special_prompt, is_zai_request
+
+
+def test_zai_request_detects_provider_model_and_endpoint():
+    assert is_zai_request(SimpleNamespace(provider="zai", model="glm-5.2", base_url=""))
+    assert is_zai_request(SimpleNamespace(provider="custom", model="glm-5.2", base_url=""))
+    assert is_zai_request(SimpleNamespace(provider="custom", model="x", base_url="https://api.z.ai/api/coding/paas/v4"))
+    assert not is_zai_request(SimpleNamespace(provider="openai-codex", model="gpt-5.5", base_url=""))
+
+
+def test_zai_special_prompt_strips_hermes_branding_without_mutating_input():
+    agent = SimpleNamespace(provider="zai", model="glm-5.2", base_url="https://api.z.ai/api/coding/paas/v4")
+    messages = [
+        {"role": "system", "content": "You are Hermes Agent, an intelligent AI assistant created by Nous Research.\nYou run on Hermes Agent (by Nous Research).\nUse tools."},
+        {"role": "user", "content": "hi"},
+    ]
+    out = apply_zai_special_prompt(agent, messages)
+
+    assert out is not messages
+    assert messages[0]["content"].startswith("You are Hermes Agent")
+    system = out[0]["content"]
+    assert system.startswith("You are a precise local AI coding and operations assistant.")
+    assert "Hermes" not in system
+    assert "Nous Research" not in system
+    assert "Use tools." in system
+    assert out[1] == {"role": "user", "content": "hi"}
+
+
+def test_zai_special_prompt_inserts_system_when_missing():
+    agent = SimpleNamespace(provider="zai", model="glm-5.2", base_url="")
+    out = apply_zai_special_prompt(agent, [{"role": "user", "content": "hi"}])
+    assert out[0]["role"] == "system"
+    assert "local AI" in out[0]["content"]
+    assert out[1]["role"] == "user"
+
+
+def test_non_zai_request_is_unchanged():
+    agent = SimpleNamespace(provider="openai-codex", model="gpt-5.5", base_url="")
+    messages = [{"role": "system", "content": "You are Hermes Agent."}]
+    assert apply_zai_special_prompt(agent, messages) is messages