附录 D：随书 Harness 完整代码

"""Teaching harness: deterministic model, durable tool protocol, no API keys.

Path checks here are application checks, NOT an operating-system sandbox.
The journal assumes a local filesystem, one writer and cooperative tools.
"""
from __future__ import annotations
from contextlib import contextmanager
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Callable, Iterable, Protocol
import hashlib
import json
import os
import threading


def digest(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def canonical(value: object) -> bytes:
    return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode()


class HarnessError(Exception):
    pass


class ProtocolError(HarnessError):
    pass


class ReviewRequired(HarnessError):
    pass


class BudgetExceeded(HarnessError):
    pass


class VerificationFailed(HarnessError):
    pass


class TransientModelError(HarnessError):
    """Only a model adapter may mark a pre-reply request safe to retry."""


@dataclass(frozen=True)
class Call:
    id: str
    name: str
    arguments: dict


@dataclass(frozen=True)
class Reply:
    text: str
    calls: tuple[Call, ...] = ()


def decode_stream(events: Iterable[dict]) -> Reply:
    """Buffer fragments until a complete end marker; never execute partial calls."""
    text = []
    fragments: dict[str, dict] = {}
    ended = False
    for event in events:
        if ended:
            raise ProtocolError("event after reply.end")
        if event.get("type") == "text.delta":
            if not isinstance(event.get("text"), str):
                raise ProtocolError("text delta must be a string")
            text.append(event["text"])
        elif event.get("type") == "tool.delta":
            cid, name, args = event.get("id"), event.get("name"), event.get("arguments")
            if not isinstance(cid, str) or not cid or not isinstance(args, str):
                raise ProtocolError("invalid tool fragment")
            if cid not in fragments:
                if not isinstance(name, str) or not name:
                    raise ProtocolError("first fragment needs a tool name")
                fragments[cid] = {"name": name, "arguments": ""}
            elif name is not None and name != fragments[cid]["name"]:
                raise ProtocolError("tool name changed inside stream")
            fragments[cid]["arguments"] += args
        elif event.get("type") == "reply.end":
            if event.get("reason") != "stop":
                raise ProtocolError("truncated, cancelled or failed response")
            ended = True
        else:
            raise ProtocolError("unknown model stream event")
    if not ended:
        raise ProtocolError("stream ended without reply.end")
    calls = []
    for cid, data in fragments.items():
        try:
            args = json.loads(data["arguments"])
        except (ValueError, TypeError) as exc:
            raise ProtocolError("invalid tool JSON") from exc
        if not isinstance(args, dict):
            raise ProtocolError("tool arguments must be an object")
        calls.append(Call(cid, data["name"], args))
    return Reply("".join(text), tuple(calls))


class Model(Protocol):
    def stream(self, messages: list[dict], step: int) -> Iterable[dict]: ...


class Journal:
    def __init__(self, path: Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.lock_path = self.path.with_suffix(self.path.suffix + ".lock")
        self.events = self._load()

    def _load(self) -> list[dict]:
        events = []
        previous = "0" * 64
        if not self.path.exists():
            return events
        raw = self.path.read_bytes()
        if raw and not raw.endswith(b"\n"):
            raise ProtocolError("torn journal tail; preserve and repair explicitly")
        for line in raw.splitlines():
            try:
                event = json.loads(line)
                signature = event.pop("hash")
                if event["seq"] != len(events) or event["previous"] != previous:
                    raise ValueError("sequence or hash chain mismatch")
                if digest(canonical(event)) != signature:
                    raise ValueError("record hash mismatch")
                event["hash"] = signature
                previous = signature
                events.append(event)
            except (ValueError, KeyError, TypeError, AttributeError) as exc:
                raise ProtocolError("invalid journal record") from exc
        return events

    @contextmanager
    def writer(self):
        try:
            fd = os.open(self.lock_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        except FileExistsError as exc:
            raise HarnessError("journal writer lock exists; verify its owner before recovery") from exc
        try:
            with os.fdopen(fd, "w") as lock:
                lock.write(str(os.getpid()))
            self.events = self._load()
            yield self
        finally:
            self.lock_path.unlink(missing_ok=True)

    def append(self, kind: str, **data) -> dict:
        if not self.lock_path.exists():
            raise HarnessError("writer lock required")
        event = dict(seq=len(self.events), previous=self.events[-1]["hash"] if self.events else "0" * 64,
                     type=kind, data=data)
        event["hash"] = digest(canonical(event))
        with self.path.open("ab") as handle:
            handle.write(canonical(event) + b"\n")
            handle.flush()
            os.fsync(handle.fileno())
        self.events.append(event)
        return event


SCHEMAS = {
    "read_file": {"path": str},
    "write_file": {"path": str, "text": str, "expected_sha256": str},
    "check_file": {"path": str, "expected_text": str},
}


@dataclass(frozen=True)
class Policy:
    allow_write: bool = False
    max_read_bytes: int = 65536


class Tools:
    def __init__(self, workspace: Path, policy: Policy):
        self.workspace = workspace.resolve(strict=True)
        self.policy = policy

    def checked_path(self, value: str) -> Path:
        target = (self.workspace / value).resolve()
        if not target.is_relative_to(self.workspace):
            raise PermissionError("path escapes workspace")
        relative = target.relative_to(self.workspace)
        if relative.parts and relative.parts[0] == ".harness":
            raise PermissionError("harness state is reserved")
        return target

    def execute(self, call: Call) -> dict:
        schema = SCHEMAS.get(call.name)
        if schema is None or set(call.arguments) != set(schema):
            return dict(status="invalid", content="unknown tool or fields")
        if any(type(call.arguments[k]) is not t for k, t in schema.items()):
            return dict(status="invalid", content="wrong argument type")
        try:
            path = self.checked_path(call.arguments["path"])
            if call.name == "write_file":
                if not self.policy.allow_write:
                    raise PermissionError("write denied by policy")
                old = path.read_bytes() if path.exists() else None
                actual = digest(old) if old is not None else "missing"
                if actual != call.arguments["expected_sha256"]:
                    return dict(status="conflict", content="file changed; read again before proposing a write")
                # Teaching implementation: optimistic check + ordinary write is NOT atomic compare-and-swap.
                payload = call.arguments["text"].encode()
                if len(payload) > self.policy.max_read_bytes:
                    raise ValueError("write exceeds teaching size limit")
                with path.open("wb") as handle:
                    handle.write(payload)
                    handle.flush()
                    os.fsync(handle.fileno())
                return dict(status="ok", content=json.dumps({"sha256": digest(payload)}, sort_keys=True))
            with path.open("rb") as handle:
                raw = handle.read(self.policy.max_read_bytes + 1)
            if len(raw) > self.policy.max_read_bytes:
                raise ValueError("file exceeds read limit")
            text = raw.decode("utf-8")
            if call.name == "read_file":
                return dict(status="ok", content=json.dumps({"sha256": digest(raw), "text": text}, ensure_ascii=False))
            matched = text == call.arguments["expected_text"]
            return dict(status="ok" if matched else "error", content=json.dumps({"matched": matched}))
        except PermissionError as exc:
            return dict(status="denied", content=str(exc))
        except (OSError, ValueError, UnicodeError) as exc:
            return dict(status="error", content=str(exc))


@dataclass(frozen=True)
class Limits:
    model_requests: int = 8
    tool_calls: int = 12
    preview_chars: int = 2000
    transient_retries: int = 1


class Engine:
    def __init__(self, journal: Journal, tools: Tools, model: Model, *, limits: Limits = Limits(),
                 cancel: threading.Event | None = None):
        self.journal, self.tools, self.model = journal, tools, model
        self.limits = limits
        self.cancel = cancel or threading.Event()

    def events(self, kind: str) -> list[dict]:
        return [e for e in self.journal.events if e["type"] == kind]

    def messages(self) -> list[dict]:
        # The model projection excludes timing/administrative events.
        result = []
        for event in self.journal.events:
            if event["type"] == "user.input":
                result.append(dict(role="user", content=event["data"]["text"]))
            elif event["type"] == "model.reply":
                result.append(dict(role="assistant", **event["data"]))
            elif event["type"] == "tool.result":
                result.append(dict(role="tool", **event["data"]))
        return result

    def _finish_tool(self, call: Call, result: dict) -> None:
        content = result["content"]
        if len(content) > self.limits.preview_chars:
            artifacts = self.journal.path.parent / "artifacts"
            artifacts.mkdir(exist_ok=True)
            raw = content.encode()
            key = digest(raw)
            target = artifacts / (key + ".txt")
            with target.open("wb") as handle:
                handle.write(raw)
                handle.flush()
                os.fsync(handle.fileno())
            result = dict(result, content=content[:self.limits.preview_chars] + "\n[truncated]",
                          artifact="artifacts/" + target.name, sha256=key)
        self.journal.append("tool.result", call_id=call.id, name=call.name, **result)

    def _recover(self) -> None:
        completed = {e["data"]["call_id"] for e in self.events("tool.result")}
        for event in self.events("tool.started"):
            data = event["data"]
            if data["id"] not in completed:
                self._finish_tool(Call(data["id"], data["name"], data["arguments"]),
                                  dict(status="unknown", content="execution started, durable outcome missing; inspect external state"))
                completed.add(data["id"])
        if any(e["data"]["status"] == "unknown" for e in self.events("tool.result")):
            raise ReviewRequired("unknown outcome exists; do not auto-retry this session")

    def _pending(self) -> list[Call]:
        done = {e["data"]["call_id"] for e in self.events("tool.result")}
        return [Call(**call) for e in self.events("model.reply") for call in e["data"]["calls"] if call["id"] not in done]

    def _check_cancel(self) -> None:
        if self.cancel.is_set():
            self.journal.append("run.cancelled")
            raise HarnessError("cancelled at cooperative boundary")

    def run(self, goal: str, acceptance: Callable[[], bool]) -> str:
        with self.journal.writer():
            if not self.journal.events:
                self.journal.append("session.started", version=1, workspace=str(self.tools.workspace), policy=asdict(self.tools.policy))
                self.journal.append("user.input", text=goal)
            header = self.events("session.started")[0]["data"]
            if header["version"] != 1 or header["workspace"] != str(self.tools.workspace) or header["policy"] != asdict(self.tools.policy):
                raise ProtocolError("session version, workspace or policy changed")
            if self.events("user.input")[0]["data"]["text"] != goal:
                raise ProtocolError("new goal needs a new session")
            self._recover()
            finished = self.events("run.finished")
            if finished:
                if not acceptance():
                    raise VerificationFailed("saved completion no longer matches workspace")
                return finished[-1]["data"]["answer"]
            while True:
                self._check_cancel()
                for call in self._pending():
                    self._check_cancel()
                    if len(self.events("tool.started")) >= self.limits.tool_calls:
                        raise BudgetExceeded("tool budget exhausted")
                    # Durable intent BEFORE any possible side effect.
                    self.journal.append("tool.started", **asdict(call))
                    self._finish_tool(call, self.tools.execute(call))
                replies = self.events("model.reply")
                if replies and not replies[-1]["data"]["calls"]:
                    accepted = bool(acceptance())
                    self.journal.append("acceptance.checked", accepted=accepted)
                    if not accepted:
                        raise VerificationFailed("model answer does not satisfy acceptance")
                    answer = replies[-1]["data"]["text"]
                    self.journal.append("run.finished", answer=answer)
                    return answer
                reply = None
                for attempt in range(self.limits.transient_retries + 1):
                    if len(self.events("model.started")) >= self.limits.model_requests:
                        raise BudgetExceeded("model request budget exhausted")
                    self._check_cancel()
                    self.journal.append("model.started", step=len(replies), attempt=attempt)
                    try:
                        reply = decode_stream(self.model.stream(self.messages(), len(replies)))
                        self._check_cancel()
                        break
                    except TransientModelError as exc:
                        self.journal.append("model.retryable_error", error=str(exc))
                        if attempt == self.limits.transient_retries:
                            raise
                assert reply is not None
                used = {c["id"] for e in replies for c in e["data"]["calls"]}
                if any(c.id in used for c in reply.calls):
                    raise ProtocolError("tool call id reused across replies")
                self.journal.append("model.reply", text=reply.text, calls=[asdict(c) for c in reply.calls])


class DemoModel:
    """Scripted provider: exercise the same host boundaries without a model API."""
    def stream(self, messages: list[dict], step: int) -> Iterable[dict]:
        if step == 0:
            reply = Reply("先读取实际文件。", (Call("read-1", "read_file", {"path": "hello.txt"}),))
        elif step == 1:
            previous = next(m for m in reversed(messages) if m["role"] == "tool")
            if previous["status"] != "ok":
                reply = Reply("读取失败，停止。")
            else:
                observed = json.loads(previous["content"])
                reply = Reply("以读取到的哈希为前置条件修复拼写。", (Call("write-1", "write_file", {
                    "path": "hello.txt", "text": "hello, harness\n", "expected_sha256": observed["sha256"]}),))
        elif step == 2:
            reply = Reply("检查文件内容。", (Call("check-1", "check_file", {
                "path": "hello.txt", "expected_text": "hello, harness\n"}),))
        else:
            reply = Reply("工具阶段结束，是否完成由宿主验收。")
        yield {"type": "text.delta", "text": reply.text}
        for call in reply.calls:
            raw = json.dumps(call.arguments, ensure_ascii=False)
            middle = len(raw) // 2
            yield {"type": "tool.delta", "id": call.id, "name": call.name, "arguments": raw[:middle]}
            yield {"type": "tool.delta", "id": call.id, "arguments": raw[middle:]}
        yield {"type": "reply.end", "reason": "stop"}

"""Run from repository root: python3 lab/demo.py --workspace /tmp/harness-book-demo"""
from pathlib import Path
import argparse
import json
from harness import DemoModel, Engine, HarnessError, Journal, Policy, Tools

parser = argparse.ArgumentParser()
parser.add_argument('--workspace', type=Path, required=True)
parser.add_argument('--read-only', action='store_true')
args = parser.parse_args()
workspace = args.workspace.resolve()
workspace.mkdir(parents=True, exist_ok=True)
target = workspace / 'hello.txt'
if not target.exists():
    target.write_text('helo, harness\n', encoding='utf-8')
journal = Journal(workspace / '.harness' / 'events.jsonl')
tools = Tools(workspace, Policy(allow_write=not args.read_only))
def acceptance():
    successful_check = any(e['type'] == 'tool.result' and e['data']['name'] == 'check_file'
                           and e['data']['status'] == 'ok' for e in journal.events)
    return successful_check and target.read_text(encoding='utf-8') == 'hello, harness\n'
try:
    answer = Engine(journal, tools, DemoModel()).run('修正 hello.txt 的拼写并检查结果', acceptance)
    print(json.dumps({'status': 'verified', 'answer': answer, 'events': len(journal.events)}, ensure_ascii=False))
except HarnessError as exc:
    print(json.dumps({'status': 'stopped', 'reason': str(exc)}, ensure_ascii=False))
    raise SystemExit(2)