c089a5ec2a
HUX-04/HUX-08: after a successful tool execution the hux-runtime plugin now registers freshly written files as typed artifacts (create or immutable version by conversation-scoped title, bounded to 1 MiB, deterministic idempotency keys) and records up to three deduplicated web sources from real network/web_search output. Emitters are fail-open and can never break the tool result. Delivered through the plugin ConfigMap; 95% branch coverage; quality contract updated. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BvMSXH8VH2tMWXanb8SJdf
307 lines
11 KiB
Python
307 lines
11 KiB
Python
"""Hermes execution middleware backed by the loopback HUX policy service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Any, Callable, Mapping
|
|
|
|
from .hux_hook import (
|
|
HuxClient,
|
|
after_tool,
|
|
before_tool,
|
|
canonical_argument_hash,
|
|
emit,
|
|
record_spend,
|
|
)
|
|
|
|
from .context_ids import ContextIds, ContextUnavailable
|
|
from .emitters import sync_after_tool
|
|
from .tool_policy import classify
|
|
|
|
BLOCK_SCHEMA = "hux.tool_block.v1"
|
|
MAX_REPORTED_BYTES = 2**31 - 1
|
|
TRUE = frozenset({"1", "true", "yes", "on"})
|
|
|
|
|
|
def enabled(environ: Mapping[str, str], name: str) -> bool:
|
|
"""Require an explicit truthy environment value."""
|
|
return environ.get(name, "").strip().lower() in TRUE
|
|
|
|
|
|
def _block(reason: str, approval_id: str | None = None) -> str:
|
|
"""Return a model-visible denial without request arguments or service details."""
|
|
body: dict[str, Any] = {
|
|
"schema": BLOCK_SCHEMA,
|
|
"status": "blocked",
|
|
"error": "HUX policy did not release this tool call",
|
|
"reason": str(reason)[:120] or "hux_unavailable",
|
|
}
|
|
if approval_id:
|
|
body["approval_id"] = str(approval_id)[:120]
|
|
return json.dumps(body, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def _result_size(result: Any) -> int:
|
|
"""Measure output for telemetry without persisting or returning its content."""
|
|
if isinstance(result, bytes):
|
|
size = len(result)
|
|
elif isinstance(result, str):
|
|
size = len(result.encode("utf-8", errors="replace"))
|
|
else:
|
|
try:
|
|
size = len(json.dumps(result, ensure_ascii=False, default=str).encode("utf-8"))
|
|
except Exception:
|
|
size = 0
|
|
return min(MAX_REPORTED_BYTES, max(0, size))
|
|
|
|
|
|
def _result_ok(result: Any) -> bool:
|
|
"""Detect the common Hermes error envelopes without inspecting free-form output."""
|
|
value = result
|
|
if isinstance(result, str) and result.startswith("{"):
|
|
try:
|
|
value = json.loads(result)
|
|
except ValueError:
|
|
return True
|
|
if not isinstance(value, dict):
|
|
return True
|
|
status = str(value.get("status", "")).lower()
|
|
return "error" not in value and status not in {"blocked", "cancelled", "error", "failed"}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CallContext:
|
|
"""Stable HUX scope resolved from host-provided middleware metadata."""
|
|
|
|
conversation_id: str
|
|
run_id: str
|
|
tool_call_id: str
|
|
raw_session_id: str
|
|
session_id: str
|
|
project_id: str
|
|
|
|
|
|
class Runtime:
|
|
"""One process-wide HUX execution boundary."""
|
|
|
|
def __init__(
|
|
self,
|
|
client: HuxClient,
|
|
ids: ContextIds,
|
|
enforce: bool,
|
|
project_source: str = "profile:default",
|
|
) -> None:
|
|
self.client = client
|
|
self.ids = ids
|
|
self.enforce = enforce
|
|
if not isinstance(project_source, str) or not project_source or len(project_source) > 200:
|
|
raise ContextUnavailable("HUX project source is malformed")
|
|
self.project_source = project_source
|
|
self._started: set[str] = set()
|
|
self._bootstrapped: set[str] = set()
|
|
self._guard = threading.Lock()
|
|
|
|
@classmethod
|
|
def from_env(cls, environ: Mapping[str, str] | None = None) -> "Runtime":
|
|
"""Build only from projected files and literal loopback configuration."""
|
|
env = os.environ if environ is None else environ
|
|
slot = env.get("HUX_TENANT_SLOT", "")
|
|
subject_file = env.get("HUX_SUBJECT_FILE", "")
|
|
key_file = env.get("HUX_WORKER_KEY_FILE", "")
|
|
context_key_file = env.get("HUX_CONTEXT_KEY_FILE", "")
|
|
if not subject_file or not key_file or not context_key_file:
|
|
raise ContextUnavailable("required HUX runtime files are unavailable")
|
|
client = HuxClient(
|
|
env.get("HUX_BASE_URL", "http://127.0.0.1:8790"),
|
|
{"tenant_slot": slot, "surface": "worker", "trust": "worker"},
|
|
key_file=key_file,
|
|
subject_file=subject_file,
|
|
timeout=float(env.get("HUX_TIMEOUT_SECONDS", "3")),
|
|
)
|
|
ids = ContextIds(context_key_file, slot, client.identity["subject"])
|
|
return cls(
|
|
client,
|
|
ids,
|
|
enabled(env, "HUX_TOOL_ENFORCEMENT"),
|
|
env.get("HUX_PROJECT_SOURCE", "profile:default"),
|
|
)
|
|
|
|
def _scope(self, metadata: Mapping[str, Any]) -> CallContext:
|
|
"""Resolve IDs only from stable host session/turn values."""
|
|
raw_session = metadata.get("session_id")
|
|
raw_turn = metadata.get("turn_id")
|
|
if not isinstance(raw_session, str) or not isinstance(raw_turn, str):
|
|
raise ContextUnavailable("Hermes session context is unavailable")
|
|
return CallContext(
|
|
self.ids.conversation(raw_session),
|
|
self.ids.run(raw_turn),
|
|
str(metadata.get("tool_call_id", ""))[:120],
|
|
raw_session,
|
|
self.ids.session(raw_session),
|
|
self.ids.project(self.project_source),
|
|
)
|
|
|
|
def _bootstrap(self, call: CallContext) -> None:
|
|
"""Idempotently register shared deterministic context before all other HUX use."""
|
|
with self._guard:
|
|
if call.conversation_id in self._bootstrapped:
|
|
return
|
|
self.client.post(
|
|
"/hux/v1/context/bootstrap",
|
|
{
|
|
"raw_session_id": call.raw_session_id,
|
|
"project_source": self.project_source,
|
|
"session_id": call.session_id,
|
|
"conversation_id": call.conversation_id,
|
|
"project_id": call.project_id,
|
|
},
|
|
idempotency_key=f"context:{call.conversation_id}"[:120],
|
|
)
|
|
self._bootstrapped.add(call.conversation_id)
|
|
|
|
def _start_once(self, call: CallContext) -> None:
|
|
"""Emit at most one run-start event per process and stable turn."""
|
|
with self._guard:
|
|
first = call.run_id not in self._started
|
|
if first:
|
|
self._started.add(call.run_id)
|
|
if first:
|
|
emit(
|
|
self.client,
|
|
call.conversation_id,
|
|
"run.started",
|
|
"Hermes started a tool-using turn",
|
|
run_id=call.run_id,
|
|
idempotency_key=f"{call.run_id}:started",
|
|
)
|
|
|
|
def tool_execution(
|
|
self,
|
|
*,
|
|
tool_name: str,
|
|
args: dict[str, Any],
|
|
next_call: Callable[[dict[str, Any]], Any],
|
|
**metadata: Any,
|
|
) -> Any:
|
|
"""Gate the exact effective arguments, execute once, then emit bounded telemetry."""
|
|
try:
|
|
call = self._scope(metadata)
|
|
argument_hash = canonical_argument_hash(tool_name, args)
|
|
policy = classify(tool_name)
|
|
self._bootstrap(call)
|
|
self._start_once(call)
|
|
emit(
|
|
self.client,
|
|
call.conversation_id,
|
|
"tool.call",
|
|
f"{tool_name} requested ({argument_hash[:23]})",
|
|
detail={"tool": tool_name, "capability": policy.capability},
|
|
evidence=[{"kind": "tool_call", "id": call.tool_call_id or argument_hash[7:23], "hash": argument_hash}],
|
|
run_id=call.run_id,
|
|
correlation_id=call.tool_call_id or None,
|
|
idempotency_key=f"{call.run_id}:call:{call.tool_call_id or argument_hash[7:23]}"[:120],
|
|
)
|
|
if self.enforce:
|
|
decision = before_tool(
|
|
self.client,
|
|
call.run_id,
|
|
call.conversation_id,
|
|
tool_name,
|
|
args,
|
|
policy.capability,
|
|
external=policy.external,
|
|
risk=policy.risk,
|
|
)
|
|
if not decision.proceed:
|
|
return _block(decision.reason, decision.approval_id)
|
|
except Exception:
|
|
if self.enforce:
|
|
return _block("hux_unavailable")
|
|
return next_call(args)
|
|
|
|
started = time.monotonic()
|
|
try:
|
|
result = next_call(args)
|
|
except BaseException:
|
|
self._after(call, tool_name, argument_hash, False, 0, started)
|
|
raise
|
|
ok = _result_ok(result)
|
|
self._after(call, tool_name, argument_hash, ok, _result_size(result), started)
|
|
if ok:
|
|
# HUX-04/08: real written files and real web output leave records
|
|
# behind. Emitters are bounded and can never break the tool result.
|
|
sync_after_tool(self.client, call, policy.capability, args, result)
|
|
return result
|
|
|
|
def _after(
|
|
self,
|
|
call: CallContext,
|
|
tool_name: str,
|
|
argument_hash: str,
|
|
ok: bool,
|
|
bytes_out: int,
|
|
started: float,
|
|
) -> None:
|
|
"""Best-effort status-only event and tool budget increment."""
|
|
duration = max(0, int((time.monotonic() - started) * 1000))
|
|
try:
|
|
after_tool(
|
|
self.client,
|
|
call.run_id,
|
|
call.conversation_id,
|
|
tool_name,
|
|
ok,
|
|
bytes_out,
|
|
argument_hash=argument_hash,
|
|
duration_ms=duration,
|
|
)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
record_spend(
|
|
self.client,
|
|
call.run_id,
|
|
call.conversation_id,
|
|
tool_calls=1,
|
|
wall_clock_seconds=max(0, duration // 1000),
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
def session_end(self, **metadata: Any) -> None:
|
|
"""Record run completion; never claim cancellation without process-registry evidence."""
|
|
try:
|
|
call = self._scope(metadata)
|
|
self._bootstrap(call)
|
|
completed = bool(metadata.get("completed")) and not bool(metadata.get("interrupted"))
|
|
kind = "run.completed" if completed else "run.failed"
|
|
summary = "Hermes completed the turn" if completed else "Hermes ended the turn without a verified cancellation"
|
|
emit(
|
|
self.client,
|
|
call.conversation_id,
|
|
kind,
|
|
summary,
|
|
run_id=call.run_id,
|
|
idempotency_key=f"{call.run_id}:ended",
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class UnavailableRuntime:
|
|
"""Fail-closed middleware installed when an opted-in runtime is misconfigured."""
|
|
|
|
def __init__(self, enforce: bool) -> None:
|
|
self.enforce = enforce
|
|
|
|
def tool_execution(self, *, args: dict[str, Any], next_call: Callable, **_metadata: Any) -> Any:
|
|
"""Block enforcement mode; preserve telemetry-only fail-open behavior."""
|
|
return _block("hux_unavailable") if self.enforce else next_call(args)
|
|
|
|
def session_end(self, **_metadata: Any) -> None:
|
|
"""There is no trusted scope in which to report telemetry."""
|