mirror of
https://github.com/bdbrown4/peltier.io.git
synced 2026-11-07 10:58:28 +00:00
Claude
19e83d878a
The agent's only door to the trust layer (SPEC 3.5, 10): harnessd serves read_profile, read_ledger, read_playbook, read_target_source, propose_patch, run_verdict over stdio JSON lines. propose_patch enforces the path allowlist (relative paths only, no parent components, no .git, target name validated) before git apply --check + apply rooted in the target workspace; run_verdict builds the candidate into an isolated dir and hands off to the verdict pipeline (pristine-rebuilt baseline included). Python client (agent/hotpath_agent/harness.py) + tools.py rewired from stubs to live calls. E2E evidence: escape attempts refused (bad target name, ../ paths, trust-layer diff touching config/accept.toml rejected before any apply); benign doc patch applied, candidate built, full verdict ran mechanically end to end -> +8.9% [1.079, 1.104] accepted (ship-tree patches ride along as the candidate), ledger row phase1-ipc-smoke-001. Phase 1 caveat recorded in the daemon header: same-uid filesystem read-only enforcement still needs the separate-user/container setup. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EcKPWQ3MQD4NdEHwar96u2
45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
"""Client for harnessd — the agent's only door to the trust layer.
|
|
|
|
Spawns the Rust daemon and speaks one JSON object per line over its
|
|
stdio. The client is deliberately dumb: no retries that could double-
|
|
apply a patch, no local fallbacks that could fabricate evidence.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from typing import Any
|
|
|
|
|
|
class HarnessError(RuntimeError):
|
|
"""The harness refused or failed an operation."""
|
|
|
|
|
|
class HarnessClient:
|
|
def __init__(self, repo_root: str, binary: str = "target/debug/harnessd"):
|
|
self._proc = subprocess.Popen(
|
|
[binary],
|
|
cwd=repo_root,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
|
|
def call(self, op: str, **kwargs: Any) -> dict[str, Any]:
|
|
assert self._proc.stdin and self._proc.stdout
|
|
self._proc.stdin.write(json.dumps({"op": op, **kwargs}) + "\n")
|
|
self._proc.stdin.flush()
|
|
line = self._proc.stdout.readline()
|
|
if not line:
|
|
raise HarnessError("harnessd exited")
|
|
resp = json.loads(line)
|
|
if not resp.get("ok"):
|
|
raise HarnessError(resp.get("error", "unknown harness error"))
|
|
return resp["result"]
|
|
|
|
def close(self) -> None:
|
|
if self._proc.stdin:
|
|
self._proc.stdin.close()
|
|
self._proc.wait(timeout=10)
|