RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(codex): defang reserved Harmony tokens in requests

This commit is contained in:
Xipong 2026-07-20 17:37:22 +03:00 committed by Teknium
parent 0d9892379c
commit 9ceb0858ab
6 changed files with 390 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

112
agent/codex_responses_adapter.py
View File

@ -14,6 +14,7 @@ import hashlib
import json
import logging
import re
import unicodedata
import uuid
from types import SimpleNamespace
from typing import Any, Dict, List, Optional
@ -73,6 +74,79 @@ _TOOL_CALL_LEAK_PATTERN = re.compile(
)
# The ChatGPT Codex backend reserves these Harmony wire tokens. If their
# literal spellings are replayed anywhere in request text, the backend rejects
# the request before inference with ``invalid_prompt: Request blocked.``.
# Category-Cf handling covers persisted sessions from an earlier U+200B weak
# defang; fullwidth bars survive format-character stripping while keeping the
# inspected source legible.
_HARMONY_CONTROL_TOKEN_RE = re.compile(
r"<\|(start|end|channel|message|constrain|return|call)\|>"
)
_FULLWIDTH_PIPE = "\uff5c"
def _neutralize_harmony_tokens(text: str) -> str:
"""Keep Harmony source readable without emitting reserved wire tokens."""
if not text or "<" not in text or "|" not in text:
return text
replacement = rf"<{_FULLWIDTH_PIPE}\1{_FULLWIDTH_PIPE}>"
if not any(unicodedata.category(char) == "Cf" for char in text):
return _HARMONY_CONTROL_TOKEN_RE.sub(replacement, text)
# U+200B is confirmed to be stripped by the Codex backend before its
# reserved-token check. Treat every Unicode format control equivalently so
# moving the character elsewhere in the token (or swapping in another Cf)
# cannot recreate the same visually hidden form.
visible_chars: List[str] = []
original_positions: List[int] = []
for index, char in enumerate(text):
if unicodedata.category(char) == "Cf":
continue
visible_chars.append(char)
original_positions.append(index)
visible_text = "".join(visible_chars)
matches = list(_HARMONY_CONTROL_TOKEN_RE.finditer(visible_text))
if not matches:
return text
result: List[str] = []
original_cursor = 0
for match in matches:
original_start = original_positions[match.start()]
original_end = original_positions[match.end() - 1] + 1
result.append(text[original_cursor:original_start])
result.append(f"<{_FULLWIDTH_PIPE}{match.group(1)}{_FULLWIDTH_PIPE}>")
original_cursor = original_end
result.append(text[original_cursor:])
return "".join(result)
def _neutralize_harmony_structure(value: Any) -> Any:
"""Neutralize JSON-like values; normalize tuples and reject unsafe keys.
Rewriting an object key could desynchronize a tool schema from the executor
contract, so a reserved token there is rejected explicitly instead.
"""
if isinstance(value, str):
return _neutralize_harmony_tokens(value)
if isinstance(value, (list, tuple)):
return [_neutralize_harmony_structure(item) for item in value]
if isinstance(value, dict):
normalized = {}
for key, item in value.items():
if isinstance(key, str) and _neutralize_harmony_tokens(key) != key:
raise ValueError(
"Reserved Harmony tokens in a JSON object key cannot be "
"neutralized without changing its contract."
)
normalized[key] = _neutralize_harmony_structure(item)
return normalized
return value
# ---------------------------------------------------------------------------
# Multimodal content helpers
# ---------------------------------------------------------------------------
@ -627,10 +701,16 @@ def _preflight_codex_input_items(
raw_items: Any,
*,
is_github_responses: bool = False,
sanitize_harmony_tokens: bool = False,
) -> List[Dict[str, Any]]:
if not isinstance(raw_items, list):
raise ValueError("Codex Responses input must be a list of input items.")
sanitize_text = (
_neutralize_harmony_tokens
if sanitize_harmony_tokens
else lambda text: text
)
normalized: List[Dict[str, Any]] = []
seen_ids: set = set()
for idx, item in enumerate(raw_items):
@ -651,7 +731,7 @@ def _preflight_codex_input_items(
arguments = json.dumps(arguments, ensure_ascii=False)
elif not isinstance(arguments, str):
arguments = str(arguments)
arguments = arguments.strip() or "{}"
arguments = sanitize_text(arguments.strip() or "{}")
normalized.append(
{
@ -685,7 +765,7 @@ def _preflight_codex_input_items(
if ptype == "input_text":
text = part.get("text")
if isinstance(text, str) and text:
cleaned.append({"type": "input_text", "text": text})
cleaned.append({"type": "input_text", "text": sanitize_text(text)})
elif ptype == "input_image":
url = part.get("image_url")
if isinstance(url, str) and url:
@ -709,7 +789,7 @@ def _preflight_codex_input_items(
{
"type": "function_call_output",
"call_id": call_id.strip(),
"output": output,
"output": sanitize_text(output),
}
)
continue
@ -722,14 +802,21 @@ def _preflight_codex_input_items(
if item_id in seen_ids:
continue
seen_ids.add(item_id)
reasoning_item = {"type": "reasoning", "encrypted_content": encrypted}
reasoning_item: Dict[str, Any] = {
"type": "reasoning",
"encrypted_content": encrypted,
}
# Do NOT include the "id" in the outgoing item — with
# store=False (our default) the API tries to resolve the
# id server-side and returns 404. The id is still used
# above for local deduplication via seen_ids.
summary = item.get("summary")
if isinstance(summary, list):
reasoning_item["summary"] = summary
reasoning_item["summary"] = (
_neutralize_harmony_structure(summary)
if sanitize_harmony_tokens
else summary
)
else:
reasoning_item["summary"] = []
normalized.append(reasoning_item)
@ -758,7 +845,7 @@ def _preflight_codex_input_items(
text = ""
if not isinstance(text, str):
text = str(text)
normalized_content.append({"type": "output_text", "text": text})
normalized_content.append({"type": "output_text", "text": sanitize_text(text)})
if not normalized_content:
raise ValueError(f"Codex Responses input[{idx}] message item must contain at least one text part.")
normalized_item: Dict[str, Any] = {
@ -798,7 +885,7 @@ def _preflight_codex_input_items(
for part_idx, part in enumerate(content):
if isinstance(part, str):
if part:
validated.append({"type": text_type, "text": part})
validated.append({"type": text_type, "text": sanitize_text(part)})
continue
if not isinstance(part, dict):
raise ValueError(
@ -809,7 +896,7 @@ def _preflight_codex_input_items(
text = part.get("text", "")
if not isinstance(text, str):
text = str(text or "")
validated.append({"type": text_type, "text": text})
validated.append({"type": text_type, "text": sanitize_text(text)})
elif ptype in {"input_image", "image_url"}:
image_ref = part.get("image_url", "")
detail = part.get("detail")
@ -833,7 +920,7 @@ def _preflight_codex_input_items(
if not isinstance(content, str):
content = str(content)
normalized.append({"role": role, "content": content})
normalized.append({"role": role, "content": sanitize_text(content)})
continue
raise ValueError(
@ -848,6 +935,7 @@ def _preflight_codex_api_kwargs(
*,
allow_stream: bool = False,
is_github_responses: bool = False,
sanitize_harmony_tokens: bool = False,
) -> Dict[str, Any]:
if not isinstance(api_kwargs, dict):
raise ValueError("Codex Responses request must be a dict.")
@ -868,10 +956,13 @@ def _preflight_codex_api_kwargs(
if not isinstance(instructions, str):
instructions = str(instructions)
instructions = instructions.strip() or DEFAULT_AGENT_IDENTITY
if sanitize_harmony_tokens:
instructions = _neutralize_harmony_tokens(instructions)
normalized_input = _preflight_codex_input_items(
api_kwargs.get("input"),
is_github_responses=is_github_responses,
sanitize_harmony_tokens=sanitize_harmony_tokens,
)
tools = api_kwargs.get("tools")
@ -928,6 +1019,9 @@ def _preflight_codex_api_kwargs(
}
)
if sanitize_harmony_tokens and normalized_tools is not None:
normalized_tools = _neutralize_harmony_structure(normalized_tools)
store = api_kwargs.get("store", False)
if store is not False:
raise ValueError("Codex Responses contract requires 'store' to be false.")

2
agent/conversation_loop.py
View File

@ -2131,6 +2131,7 @@ def run_conversation(
api_kwargs,
allow_stream=False,
is_github_responses=agent._is_copilot_url(),
sanitize_harmony_tokens=agent._is_codex_backend(),
)
# Copilot x-initiator: the first API call of a user turn is
# marked "user" so Copilot bills a premium request; tool-loop
@ -2290,6 +2291,7 @@ def run_conversation(
next_api_kwargs,
allow_stream=False,
is_github_responses=agent._is_copilot_url(),
sanitize_harmony_tokens=agent._is_codex_backend(),
)
if _use_streaming:
return agent._interruptible_streaming_api_call(

4
agent/transports/codex.py
View File

@ -544,10 +544,13 @@ class ResponsesApiTransport(ProviderTransport):
*,
allow_stream: bool = False,
is_github_responses: bool = False,
sanitize_harmony_tokens: bool = False,
) -> dict:
"""Validate and sanitize Codex API kwargs before the call.
Normalizes input items, strips unsupported fields, validates structure.
``sanitize_harmony_tokens`` is enabled only for the ChatGPT Codex
backend, which rejects literal reserved Harmony wire tokens in text.
"""
from agent.codex_responses_adapter import _preflight_codex_api_kwargs
@ -555,6 +558,7 @@ class ResponsesApiTransport(ProviderTransport):
api_kwargs,
allow_stream=allow_stream,
is_github_responses=is_github_responses,
sanitize_harmony_tokens=sanitize_harmony_tokens,
)
if "prompt_cache_key" in normalized:
bounded = _bounded_prompt_cache_key(normalized["prompt_cache_key"])

9
run_agent.py
View File

@ -1500,6 +1500,15 @@ class AIAgent:
return True
return self._is_copilot_url()
def _is_codex_backend(self) -> bool:
"""Return True for the ChatGPT OAuth Codex Responses backend."""
return (
getattr(self, "api_mode", None) == "codex_responses"
and getattr(self, "_base_url_hostname", "") == "chatgpt.com"
and "/backend-api/codex"
in (getattr(self, "_base_url_lower", "") or "")
)
def _anthropic_prompt_cache_policy(
self,
*,

195
tests/agent/test_codex_responses_adapter.py
View File

@ -6,11 +6,196 @@ from agent.codex_responses_adapter import (
_chat_messages_to_responses_input,
_format_responses_error,
_normalize_codex_response,
_neutralize_harmony_tokens,
_preflight_codex_api_kwargs,
_preflight_codex_input_items,
)
_HARMONY_SOURCE_SNIPPET = (
"<|end|><|start|>assistant<|channel|>analysis<|message|>"
"Need to generate one image according to the description."
"<|end|><|start|>assistant<|channel|>final<|message|>"
)
def _harmony_token(name: str) -> str:
"""Build a literal Harmony token without spelling it contiguously here."""
return f"<\x7c{name}\x7c>"
def test_codex_preflight_gate_off_preserves_harmony_tokens_byte_for_byte():
raw = [{
"type": "function_call_output",
"call_id": "call_1",
"output": _HARMONY_SOURCE_SNIPPET,
}]
normalized = _preflight_codex_input_items(raw)
assert normalized[0]["output"] == _HARMONY_SOURCE_SNIPPET
def test_harmony_neutralizer_defangs_only_reserved_control_tokens():
for name in ("start", "end", "channel", "message", "constrain", "return", "call"):
literal = _harmony_token(name)
assert _neutralize_harmony_tokens(literal) == f"<｜{name}｜>"
qwen = f"<|im_{name}|>"
assert _neutralize_harmony_tokens(qwen) == qwen
def test_harmony_neutralizer_upgrades_zwsp_and_is_idempotent():
weak = "<\u200b|start|>assistant<\u200b|channel|>analysis"
once = _neutralize_harmony_tokens(weak)
assert "\u200b" not in once
assert once == "<｜start｜>assistant<｜channel｜>analysis"
assert _neutralize_harmony_tokens(once) == once
def test_harmony_neutralizer_handles_repeated_zwsp_before_pipe():
weak = "<\u200b\u200b|start|>assistant<\u200b\u200b\u200b|message|>"
assert _neutralize_harmony_tokens(weak) == "<｜start｜>assistant<｜message｜>"
def test_harmony_neutralizer_handles_format_controls_anywhere_in_token():
disguised = (
"<\u200c|start|>",
"<|\u200bstart|>",
"<|st\u200dart|>",
"<|start\u2060|>",
"<|start|\ufeff>",
)
for token in disguised:
assert _neutralize_harmony_tokens(token) == "<｜start｜>"
def test_codex_api_preflight_sanitizes_tuple_values_in_tool_schemas():
kwargs = {
"model": "gpt-5-codex",
"instructions": "test",
"input": [{"role": "user", "content": "hello"}],
"tools": [{
"type": "function",
"name": "choose_mode",
"parameters": {
"type": "object",
"properties": {
"mode": {
"type": "string",
"enum": (_harmony_token("call"), "plain"),
},
},
},
}],
"store": False,
}
normalized = _preflight_codex_api_kwargs(kwargs, sanitize_harmony_tokens=True)
assert normalized["tools"][0]["parameters"]["properties"]["mode"]["enum"] == [
"<｜call｜>",
"plain",
]
def test_codex_api_preflight_rejects_reserved_token_in_structural_key():
kwargs = {
"model": "gpt-5-codex",
"instructions": "test",
"input": [{"role": "user", "content": "hello"}],
"tools": [{
"type": "function",
"name": "unsafe_schema",
"parameters": {
"type": "object",
"properties": {
_harmony_token("start"): {"type": "string"},
},
},
}],
"store": False,
}
with pytest.raises(ValueError, match="JSON object key"):
_preflight_codex_api_kwargs(kwargs, sanitize_harmony_tokens=True)
def test_codex_api_preflight_defangs_every_outbound_text_carrier():
raw = [
{
"type": "function_call",
"call_id": "call_args",
"name": "terminal",
"arguments": '{"command":"echo ' + _harmony_token("channel") + '"}',
},
{
"type": "function_call_output",
"call_id": "call_output_parts",
"output": [{"type": "input_text", "text": _HARMONY_SOURCE_SNIPPET}],
},
{
"type": "reasoning",
"encrypted_content": "opaque-reasoning-carrier",
"summary": [{
"type": "summary_text",
"text": "Summary containing " + _harmony_token("constrain"),
}],
},
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": _HARMONY_SOURCE_SNIPPET}],
},
{
"role": "user",
"content": [
_HARMONY_SOURCE_SNIPPET,
{"type": "input_text", "text": _HARMONY_SOURCE_SNIPPET},
],
},
{
"role": "user",
"content": _HARMONY_SOURCE_SNIPPET + " qwen=<|im_start|>",
},
]
kwargs = {
"model": "gpt-5-codex",
"instructions": "Inspect this wire token: " + _harmony_token("start"),
"input": raw,
"tools": [{
"type": "function",
"name": "inspect_wire_format",
"description": "Inspect " + _harmony_token("message"),
"parameters": {
"type": "object",
"properties": {
"source": {
"type": "string",
"description": "Source containing " + _harmony_token("return"),
},
},
},
}],
"store": False,
}
normalized = _preflight_codex_api_kwargs(
kwargs,
sanitize_harmony_tokens=True,
)
serialized = str(normalized)
for name in ("start", "end", "channel", "message", "constrain", "return"):
assert _harmony_token(name) not in serialized
assert serialized.count("Need to generate one image according to the description.") == 5
assert normalized["instructions"] == "Inspect this wire token: <｜start｜>"
assert "<｜message｜>" in str(normalized["tools"])
assert "<|im_start|>" in serialized
def test_normalize_codex_response_treats_summary_only_reasoning_as_incomplete():
@ -301,12 +486,4 @@ def _xai_reasoning_only_response(reasoning_text):
summary=[SimpleNamespace(text=reasoning_text)],
)
],
)
)

86
tests/run_agent/test_run_agent_codex_responses.py
View File

@ -260,6 +260,7 @@ def test_api_mode_uses_explicit_provider_when_codex(monkeypatch):
)
assert agent.api_mode == "codex_responses"
assert agent.provider == "openai-codex"
assert agent._is_codex_backend() is False
@ -687,6 +688,91 @@ def test_run_conversation_codex_plain_text(monkeypatch):
assert result["messages"][-1]["content"] == "OK"
def test_codex_preflight_defangs_harmony_tokens_before_and_after_middleware(monkeypatch):
"""Both mutable request boundaries must reject literal Harmony wire tokens."""
agent = _build_agent(monkeypatch)
setattr(agent, "_disable_streaming", True)
token = f"<\x7cstart\x7c>"
captured = {}
def _request_middleware(request, **_context):
# Initial preflight runs before request middleware.
assert token not in str(request["input"])
replacement = dict(request)
replacement["instructions"] = "Inspect source containing " + token
replacement["input"] = [{
"type": "function_call_output",
"call_id": "call_poisoned",
"output": "source contains " + token,
}]
return SimpleNamespace(
payload=replacement,
original_payload=request,
changed=True,
trace=[],
)
def _execution_middleware(request, next_call, **_context):
# Request middleware can reintroduce a reserved token after initial
# preflight, so it must still be present before the dispatch chokepoint.
assert token in str(request)
return next_call(request)
def _capture_api_call(api_kwargs):
captured.update(api_kwargs)
return _codex_message_response("OK")
monkeypatch.setattr(
"hermes_cli.middleware.apply_llm_request_middleware",
_request_middleware,
)
monkeypatch.setattr(
"hermes_cli.middleware.run_llm_execution_middleware",
_execution_middleware,
)
monkeypatch.setattr(agent, "_interruptible_api_call", _capture_api_call)
result = agent.run_conversation("Read " + token)
assert result["completed"] is True
assert token not in captured["instructions"]
assert token not in str(captured["input"])
assert "<｜start｜>" in captured["instructions"]
def test_copilot_responses_preflight_preserves_harmony_tokens(monkeypatch):
"""Other Responses-compatible providers remain byte-identical."""
agent = _build_copilot_agent(monkeypatch)
setattr(agent, "_disable_streaming", True)
token = f"<\x7cstart\x7c>"
captured = {}
def _capture_api_call(api_kwargs):
captured.update(api_kwargs)
return _codex_message_response("OK")
monkeypatch.setattr(agent, "_interruptible_api_call", _capture_api_call)
result = agent.run_conversation("Read " + token)
assert result["completed"] is True
assert token in str(captured["input"])
def test_codex_backend_detection_is_narrow(monkeypatch):
codex = _build_agent(monkeypatch)
copilot = _build_copilot_agent(monkeypatch)
assert codex._is_codex_backend() is True
assert copilot._is_codex_backend() is False
# Exact backend URL detection still works for an explicitly custom route.
setattr(codex, "provider", "custom")
assert codex._is_codex_backend() is True
setattr(codex, "api_mode", "chat_completions")
assert codex._is_codex_backend() is False
def test_copilot_final_preflight_sanitizes_both_middleware_layers(monkeypatch):
"""The dispatch chokepoint must sanitize after every mutable layer."""
agent = _build_copilot_agent(monkeypatch)