Files
peltier.io/agent/hotpath_agent/tools.py
T
Claude 19e83d878a agent: harnessd IPC — the six-tool surface over line-delimited JSON
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
2026-07-07 00:03:27 +00:00

72 lines
2.4 KiB
Python

"""Tool surface exposed to the model (SPEC §3.5).
Exactly six tools, all mediated by harnessd over line-delimited JSON —
the agent process has no direct filesystem or shell capability beyond
what these return. Deliberately NOT exposed: shell on the host, writes
outside targets/<name>/, any access to crates/, config/, corpora/.
Phase 2 wiring: each function becomes a Claude Agent SDK tool
definition. The bodies talk to the live harness; construct a client
with `connect(repo_root)` first.
"""
from __future__ import annotations
from typing import Any
from .harness import HarnessClient
_client: HarnessClient | None = None
def connect(repo_root: str, binary: str = "target/debug/harnessd") -> None:
global _client
_client = HarnessClient(repo_root, binary)
def _c() -> HarnessClient:
if _client is None:
raise RuntimeError("call connect(repo_root) first")
return _client
def read_profile(target: str) -> dict[str, Any]:
"""Latest profile for a target: ranked hotspots. Read-only."""
return _c().call("read_profile", target=target)
def read_ledger(target: str) -> dict[str, Any]:
"""Prior attempts against this target — the anti-double-attempt
memory. Never re-grind a class that already has a verdict."""
return _c().call("read_ledger", target=target)
def read_playbook(class_number: int) -> str:
"""Playbook class markdown (1-7)."""
return _c().call("read_playbook", **{"class": class_number})["markdown"]
def read_target_source(target: str, path: str) -> str:
"""Read-only view of the target workspace at the pinned commit."""
return _c().call("read_target_source", target=target, path=path)["content"]
def propose_patch(target: str, diff: str, hypothesis: str) -> str:
"""Submit a unified diff plus its hypothesis. The harness applies it
via `git apply` after a path-allowlist check (SPEC §10); anything
outside targets/<target>/workspace is rejected before any gate
runs. Returns a patch id."""
return _c().call("propose_patch", target=target, diff=diff, hypothesis=hypothesis)["patch_id"]
def run_verdict(patch_id: str, run_id: str, playbook_class: int, hotspot: str) -> dict[str, Any]:
"""Gates + bench + ledger row for a proposed patch. This is the only
way the agent learns whether a change 'worked'."""
return _c().call(
"run_verdict",
patch_id=patch_id,
run_id=run_id,
playbook_class=str(playbook_class),
hotspot=hotspot,
)