diff --git a/src/main/hermes/hook-service.test.ts b/src/main/hermes/hook-service.test.ts index 33218042b46..df1d95a077f 100644 --- a/src/main/hermes/hook-service.test.ts +++ b/src/main/hermes/hook-service.test.ts @@ -224,4 +224,46 @@ describe('HermesHookService', () => { } }) }) + + it('bounds generated plugin payload normalization before JSON encoding', () => { + const pythonAvailable = spawnSync('python3', ['--version'], { encoding: 'utf-8' }).status === 0 + if (!pythonAvailable) { + return + } + new HermesHookService().install() + + const initPath = join(homeDir, 'plugins', _internals.HERMES_PLUGIN_NAME, '__init__.py') + const script = [ + 'import importlib.util, json', + `spec = importlib.util.spec_from_file_location("orca_status", ${JSON.stringify(initPath)})`, + 'mod = importlib.util.module_from_spec(spec)', + 'spec.loader.exec_module(mod)', + 'payload = mod._payload_for_event("post_tool_call", {', + ' "session_id": "sess-1",', + ' "tool_name": "BigTool",', + ' "args": {"long": "x" * 9000, "items": list(range(200))},', + ' "result": {"text": "y" * 9000, "nested": [{"value": i} for i in range(200)]},', + ' "duration_ms": 12,', + '})', + 'print(json.dumps(payload))' + ].join('\n') + + const output = execFileSync('python3', ['-c', script], { + encoding: 'utf-8', + timeout: 15_000 + }) + const payload = JSON.parse(output) as { + args: { long: string; items: unknown[] } + result: { text: string; nested: unknown[] } + tool_input: { items: unknown[] } + } + + expect(payload.args.long).toHaveLength(8192 + '...[truncated]'.length) + expect(payload.args.long.endsWith('...[truncated]')).toBe(true) + expect(payload.args.items).toHaveLength(51) + expect(payload.args.items.at(-1)).toBe('...[truncated]') + expect(payload.result.text.endsWith('...[truncated]')).toBe(true) + expect(payload.result.nested).toHaveLength(51) + expect(payload.tool_input.items).toHaveLength(51) + }) }) diff --git a/src/main/hermes/hook-service.ts b/src/main/hermes/hook-service.ts index 0f3e9bbd890..2c9ea3f498c 100644 --- a/src/main/hermes/hook-service.ts +++ b/src/main/hermes/hook-service.ts @@ -289,9 +289,16 @@ import json import os import urllib.error import urllib.request -from typing import Any, Callable +from typing import Any, Callable, Optional EVENTS = ${JSON.stringify(HERMES_EVENTS)} +# Why: hook args/results can include full tool payloads. Bound traversal before +# JSON encoding so status hooks stay best-effort and cheap. +MAX_JSONABLE_DEPTH = 5 +MAX_JSONABLE_ITEMS = 50 +MAX_JSONABLE_NODES = 500 +MAX_JSONABLE_STRING = 8192 +TRUNCATED = "...[truncated]" SELECTED_KEYS = { "on_session_start": ("session_id", "model", "platform"), "pre_llm_call": ("session_id", "user_message", "is_first_turn", "model", "platform", "sender_id"), @@ -306,16 +313,41 @@ SELECTED_KEYS = { } -def _jsonable(value: Any, depth: int = 0) -> Any: - if depth > 5: - return repr(value) - if value is None or isinstance(value, (str, int, float, bool)): +def _truncate_string(value: str) -> str: + if len(value) <= MAX_JSONABLE_STRING: return value + return value[:MAX_JSONABLE_STRING] + TRUNCATED + + +def _jsonable(value: Any, depth: int = 0, budget: Optional[list[int]] = None) -> Any: + if budget is None: + budget = [MAX_JSONABLE_NODES] + if budget[0] <= 0: + return TRUNCATED + budget[0] -= 1 + if depth > MAX_JSONABLE_DEPTH: + return _truncate_string(repr(value)) + if value is None or isinstance(value, (int, float, bool)): + return value + if isinstance(value, str): + return _truncate_string(value) if isinstance(value, dict): - return {str(k): _jsonable(v, depth + 1) for k, v in value.items()} + out: dict[str, Any] = {} + for index, (k, v) in enumerate(value.items()): + if index >= MAX_JSONABLE_ITEMS: + out[TRUNCATED] = True + break + out[_truncate_string(str(k))] = _jsonable(v, depth + 1, budget) + return out if isinstance(value, (list, tuple, set)): - return [_jsonable(v, depth + 1) for v in value] - return repr(value) + out = [] + for index, item in enumerate(value): + if index >= MAX_JSONABLE_ITEMS: + out.append(TRUNCATED) + break + out.append(_jsonable(item, depth + 1, budget)) + return out + return _truncate_string(repr(value)) def _endpoint_env() -> dict[str, str]: