mirrors/hkuds--openharness
1
0
Fork 0
Code Issues Pull Requests Actions 2 Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(gateway): improve codex runtime diagnostics

Browse Source
This commit is contained in:
tjb-tech
2026-04-08 07:22:08 +00:00
parent 69c85e411c
commit 32caf0f381
8 changed files with 302 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ohmo/cli.py
+14
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import asyncio
import logging
import sys
from pathlib import Path
@@ -340,6 +341,18 @@ def _maybe_restart_gateway(*, cwd: str | Path, workspace: str | Path) -> None:
print(f"ohmo gateway restarted (pid={pid})")
def _configure_gateway_logging(workspace: str | Path | None = None) -> None:
"""Configure foreground gateway logging."""
config = load_gateway_config(workspace)
level_name = str(config.log_level or "INFO").upper()
level = getattr(logging, level_name, logging.INFO)
logging.basicConfig(
level=level,
format="%(asctime)s [%(name)s] %(levelname)s %(message)s",
force=True,
)
@app.callback(invoke_without_command=True)
def main(
ctx: typer.Context,
@@ -556,6 +569,7 @@ def gateway_run_cmd(
workspace: str | None = typer.Option(None, "--workspace", help="Path to the ohmo workspace (defaults to ~/.ohmo)"),
) -> None:
"""Run the ohmo gateway in the foreground."""
_configure_gateway_logging(workspace)
service = OhmoGatewayService(cwd, workspace)
raise SystemExit(asyncio.run(service.run_foreground()))
ohmo/gateway/bridge.py
+45 -1
View File
@@ -14,6 +14,14 @@ from ohmo.gateway.runtime import OhmoSessionRuntimePool
logger = logging.getLogger(__name__)
def _content_snippet(text: str, *, limit: int = 160) -> str:
"""Return a single-line preview suitable for logs."""
normalized = " ".join(text.split())
if len(normalized) <= limit:
return normalized
return normalized[: limit - 3] + "..."
def _format_gateway_error(exc: Exception) -> str:
"""Return a short, user-facing gateway error message."""
message = str(exc).strip() or exc.__class__.__name__
@@ -60,6 +68,14 @@ class OhmoGatewayBridge:
break
session_key = session_key_for_message(message)
logger.info(
"ohmo inbound received channel=%s chat_id=%s sender_id=%s session_key=%s content=%r",
message.channel,
message.chat_id,
message.sender_id,
session_key,
_content_snippet(message.content),
)
try:
reply = ""
async for update in self._runtime_pool.stream_message(message, session_key):
@@ -68,6 +84,14 @@ class OhmoGatewayBridge:
continue
if not update.text:
continue
logger.info(
"ohmo outbound update channel=%s chat_id=%s session_key=%s kind=%s content=%r",
message.channel,
message.chat_id,
session_key,
update.kind,
_content_snippet(update.text),
)
await self._bus.publish_outbound(
OutboundMessage(
channel=message.channel,
@@ -77,10 +101,30 @@ class OhmoGatewayBridge:
)
)
except Exception as exc: # pragma: no cover - gateway failure path
logger.exception("ohmo gateway failed to process inbound message")
logger.exception(
"ohmo gateway failed to process inbound message channel=%s chat_id=%s sender_id=%s session_key=%s content=%r",
message.channel,
message.chat_id,
message.sender_id,
session_key,
_content_snippet(message.content),
)
reply = _format_gateway_error(exc)
if not reply:
logger.info(
"ohmo inbound finished without final reply channel=%s chat_id=%s session_key=%s",
message.channel,
message.chat_id,
session_key,
)
continue
logger.info(
"ohmo outbound final channel=%s chat_id=%s session_key=%s content=%r",
message.channel,
message.chat_id,
session_key,
_content_snippet(reply),
)
await self._bus.publish_outbound(
OutboundMessage(
channel=message.channel,
ohmo/gateway/runtime.py
+75
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
from dataclasses import dataclass
import logging
from pathlib import Path
import json
@@ -20,6 +21,8 @@ from openharness.ui.runtime import RuntimeBundle, build_runtime, start_runtime
from ohmo.prompts import build_ohmo_system_prompt
from ohmo.session_storage import OhmoSessionBackend
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class GatewayStreamUpdate:
@@ -58,12 +61,24 @@ class OhmoSessionRuntimePool:
"""Return an existing bundle or create a new one."""
bundle = self._bundles.get(session_key)
if bundle is not None:
logger.info(
"ohmo runtime reusing session session_key=%s session_id=%s prompt=%r",
session_key,
bundle.session_id,
_content_snippet(latest_user_prompt or ""),
)
bundle.engine.set_system_prompt(
build_ohmo_system_prompt(self._cwd, workspace=self._workspace, extra_prompt=None)
)
return bundle
snapshot = self._session_backend.load_latest_for_session_key(session_key)
logger.info(
"ohmo runtime creating session session_key=%s restored=%s prompt=%r",
session_key,
bool(snapshot),
_content_snippet(latest_user_prompt or ""),
)
bundle = await build_runtime(
model=self._model,
max_turns=self._max_turns,
@@ -76,12 +91,26 @@ class OhmoSessionRuntimePool:
if snapshot and snapshot.get("session_id"):
bundle.session_id = str(snapshot["session_id"])
await start_runtime(bundle)
logger.info(
"ohmo runtime started session_key=%s session_id=%s restored_messages=%s",
session_key,
bundle.session_id,
len(snapshot.get("messages") or []) if snapshot else 0,
)
self._bundles[session_key] = bundle
return bundle
async def stream_message(self, message: InboundMessage, session_key: str):
"""Submit an inbound channel message and yield progress + final reply updates."""
bundle = await self.get_bundle(session_key, latest_user_prompt=message.content)
logger.info(
"ohmo runtime processing start channel=%s chat_id=%s session_key=%s session_id=%s content=%r",
message.channel,
message.chat_id,
session_key,
bundle.session_id,
_content_snippet(message.content),
)
bundle.engine.set_system_prompt(
build_ohmo_system_prompt(self._cwd, workspace=self._workspace, extra_prompt=None)
)
@@ -96,6 +125,12 @@ class OhmoSessionRuntimePool:
reply_parts.append(event.text)
continue
if isinstance(event, StatusEvent):
logger.info(
"ohmo runtime status session_key=%s session_id=%s message=%r",
session_key,
bundle.session_id,
_content_snippet(event.message),
)
yield GatewayStreamUpdate(
kind="progress",
text=event.message,
@@ -104,6 +139,13 @@ class OhmoSessionRuntimePool:
continue
if isinstance(event, ToolExecutionStarted):
summary = _summarize_tool_input(event.tool_name, event.tool_input)
logger.info(
"ohmo runtime tool start session_key=%s session_id=%s tool=%s summary=%r",
session_key,
bundle.session_id,
event.tool_name,
summary,
)
hint = f"Using {event.tool_name}"
if summary:
hint = f"{hint}: {summary}"
@@ -118,8 +160,20 @@ class OhmoSessionRuntimePool:
)
continue
if isinstance(event, ToolExecutionCompleted):
logger.info(
"ohmo runtime tool complete session_key=%s session_id=%s tool=%s",
session_key,
bundle.session_id,
event.tool_name,
)
continue
if isinstance(event, ErrorEvent):
logger.error(
"ohmo runtime error session_key=%s session_id=%s message=%r",
session_key,
bundle.session_id,
_content_snippet(event.message),
)
yield GatewayStreamUpdate(
kind="error",
text=event.message,
@@ -138,7 +192,20 @@ class OhmoSessionRuntimePool:
session_id=bundle.session_id,
session_key=session_key,
)
logger.info(
"ohmo runtime saved snapshot session_key=%s session_id=%s message_count=%s reply_chars=%s",
session_key,
bundle.session_id,
len(bundle.engine.messages),
len(reply),
)
if reply:
logger.info(
"ohmo runtime processing complete session_key=%s session_id=%s reply=%r",
session_key,
bundle.session_id,
_content_snippet(reply),
)
yield GatewayStreamUpdate(
kind="final",
text=reply,
@@ -146,6 +213,14 @@ class OhmoSessionRuntimePool:
)
def _content_snippet(text: str, *, limit: int = 160) -> str:
"""Return a compact single-line preview for logs."""
normalized = " ".join(text.split())
if len(normalized) <= limit:
return normalized
return normalized[: limit - 3] + "..."
def _summarize_tool_input(tool_name: str, tool_input: dict[str, object]) -> str:
if not tool_input:
return ""
src/openharness/api/codex_client.py
+34 -6
View File
@@ -167,6 +167,31 @@ def _format_error_message(status_code: int, payload: str) -> str:
return f"Codex request failed with status {status_code}"
def _format_codex_stream_error(event: dict[str, Any], *, fallback: str) -> str:
error = event.get("error")
payload = error if isinstance(error, dict) else event
message = payload.get("message") if isinstance(payload, dict) else None
code = payload.get("code") if isinstance(payload, dict) else None
request_id = (
(payload.get("request_id") if isinstance(payload, dict) else None)
or event.get("request_id")
)
parts: list[str] = []
if isinstance(message, str) and message.strip():
parts.append(message.strip())
elif isinstance(code, str) and code.strip():
parts.append(code.strip())
else:
parts.append(fallback)
if isinstance(code, str) and code.strip():
parts.append(f"(code={code.strip()})")
if isinstance(request_id, str) and request_id.strip():
parts.append(f"[request_id={request_id.strip()}]")
return " ".join(parts)
def _translate_status_error(status_code: int, message: str) -> OpenHarnessApiError:
if status_code in {401, 403}:
return AuthenticationFailure(message)
@@ -282,14 +307,17 @@ class CodexApiClient:
elif event_type == "response.failed":
response_payload = event.get("response")
if isinstance(response_payload, dict):
error = response_payload.get("error")
if isinstance(error, dict):
message = str(error.get("message") or error.get("code") or "Codex response failed")
raise RequestFailure(message)
raise RequestFailure(
_format_codex_stream_error(
response_payload,
fallback="Codex response failed",
)
)
raise RequestFailure("Codex response failed")
elif event_type == "error":
message = str(event.get("message") or event.get("code") or "Codex error")
raise RequestFailure(message)
raise RequestFailure(
_format_codex_stream_error(event, fallback="Codex error")
)
if current_text_parts and not any(isinstance(block, TextBlock) for block in content):
content.insert(0, TextBlock(text="".join(current_text_parts)))
src/openharness/config/settings.py
+4 -1
View File
@@ -655,8 +655,11 @@ class Settings(BaseModel):
if not updates:
return merged
profile_keys = {"model", "base_url", "api_format", "provider", "api_key", "active_profile", "profiles"}
if profile_keys.isdisjoint(updates):
profile_updates = profile_keys.intersection(updates)
if not profile_updates:
return merged
if profile_updates.issubset({"active_profile"}):
return merged.materialize_active_profile()
return merged.sync_active_profile_from_flat_fields().materialize_active_profile()
tests/test_api/test_codex_client.py
+19 -1
View File
@@ -6,7 +6,12 @@ from typing import Any
import pytest
from openharness.api.client import ApiMessageRequest, ApiMessageCompleteEvent, ApiTextDeltaEvent
from openharness.api.codex_client import CodexApiClient, _convert_messages_to_codex, _resolve_codex_url
from openharness.api.codex_client import (
CodexApiClient,
_convert_messages_to_codex,
_format_codex_stream_error,
_resolve_codex_url,
)
from openharness.engine.messages import ConversationMessage, TextBlock, ToolResultBlock, ToolUseBlock
@@ -99,6 +104,19 @@ def test_resolve_codex_url_ignores_unrelated_base_url():
assert _resolve_codex_url("https://api.moonshot.cn/anthropic") == "https://chatgpt.com/backend-api/codex/responses"
def test_format_codex_stream_error_includes_code_and_request_id():
message = _format_codex_stream_error(
{
"type": "error",
"message": "Upstream overloaded",
"code": "overloaded",
"request_id": "req_123",
},
fallback="Codex error",
)
assert message == "Upstream overloaded (code=overloaded) [request_id=req_123]"
@pytest.mark.asyncio
async def test_codex_client_streams_text(monkeypatch):
sink: dict[str, Any] = {}
tests/test_config/test_settings.py
+38
View File
@@ -198,6 +198,44 @@ class TestLoadSaveSettings:
assert materialized.api_format == "openai"
assert materialized.model == "gpt-5"
def test_merge_cli_active_profile_does_not_inherit_flat_provider_fields(self):
settings = Settings(
active_profile="moonshot",
provider="moonshot",
api_format="openai",
base_url="https://api.moonshot.cn/v1",
model="kimi-k2.5",
profiles={
"moonshot": ProviderProfile(
label="Moonshot",
provider="moonshot",
api_format="openai",
auth_source="moonshot_api_key",
default_model="kimi-k2.5",
last_model="kimi-k2.5",
base_url="https://api.moonshot.cn/v1",
),
"codex": ProviderProfile(
label="Codex Subscription",
provider="openai_codex",
api_format="openai",
auth_source="codex_subscription",
default_model="gpt-5.4",
last_model="gpt-5.4",
),
},
)
updated = settings.merge_cli_overrides(active_profile="codex")
profile_name, profile = updated.resolve_profile()
assert profile_name == "codex"
assert updated.provider == "openai_codex"
assert updated.base_url is None
assert updated.model == "gpt-5.4"
assert profile.provider == "openai_codex"
assert profile.auth_source == "codex_subscription"
def test_claude_profile_materializes_alias_to_concrete_model(self):
settings = Settings(
active_profile="claude-subscription",
tests/test_ohmo/test_gateway.py
+73
View File
@@ -1,4 +1,5 @@
import asyncio
import logging
from types import SimpleNamespace
from datetime import datetime
import json
@@ -211,3 +212,75 @@ async def test_gateway_bridge_publishes_progress_updates():
assert second.metadata["_tool_hint"] is True
assert "Using web_fetch" in second.content
assert third.content == "Done"
@pytest.mark.asyncio
async def test_gateway_bridge_logs_inbound_and_final(caplog):
bus = MessageBus()
class FakeRuntimePool:
async def stream_message(self, message, session_key):
yield SimpleNamespace(kind="progress", text="Thinking...", metadata={"_progress": True, "_session_key": session_key})
yield SimpleNamespace(kind="final", text="Done", metadata={"_session_key": session_key})
bridge = OhmoGatewayBridge(bus=bus, runtime_pool=FakeRuntimePool())
task = asyncio.create_task(bridge.run())
caplog.set_level(logging.INFO)
try:
await bus.publish_inbound(
InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content="please translate this")
)
await asyncio.wait_for(bus.consume_outbound(), timeout=1.0)
await asyncio.wait_for(bus.consume_outbound(), timeout=1.0)
finally:
bridge.stop()
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
assert "ohmo inbound received" in caplog.text
assert "ohmo outbound final" in caplog.text
assert "please translate this" in caplog.text
@pytest.mark.asyncio
async def test_runtime_pool_logs_session_lifecycle(tmp_path, monkeypatch, caplog):
workspace = tmp_path / ".ohmo-home"
initialize_workspace(workspace)
async def fake_build_runtime(**kwargs):
class FakeEngine:
messages = []
total_usage = UsageSnapshot()
def set_system_prompt(self, prompt):
return None
async def submit_message(self, content):
yield ToolExecutionStarted(tool_name="web_fetch", tool_input={"url": "https://example.com"})
yield AssistantTextDelta(text="done")
return SimpleNamespace(
engine=FakeEngine(),
session_id="sess123",
current_settings=lambda: SimpleNamespace(model="gpt-5.4"),
)
async def fake_start_runtime(bundle):
return None
monkeypatch.setattr("ohmo.gateway.runtime.build_runtime", fake_build_runtime)
monkeypatch.setattr("ohmo.gateway.runtime.start_runtime", fake_start_runtime)
pool = OhmoSessionRuntimePool(cwd=tmp_path, workspace=workspace, provider_profile="codex")
message = InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content="check")
caplog.set_level(logging.INFO)
updates = [u async for u in pool.stream_message(message, "feishu:c1")]
assert updates[-1].text == "done"
assert "ohmo runtime processing start" in caplog.text
assert "ohmo runtime tool start" in caplog.text
assert "ohmo runtime saved snapshot" in caplog.text
assert "ohmo runtime processing complete" in caplog.text