null: agentloop -- budgeted agent tool-loop with a hash-chained JSONL transcript; 55 tests + independent tamper leg, negative control held

wizard-work/null/agentloop/CHECK

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+# agentloop gate -- one command, and the exit code IS the verdict.
+#
+#   0  every leg passed
+#   1  a graded fault (a leg reported one)
+#   2  could not grade (no python3, the package will not import, ...)
+#
+# Resolves its own directory, so the copy in a clone grades that clone.
+#
+# The last line is `exit "$rc"`, not a constant: a gate whose exit code is a
+# literal cannot fail, and a gate that cannot fail is not evidence.
+set -u
+
+here="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+py=""
+if command -v python3 >/dev/null 2>&1; then
+    py="python3"
+fi
+
+if [ -z "$py" ]; then
+    echo "COULD NOT GRADE: no python3 on PATH"
+    rc=2
+else
+    "$py" "$here/check.py"
+    rc=$?
+fi
+
+exit "$rc"

wizard-work/null/agentloop/README-agentloop.md

@@ -0,0 +1,117 @@
+# agentloop — a budgeted agent tool-loop with a hash-chained transcript
+
+Lane: Codeforge. Stdlib only, Python 3.8+. No third-party imports anywhere.
+
+    python3 check.py          # the gate; exit code IS the verdict
+    python3 demo.py           # three end-to-end runs, writes demo-*.jsonl
+
+## What it is
+
+Three moving parts, each small enough to read in one sitting:
+
+| file | role |
+| --- | --- |
+| `agentloop/tools.py` | `Tool` registry + a JSON-Schema-subset argument validator |
+| `agentloop/loop.py` | the budgeted loop: model turn -> tool calls -> tool results -> repeat |
+| `agentloop/transcript.py` | append-only JSONL log whose records form a SHA-256 hash chain |
+
+A model is any callable `(messages, manifest) -> ModelReply`. The loop never
+touches a clock or a random source, so **a transcript is a pure function of
+`(task, model, tools, budgets)`** — run the same loop twice and the transcripts
+are byte-identical. That determinism is what makes the hash chain a usable
+integrity check instead of decoration.
+
+## The transcript is the point
+
+Every record carries the hash of the record before it:
+
+```json
+{"hash":"cc7b32...","kind":"tool","payload":{...},"prev":"1f08fd...","seq":2}
+```
+
+So the file is a chain, and verification breaks at the *first* affected record
+and names it. A single file digest cannot do that — it says "something changed",
+not "record 3 was rewritten".
+
+`verify_records` reports, per record index: `hash_mismatch`, `seq_gap`,
+`prev_mismatch`, `missing_fields`, `not_an_object`, `unserialisable_payload`.
+`verify_file` additionally reports `bad_json` (with the line number) and
+`unreadable`, and returns the record count.
+
+### What the chain does NOT prove — stated plainly
+
+A hash chain detects **modification and reordering of the records that are
+present**. It cannot detect that records are **missing from the end**: delete
+the last line and the remaining chain still verifies, because every surviving
+record still points at a real predecessor. Completeness needs an anchor outside
+the transcript — a commit SHA, or a digest stored elsewhere. `verify_file`
+returns the count so a caller holding an external expectation can compare it.
+`test_truncation_is_detected_only_against_an_external_count` pins this limit
+rather than hiding it.
+
+## Stop reasons and their precedence
+
+| reason | meaning |
+| --- | --- |
+| `no_tool_calls` | the assistant asked for no tools — the ordinary end of a run |
+| `sentinel` | the assistant's own text contained `stop_sentinel` |
+| `step_budget` | checked *before* each turn, so `steps` never exceeds `max_steps` |
+| `token_budget` | cumulative estimate exceeded `max_tokens`; the over-budget turn is recorded, then the loop stops **without** running its tool calls |
+| `model_error` | the model callable raised, or returned a non-`ModelReply` |
+
+Within one turn the order is: record the turn, then sentinel, then
+`no_tool_calls`, then `token_budget`. A model that says DONE and asks for a tool
+in the same turn stops on the sentinel and never runs the tool.
+
+Token accounting is `max(1, len(text) // 4)` — a heuristic, not a tokenizer. It
+bounds the loop, which is the property that matters; it does not claim to
+measure a real model's bill.
+
+## Tool failures are data, not crashes
+
+An unknown tool name, a schema violation, and a tool that raises are all turned
+into a `tool` record with `ok: false` and an `error` string, and the loop
+continues. `demo-recover.jsonl` shows all three in one run:
+
+```
+seq=2 multiply ok=False error="no tool named 'multiply' (have: add, echo, fail, greet)"
+seq=4 add      ok=False error="invalid arguments: $.add.a: expected type integer, got str"
+seq=6 fail     ok=False error="tool 'fail' raised RuntimeError: the tool is broken on purpose"
+seq=8 add      ok=True  value={'sum': 42}
+```
+
+A tool that returns a non-JSON-serialisable value is refused the same way: a
+result that cannot be recorded is not a result, and the transcript must always
+stay writable.
+
+## Evidence
+
+Measured on this host, 2026-10-25, Python 3.14.7:
+
+| claim | status | proof |
+| --- | --- | --- |
+| the gate passes | measured | `python3 check.py` -> 3 legs ok, `Ran 55 tests`, exit 0 |
+| the chain catches a payload rewrite at the right record | measured | suite `TestTamperDetection` (11 cases) + `check.py` tamper leg |
+| the chain catches a rewrite *and* recomputed hash | measured | `test_forged_record_with_a_recomputed_hash_still_breaks_the_chain` — the *next* record's `prev` breaks |
+| the chain catches a reorder even after renumbering | measured | `test_reorder_plus_renumber_is_caught_as_a_prev_mismatch` |
+| the chain cannot detect a missing tail | measured (limit, not a fault) | `test_truncation_is_detected_only_against_an_external_count` |
+| identical runs give byte-identical transcripts | measured | `test_identical_runs_produce_byte_identical_transcripts` |
+| step budget caps the loop and equals the assistant record count | measured | `test_step_budget_caps_the_loop_and_matches_the_record_count` |
+| token budget stops before running tools | measured | `test_token_budget_stops_the_loop_before_running_tools` |
+| three demo transcripts verify from disk | measured | `demo-answer.jsonl` 5, `demo-recover.jsonl` 11, `demo-budget.jsonl` 9 records, all `ok=True` |
+| the gate FAILS when the verifier is broken | measured | negative control: sabotaged copy -> 9 suite failures + tamper leg `a mutated transcript VERIFIED`, exit 1 |
+
+### Negative control
+
+A gate that cannot be shown to fail is not evidence. `check.py`'s tamper leg is
+independent of the suite — it re-derives its expectation from the file's own
+bytes, so a bug in the test fixtures cannot hide there. The sabotage run
+confirmed both legs fail together when `verify_records` is made to accept
+everything.
+
+## Replay
+
+```
+python3 check.py     # 0 = every leg passed, 1 = graded fault, 2 = could not grade
+python3 demo.py      # 0 = every demo claim held
+```

wizard-work/null/agentloop/agentloop/__init__.py

@@ -0,0 +1,75 @@
+"""agentloop -- a dependency-free agentic tool-use loop with a hash-chained transcript.
+
+Lane: Codeforge.  Stdlib only, Python 3.8+.  No third-party imports anywhere.
+
+The three moving parts:
+
+  tools.py      a Tool registry and a JSON-Schema-subset argument validator
+  loop.py       the budgeted agent loop (model -> tool calls -> tool results)
+  transcript.py an append-only JSONL log whose records form a SHA-256 hash chain
+
+The transcript is the point.  An agent run is evidence, and evidence that
+cannot be checked is just a log: every record carries the hash of the record
+before it, so a rewrite, a dropped record or a swap breaks the chain at the
+first affected record and names it.  See README-agentloop.md for what the
+chain does and does not prove.
+"""
+
+from .transcript import (
+    GENESIS,
+    KIND_ASSISTANT,
+    KIND_FINAL,
+    KIND_TOOL,
+    KIND_USER,
+    Transcript,
+    TranscriptError,
+    canonical_json,
+    verify_file,
+    verify_records,
+)
+from .tools import (
+    Tool,
+    ToolCall,
+    ToolError,
+    ToolNotFound,
+    ToolRegistry,
+    ToolResult,
+    ValidationError,
+    validate,
+)
+from .loop import (
+    AgentLoop,
+    LoopResult,
+    ModelReply,
+    StopReason,
+    estimate_tokens,
+)
+
+__version__ = "1.0.0"
+
+__all__ = [
+    "GENESIS",
+    "KIND_ASSISTANT",
+    "KIND_FINAL",
+    "KIND_TOOL",
+    "KIND_USER",
+    "Transcript",
+    "TranscriptError",
+    "canonical_json",
+    "verify_file",
+    "verify_records",
+    "Tool",
+    "ToolCall",
+    "ToolError",
+    "ToolNotFound",
+    "ToolRegistry",
+    "ToolResult",
+    "ValidationError",
+    "validate",
+    "AgentLoop",
+    "LoopResult",
+    "ModelReply",
+    "StopReason",
+    "estimate_tokens",
+    "__version__",
+]

wizard-work/null/agentloop/agentloop/loop.py

@@ -0,0 +1,218 @@
+"""The budgeted agent loop: model turn -> tool calls -> tool results -> repeat.
+
+Determinism
+-----------
+The loop takes no clock and no randomness.  A transcript is therefore a pure
+function of ``(task, model, tools, budgets)``: run the same loop twice and the
+two transcripts are byte-identical.  Tests/test_agentloop.py asserts exactly
+that, which is what makes the hash chain a usable integrity check rather than
+decoration.
+
+Stop reasons, and their precedence
+-----------------------------------
+  step_budget    checked BEFORE each model turn, so ``steps`` never exceeds
+                 ``max_steps`` and always equals the number of assistant records.
+  sentinel       the assistant's own text contains ``stop_sentinel``.
+  no_tool_calls  the assistant asked for no tools -- the ordinary end of a run.
+  token_budget   cumulative estimated tokens exceeded ``max_tokens``; the
+                 over-budget turn is recorded, then the loop stops WITHOUT
+                 executing its tool calls.
+  model_error    the model callable raised, or returned a non-ModelReply.
+
+Within one turn the order is: record the turn, then sentinel, then
+no_tool_calls, then token_budget.  A model that says DONE and asks for a tool
+in the same turn stops on the sentinel and never runs the tool; a model that
+spends its last token on tool calls stops on the budget and never runs them.
+
+Token accounting is a 4-chars-per-token heuristic, not a tokenizer.  It bounds
+the loop, which is the property that matters; it does not claim to measure a
+real model's bill.
+"""
+
+from __future__ import annotations
+
+import json
+
+from .tools import ToolCall, ToolError, ToolRegistry, ToolResult, ValidationError
+from .transcript import KIND_ASSISTANT, KIND_FINAL, KIND_TOOL, KIND_USER, Transcript, canonical_json
+
+__all__ = ["StopReason", "ModelReply", "LoopResult", "AgentLoop", "estimate_tokens"]
+
+
+class StopReason(object):
+    NO_TOOL_CALLS = "no_tool_calls"
+    SENTINEL = "sentinel"
+    STEP_BUDGET = "step_budget"
+    TOKEN_BUDGET = "token_budget"
+    MODEL_ERROR = "model_error"
+
+
+class ModelReply(object):
+    """What a model returns for one turn."""
+
+    __slots__ = ("text", "tool_calls")
+
+    def __init__(self, text="", tool_calls=None):
+        self.text = text or ""
+        self.tool_calls = list(tool_calls) if tool_calls else []
+
+    def __repr__(self):
+        return "ModelReply(%r, %r)" % (self.text, self.tool_calls)
+
+
+class LoopResult(object):
+    """The outcome of a run: messages, the budget accounting, and the log."""
+
+    __slots__ = ("messages", "steps", "tokens", "stop_reason", "transcript", "model_error")
+
+    def __init__(self, messages, steps, tokens, stop_reason, transcript, model_error=None):
+        self.messages = messages
+        self.steps = steps
+        self.tokens = tokens
+        self.stop_reason = stop_reason
+        self.transcript = transcript
+        self.model_error = model_error
+
+    @property
+    def ok(self):
+        return self.stop_reason in (StopReason.NO_TOOL_CALLS, StopReason.SENTINEL)
+
+    def tool_results(self):
+        """The ToolResult records, in order, as recorded in the transcript."""
+        out = []
+        for record in self.transcript.records:
+            if record["kind"] == KIND_TOOL:
+                p = record["payload"]
+                out.append(ToolResult(p["name"], p["ok"], p["value"], p["error"]))
+        return out
+
+    def __repr__(self):
+        return "LoopResult(stop=%r, steps=%d, tokens=%d)" % (
+            self.stop_reason,
+            self.steps,
+            self.tokens,
+        )
+
+
+def estimate_tokens(text):
+    """4 chars/token, floor 1 for non-empty text.  A heuristic, documented."""
+    if not text:
+        return 0
+    return max(1, len(text) // 4)
+
+
+class AgentLoop(object):
+    def __init__(self, registry, max_steps=8, max_tokens=4000, stop_sentinel=None):
+        if not isinstance(registry, ToolRegistry):
+            raise TypeError("registry must be a ToolRegistry, got %s" % type(registry).__name__)
+        if not isinstance(max_steps, int) or isinstance(max_steps, bool) or max_steps < 1:
+            raise ValueError("max_steps must be a positive integer")
+        if not isinstance(max_tokens, int) or isinstance(max_tokens, bool) or max_tokens < 1:
+            raise ValueError("max_tokens must be a positive integer")
+        self.registry = registry
+        self.max_steps = max_steps
+        self.max_tokens = max_tokens
+        self.stop_sentinel = stop_sentinel
+
+    # -- one tool call ---------------------------------------------------
+
+    def _execute(self, call):
+        try:
+            tool = self.registry.get(call.name)
+        except ToolError as exc:
+            return ToolResult(call.name, False, None, str(exc))
+        try:
+            value = tool(call.arguments)
+        except ValidationError as exc:
+            return ToolResult(call.name, False, None, "invalid arguments: %s" % exc)
+        except ToolError as exc:
+            return ToolResult(call.name, False, None, str(exc))
+        except Exception as exc:  # defensive: a transcript must stay writable
+            return ToolResult(
+                call.name, False, None, "tool raised %s: %s" % (type(exc).__name__, exc)
+            )
+        # A value that cannot be serialised cannot enter the transcript, and a
+        # result that cannot be recorded is not a result.
+        try:
+            canonical_json(value)
+        except (TypeError, ValueError) as exc:
+            return ToolResult(
+                call.name, False, None, "tool returned a non-serialisable value: %s" % exc
+            )
+        return ToolResult(call.name, True, value)
+
+    # -- the loop --------------------------------------------------------
+
+    def run(self, task, model, transcript=None):
+        if not callable(model):
+            raise TypeError("model must be callable")
+        tr = transcript if transcript is not None else Transcript()
+        tr.append(KIND_USER, {"text": task})
+
+        messages = [{"role": "user", "content": task}]
+        steps = 0
+        tokens = 0
+        stop_reason = None
+        model_error = None
+
+        while True:
+            if steps >= self.max_steps:
+                stop_reason = StopReason.STEP_BUDGET
+                break
+
+            try:
+                reply = model(messages, self.registry.manifest())
+            except Exception as exc:
+                stop_reason = StopReason.MODEL_ERROR
+                model_error = "model raised %s: %s" % (type(exc).__name__, exc)
+                break
+            if not isinstance(reply, ModelReply):
+                stop_reason = StopReason.MODEL_ERROR
+                model_error = "model returned %s, expected ModelReply" % type(reply).__name__
+                break
+
+            steps += 1
+            tokens += estimate_tokens(reply.text)
+
+            calls = [
+                c if isinstance(c, ToolCall) else ToolCall(**c) for c in (reply.tool_calls or [])
+            ]
+            recorded_calls = [c.as_dict() for c in calls]
+            tr.append(KIND_ASSISTANT, {"text": reply.text, "tool_calls": recorded_calls})
+            messages.append(
+                {"role": "assistant", "content": reply.text, "tool_calls": recorded_calls}
+            )
+
+            if self.stop_sentinel and self.stop_sentinel in reply.text:
+                stop_reason = StopReason.SENTINEL
+                break
+            if not calls:
+                stop_reason = StopReason.NO_TOOL_CALLS
+                break
+            if tokens > self.max_tokens:
+                stop_reason = StopReason.TOKEN_BUDGET
+                break
+
+            for call in calls:
+                result = self._execute(call)
+                tr.append(KIND_TOOL, result.as_dict())
+                content = json.dumps(result.value) if result.ok else result.error
+                messages.append(
+                    {
+                        "role": "tool",
+                        "content": content,
+                        "name": call.name,
+                        "ok": result.ok,
+                    }
+                )
+
+        tr.append(
+            KIND_FINAL,
+            {
+                "stop_reason": stop_reason,
+                "steps": steps,
+                "tokens": tokens,
+                "model_error": model_error,
+            },
+        )
+        return LoopResult(messages, steps, tokens, stop_reason, tr, model_error)

wizard-work/null/agentloop/agentloop/tools.py

@@ -0,0 +1,260 @@
+"""Tool registry, and a JSON-Schema-subset validator for tool arguments.
+
+The validator covers the subset a tool loop actually needs: ``type`` (including
+the integer/number distinction, where ``1`` is a valid integer and ``True`` is
+not), ``enum``, ``required``, ``additionalProperties``, ``items``, ``properties``
+and the numeric/string bounds.  It is deliberately not a full JSON Schema
+implementation -- ``allOf``/``$ref``/``format`` are absent, and claiming they
+were supported would be a lie a caller eventually finds out.
+
+``bool`` is checked before ``int`` because ``bool`` is a subclass of ``int`` in
+Python: without that ordering ``True`` would validate as the integer ``1``.
+"""
+
+from __future__ import annotations
+
+import re
+
+__all__ = [
+    "ToolError",
+    "ToolNotFound",
+    "ValidationError",
+    "ToolCall",
+    "ToolResult",
+    "Tool",
+    "ToolRegistry",
+    "validate",
+]
+
+
+class ToolError(Exception):
+    """Base class for every tool failure the loop can report."""
+
+
+class ToolNotFound(ToolError):
+    """The model named a tool that is not registered."""
+
+
+class ValidationError(ToolError):
+    """The model's arguments do not satisfy the tool's schema."""
+
+
+def _is_type(instance, name):
+    if name == "object":
+        return isinstance(instance, dict)
+    if name == "array":
+        return isinstance(instance, list)
+    if name == "string":
+        return isinstance(instance, str)
+    if name == "boolean":
+        return isinstance(instance, bool)
+    if name == "null":
+        return instance is None
+    if name == "integer":
+        return isinstance(instance, int) and not isinstance(instance, bool)
+    if name == "number":
+        return isinstance(instance, (int, float)) and not isinstance(instance, bool)
+    # An unknown type name is a schema bug, not a caller bug: fail loudly.
+    raise ValidationError("unknown type %r in schema" % (name,))
+
+
+def validate(instance, schema, path="$"):
+    """Raise ``ValidationError`` unless ``instance`` satisfies ``schema``."""
+    if schema is True or schema == {}:
+        return
+    if schema is False:
+        raise ValidationError("%s: no value is allowed here" % path)
+    if not isinstance(schema, dict):
+        raise ValidationError(
+            "%s: schema must be an object, got %s" % (path, type(schema).__name__)
+        )
+
+    if "enum" in schema:
+        if instance not in schema["enum"]:
+            raise ValidationError(
+                "%s: %r is not one of %r" % (path, instance, schema["enum"])
+            )
+        return
+
+    if "type" in schema:
+        declared = schema["type"]
+        names = declared if isinstance(declared, list) else [declared]
+        if not any(_is_type(instance, n) for n in names):
+            raise ValidationError(
+                "%s: expected type %s, got %s" % (path, declared, type(instance).__name__)
+            )
+
+    if isinstance(instance, str):
+        if "minLength" in schema and len(instance) < schema["minLength"]:
+            raise ValidationError(
+                "%s: shorter than minLength %d" % (path, schema["minLength"])
+            )
+        if "maxLength" in schema and len(instance) > schema["maxLength"]:
+            raise ValidationError(
+                "%s: longer than maxLength %d" % (path, schema["maxLength"])
+            )
+        if "pattern" in schema and re.search(schema["pattern"], instance) is None:
+            raise ValidationError(
+                "%s: %r does not match pattern %r" % (path, instance, schema["pattern"])
+            )
+
+    if isinstance(instance, bool):
+        return
+
+    if isinstance(instance, (int, float)):
+        if "minimum" in schema and instance < schema["minimum"]:
+            raise ValidationError("%s: %r is below minimum %r" % (path, instance, schema["minimum"]))
+        if "maximum" in schema and instance > schema["maximum"]:
+            raise ValidationError("%s: %r is above maximum %r" % (path, instance, schema["maximum"]))
+
+    if isinstance(instance, list):
+        if "minItems" in schema and len(instance) < schema["minItems"]:
+            raise ValidationError(
+                "%s: %d items is below minItems %d" % (path, len(instance), schema["minItems"])
+            )
+        if "maxItems" in schema and len(instance) > schema["maxItems"]:
+            raise ValidationError(
+                "%s: %d items is above maxItems %d" % (path, len(instance), schema["maxItems"])
+            )
+        if "items" in schema:
+            for i, item in enumerate(instance):
+                validate(item, schema["items"], "%s[%d]" % (path, i))
+
+    if isinstance(instance, dict):
+        for required in schema.get("required", []):
+            if required not in instance:
+                raise ValidationError("%s: missing required property %r" % (path, required))
+        properties = schema.get("properties", {})
+        for key, value in instance.items():
+            if key in properties:
+                validate(value, properties[key], "%s.%s" % (path, key))
+            elif schema.get("additionalProperties") is False:
+                raise ValidationError("%s: unexpected property %r" % (path, key))
+
+
+class ToolCall(object):
+    """One tool invocation requested by the model."""
+
+    __slots__ = ("name", "arguments")
+
+    def __init__(self, name, arguments=None):
+        self.name = name
+        self.arguments = dict(arguments) if arguments else {}
+
+    def __repr__(self):
+        return "ToolCall(%r, %r)" % (self.name, self.arguments)
+
+    def __eq__(self, other):
+        return (
+            isinstance(other, ToolCall)
+            and self.name == other.name
+            and self.arguments == other.arguments
+        )
+
+    def as_dict(self):
+        return {"name": self.name, "arguments": self.arguments}
+
+
+class ToolResult(object):
+    """The outcome of one tool invocation, as recorded in the transcript."""
+
+    __slots__ = ("name", "ok", "value", "error")
+
+    def __init__(self, name, ok, value=None, error=None):
+        self.name = name
+        self.ok = bool(ok)
+        self.value = value
+        self.error = error
+
+    def __repr__(self):
+        if self.ok:
+            return "ToolResult(%r, ok=True, value=%r)" % (self.name, self.value)
+        return "ToolResult(%r, ok=False, error=%r)" % (self.name, self.error)
+
+    def __eq__(self, other):
+        return (
+            isinstance(other, ToolResult)
+            and self.name == other.name
+            and self.ok == other.ok
+            and self.value == other.value
+            and self.error == other.error
+        )
+
+    def as_dict(self):
+        return {"name": self.name, "ok": self.ok, "value": self.value, "error": self.error}
+
+
+class Tool(object):
+    """A named, described, schema-checked callable."""
+
+    def __init__(self, name, description, parameters, fn):
+        if not isinstance(name, str) or not name:
+            raise ValueError("tool name must be a non-empty string")
+        if not callable(fn):
+            raise ValueError("tool %r: fn must be callable" % name)
+        if not isinstance(parameters, dict):
+            raise ValueError("tool %r: parameters must be a JSON-schema object" % name)
+        self.name = name
+        self.description = description or ""
+        self.parameters = parameters
+        self.fn = fn
+
+    def __call__(self, arguments=None):
+        arguments = dict(arguments) if arguments else {}
+        validate(arguments, self.parameters, "$.%s" % self.name)
+        try:
+            return self.fn(**arguments)
+        except ToolError:
+            raise
+        except Exception as exc:
+            raise ToolError(
+                "tool %r raised %s: %s" % (self.name, type(exc).__name__, exc)
+            )
+
+    def manifest(self):
+        return {
+            "name": self.name,
+            "description": self.description,
+            "parameters": self.parameters,
+        }
+
+    def __repr__(self):
+        return "Tool(%r)" % self.name
+
+
+class ToolRegistry(object):
+    """Name -> Tool, with duplicate rejection and a model-facing manifest."""
+
+    def __init__(self, tools=()):
+        self._tools = {}
+        for tool in tools:
+            self.add(tool)
+
+    def add(self, tool):
+        if not isinstance(tool, Tool):
+            raise TypeError("expected Tool, got %s" % type(tool).__name__)
+        if tool.name in self._tools:
+            raise ValueError("duplicate tool name %r" % tool.name)
+        self._tools[tool.name] = tool
+        return tool
+
+    def get(self, name):
+        try:
+            return self._tools[name]
+        except KeyError:
+            raise ToolNotFound("no tool named %r (have: %s)" % (name, ", ".join(sorted(self._tools))))
+
+    def has(self, name):
+        return name in self._tools
+
+    def names(self):
+        return sorted(self._tools)
+
+    def manifest(self):
+        return [self._tools[n].manifest() for n in sorted(self._tools)]
+
+    def __len__(self):
+        return len(self._tools)
+
+    def __contains__(self, name):
+        return name in self._tools

wizard-work/null/agentloop/agentloop/transcript.py

@@ -0,0 +1,230 @@
+"""Append-only JSONL transcript whose records form a SHA-256 hash chain.
+
+An agent transcript is evidence, and evidence that cannot be checked is just a
+log.  Every record carries the hash of the record before it, so the file is a
+chain: rewrite one payload, drop one record, or swap two, and verification
+breaks at the first affected record and names it.  A single file digest cannot
+do that -- it says "something changed", not "record 3 was rewritten".
+
+WHAT THIS DOES NOT PROVE -- stated plainly, because a gate that overclaims is
+worse than no gate.
+
+  A hash chain detects MODIFICATION and REORDERING of the records that are
+  present.  It cannot detect that records are MISSING FROM THE END: delete the
+  last line and the remaining chain still verifies, because every surviving
+  record still points at a real predecessor.  Completeness needs an anchor
+  outside the transcript -- a commit SHA, or a digest stored elsewhere.
+  verify_file therefore also returns the record count, so a caller holding an
+  external count can compare it.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+
+GENESIS = "0" * 64
+
+KIND_USER = "user"
+KIND_ASSISTANT = "assistant"
+KIND_TOOL = "tool"
+KIND_FINAL = "final"
+_KINDS = (KIND_USER, KIND_ASSISTANT, KIND_TOOL, KIND_FINAL)
+
+_FIELDS = ("seq", "kind", "payload", "prev", "hash")
+
+
+class TranscriptError(Exception):
+    """Raised when a transcript is malformed, or a record is rejected."""
+
+
+def canonical_json(obj):
+    """Serialise deterministically: sorted keys, no incidental whitespace.
+
+    Two runs of the same loop must produce byte-identical transcripts, so the
+    encoding has to be a function of the data and nothing else.
+    """
+    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def _record_hash(seq, kind, payload, prev):
+    body = canonical_json({"seq": seq, "kind": kind, "payload": payload, "prev": prev})
+    return hashlib.sha256(body.encode("utf-8")).hexdigest()
+
+
+class Transcript(object):
+    """An in-memory record list that can mirror itself to a JSONL file."""
+
+    def __init__(self, path=None):
+        self.path = path
+        self.records = []
+        self._prev = GENESIS
+        if path and os.path.exists(path):
+            self.load(path)
+
+    # -- writing ---------------------------------------------------------
+
+    def append(self, kind, payload):
+        if kind not in _KINDS:
+            raise TranscriptError(
+                "unknown record kind %r (expected one of: %s)" % (kind, ", ".join(_KINDS))
+            )
+        if not isinstance(payload, (dict, list)):
+            raise TranscriptError(
+                "payload must be a dict or list, got %s" % type(payload).__name__
+            )
+        # Reject a payload that cannot be written back out: a transcript that
+        # cannot be serialised is not a transcript.
+        try:
+            canonical_json(payload)
+        except (TypeError, ValueError) as exc:
+            raise TranscriptError("payload is not JSON-serialisable: %s" % exc)
+
+        seq = len(self.records)
+        digest = _record_hash(seq, kind, payload, self._prev)
+        record = {
+            "seq": seq,
+            "kind": kind,
+            "payload": payload,
+            "prev": self._prev,
+            "hash": digest,
+        }
+        self.records.append(record)
+        self._prev = digest
+        if self.path:
+            self._append_line(record)
+        return record
+
+    def _append_line(self, record):
+        with open(self.path, "a", encoding="utf-8") as fh:
+            fh.write(canonical_json(record) + "\n")
+
+    # -- reading ---------------------------------------------------------
+
+    def load(self, path=None):
+        path = path or self.path
+        if not path:
+            raise TranscriptError("no path given")
+        with open(path, "rb") as fh:
+            raw = fh.read()
+        records = []
+        for lineno, line in enumerate(raw.decode("utf-8", "replace").splitlines(), 1):
+            if not line.strip():
+                continue
+            try:
+                records.append(json.loads(line))
+            except ValueError as exc:
+                raise TranscriptError("line %d is not JSON: %s" % (lineno, exc))
+        ok, violations = verify_records(records)
+        if not ok:
+            raise TranscriptError("transcript failed verification: %s" % (violations[0],))
+        self.records = records
+        self._prev = records[-1]["hash"] if records else GENESIS
+        self.path = path
+        return self
+
+    def save(self, path=None):
+        path = path or self.path
+        if not path:
+            raise TranscriptError("no path given")
+        data = self.bytes()
+        tmp = path + ".tmp"
+        with open(tmp, "wb") as fh:
+            fh.write(data)
+        os.replace(tmp, path)
+        self.path = path
+        return path
+
+    def bytes(self):
+        return "".join(canonical_json(r) + "\n" for r in self.records).encode("utf-8")
+
+    def __len__(self):
+        return len(self.records)
+
+
+def verify_records(records):
+    """Return ``(ok, violations)``.
+
+    ``violations`` is a list of dicts, each naming the record index that broke
+    the chain and the reason.  The scan stops at the FIRST break: past a
+    broken link every later record's ``prev`` is wrong too, so continuing would
+    report one fault as many.  The first break is the one to fix.
+    """
+    violations = []
+    prev = GENESIS
+    for i, raw in enumerate(records):
+        if not isinstance(raw, dict):
+            violations.append(
+                {"seq": i, "reason": "not_an_object", "detail": type(raw).__name__}
+            )
+            break
+        missing = [f for f in _FIELDS if f not in raw]
+        if missing:
+            violations.append({"seq": i, "reason": "missing_fields", "detail": missing})
+            break
+        if raw["seq"] != i:
+            violations.append({"seq": i, "reason": "seq_gap", "detail": raw["seq"]})
+            break
+        if raw["prev"] != prev:
+            violations.append(
+                {
+                    "seq": i,
+                    "reason": "prev_mismatch",
+                    "detail": raw["prev"],
+                    "expected": prev,
+                }
+            )
+            break
+        try:
+            expected = _record_hash(raw["seq"], raw["kind"], raw["payload"], raw["prev"])
+        except (TypeError, ValueError) as exc:
+            violations.append({"seq": i, "reason": "unserialisable_payload", "detail": str(exc)})
+            break
+        if expected != raw["hash"]:
+            violations.append(
+                {
+                    "seq": i,
+                    "reason": "hash_mismatch",
+                    "detail": raw["hash"],
+                    "expected": expected,
+                }
+            )
+            break
+        prev = raw["hash"]
+    return (not violations), violations
+
+
+def verify_file(path):
+    """Verify a transcript file on disk.  Returns ``(ok, violations, count)``.
+
+    ``count`` is the number of records read.  Compare it against an external
+    expectation to detect truncation -- the chain alone cannot, see module doc.
+    """
+    try:
+        with open(path, "rb") as fh:
+            raw = fh.read()
+    except OSError as exc:
+        return False, [{"seq": None, "reason": "unreadable", "detail": str(exc)}], 0
+
+    records = []
+    for lineno, line in enumerate(raw.decode("utf-8", "replace").splitlines(), 1):
+        if not line.strip():
+            continue
+        try:
+            records.append(json.loads(line))
+        except ValueError as exc:
+            return (
+                False,
+                [
+                    {
+                        "seq": len(records),
+                        "reason": "bad_json",
+                        "line": lineno,
+                        "detail": str(exc),
+                    }
+                ],
+                len(records),
+            )
+    ok, violations = verify_records(records)
+    return ok, violations, len(records)

wizard-work/null/agentloop/check.py

@@ -0,0 +1,149 @@
+#!/usr/bin/env python3
+"""null lane gate for the agentloop artifact -- one command, exit code IS the verdict.
+
+    python3 check.py
+
+  0  every leg passed
+  1  a graded fault (a leg reported one)
+  2  COULD NOT GRADE (python3 missing, the package will not import, ...)
+
+Legs:
+  1. the package imports and exposes the documented API;
+  2. the unittest suite passes (chain integrity, tamper detection, argument
+     validation, loop budgets, determinism);
+  3. an INDEPENDENT tamper probe: build a transcript, mutate one payload byte on
+     disk, and require that verify_file names record 2 with reason hash_mismatch.
+     This leg does not reuse the suite's helpers -- it re-derives the expectation
+     from the file's own bytes, so a bug in the test fixtures cannot hide here.
+
+This file resolves its own directory, so the copy in a clone grades that clone.
+"""
+
+import os
+import subprocess
+import sys
+import tempfile
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+PY = sys.executable or "python3"
+
+
+def leg_import():
+    probe = os.path.join(HERE, "_probe_import.py")
+    with open(probe, "w", encoding="utf-8") as fh:
+        fh.write(
+            "import sys; sys.path.insert(0, %r)\n"
+            "import agentloop\n"
+            "missing = [n for n in ('Transcript','verify_file','Tool','ToolRegistry',"
+            "'AgentLoop','ModelReply','StopReason','ToolCall') if not hasattr(agentloop, n)]\n"
+            "sys.exit(3 if missing else 0)\n" % (os.path.join(HERE, ".."),)
+        )
+    try:
+        rc = subprocess.call([PY, probe])
+    finally:
+        os.unlink(probe)
+    if rc == 3:
+        return 1, "import leg: documented API is missing"
+    if rc != 0:
+        return 2, "import leg: probe exited %d" % rc
+    return 0, "import leg: agentloop imports, documented API present"
+
+
+def leg_suite():
+    env = dict(os.environ)
+    env["PYTHONPATH"] = os.path.join(HERE, "..") + os.pathsep + env.get("PYTHONPATH", "")
+    proc = subprocess.Popen(
+        [PY, "-m", "unittest", "discover", "-s", "tests", "-t", HERE, "-q"],
+        cwd=HERE, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+    )
+    out = proc.communicate()[0].decode("utf-8", "replace")
+    tail = [l for l in out.strip().splitlines() if l.strip()][-6:]
+    if proc.returncode != 0:
+        return 1, "suite leg: FAILED\n" + "\n".join(tail)
+    ran = [l for l in out.splitlines() if l.startswith("Ran ")]
+    return 0, "suite leg: PASS (%s)" % (ran[0] if ran else "no summary line")
+
+
+def leg_tamper():
+    """Independent probe: mutate a transcript on disk, require the right verdict."""
+    sys.path.insert(0, os.path.join(HERE, ".."))
+    try:
+        from agentloop import Transcript, canonical_json
+        from agentloop.transcript import KIND_ASSISTANT, KIND_FINAL, KIND_TOOL, KIND_USER
+    except Exception as exc:
+        return 2, "tamper leg: could not import agentloop (%s)" % exc
+
+    tr = Transcript()
+    tr.append(KIND_USER, {"text": "add 2 and 3"})
+    tr.append(KIND_ASSISTANT, {"text": "calling add", "tool_calls": []})
+    tr.append(KIND_TOOL, {"name": "add", "ok": True, "value": 5})
+    tr.append(KIND_FINAL, {"stop_reason": "no_tool_calls", "steps": 1, "tokens": 4})
+
+    fd, path = tempfile.mkstemp(suffix=".jsonl")
+    os.close(fd)
+    try:
+        with open(path, "wb") as fh:
+            fh.write(tr.bytes())
+
+        # Read the file's own bytes, flip one character of record 2's payload,
+        # and write it back.  Nothing here uses the suite's helpers.
+        with open(path, "rb") as fh:
+            raw = fh.read()
+        target = raw.index(b'"value":5')
+        flipped = raw[:target] + b'"value":9' + raw[target + len(b'"value":5'):]
+        self_check = flipped != raw
+        with open(path, "wb") as fh:
+            fh.write(flipped)
+
+        # Re-verify from disk, using only verify_file.
+        from agentloop import verify_file
+        ok, violations, count = verify_file(path)
+    finally:
+        os.unlink(path)
+
+    if not self_check:
+        return 2, "tamper leg: the mutation did not change the file"
+    if ok:
+        return 1, "tamper leg: a mutated transcript VERIFIED -- the chain is decorative"
+    if not violations:
+        return 1, "tamper leg: failed but named no violation"
+    v = violations[0]
+    if v.get("seq") != 2 or v.get("reason") != "hash_mismatch":
+        return 1, "tamper leg: wrong verdict -- got %r" % (v,)
+    if count != 4:
+        return 1, "tamper leg: expected 4 records, counted %d" % count
+    return 0, "tamper leg: mutation caught at seq 2 (hash_mismatch), 4 records intact"
+
+
+LEGS = (leg_import, leg_suite, leg_tamper)
+
+
+def main():
+    if not PY:
+        print("COULD NOT GRADE: no python3 interpreter")
+        return 2
+
+    rc = 0
+    print("agentloop gate -- %s" % HERE)
+    for leg in LEGS:
+        try:
+            code, message = leg()
+        except Exception as exc:
+            code, message = 2, "%s leg: raised %s: %s" % (leg.__name__, type(exc).__name__, exc)
+        print("  [%s] %s" % ("ok" if code == 0 else ("FAIL" if code == 1 else "N/A"), message))
+        if code == 1:
+            rc = 1
+        elif code == 2 and rc == 0:
+            rc = 2
+
+    if rc == 0:
+        print("VERDICT: every leg passed")
+    elif rc == 1:
+        print("VERDICT: graded fault")
+    else:
+        print("VERDICT: could not grade")
+    return rc
+
+
+if __name__ == "__main__":
+    sys.exit(main())

wizard-work/null/agentloop/demo-answer.jsonl

@@ -0,0 +1,5 @@
+{"hash":"5902d5e29878dc70da3b6794b74dd8479c3cea8951fa8e8ad39ec849b0c0e12b","kind":"user","payload":{"text":"What is 2 + 3?"},"prev":"0000000000000000000000000000000000000000000000000000000000000000","seq":0}
+{"hash":"1f08fd31fcf3136f44905485ffdd15b77eb3970f9048a4d3d2f37b44f61e7617","kind":"assistant","payload":{"text":"Let me add those.","tool_calls":[{"arguments":{"a":2,"b":3},"name":"add"}]},"prev":"5902d5e29878dc70da3b6794b74dd8479c3cea8951fa8e8ad39ec849b0c0e12b","seq":1}
+{"hash":"cc7b32293589a003b6bc75757c6b161744934de244299befd6f44b2d0950bda1","kind":"tool","payload":{"error":null,"name":"add","ok":true,"value":{"sum":5}},"prev":"1f08fd31fcf3136f44905485ffdd15b77eb3970f9048a4d3d2f37b44f61e7617","seq":2}
+{"hash":"b783e9591ef27741504c3a66889b65b04b21c1b00e60f1caef9394b586cfb0a4","kind":"assistant","payload":{"text":"2 + 3 = 5.","tool_calls":[]},"prev":"cc7b32293589a003b6bc75757c6b161744934de244299befd6f44b2d0950bda1","seq":3}
+{"hash":"93405bb51db5518e1078f639f60949bf08361901d54be2edeb3286db8ed2c59c","kind":"final","payload":{"model_error":null,"steps":2,"stop_reason":"no_tool_calls","tokens":6},"prev":"b783e9591ef27741504c3a66889b65b04b21c1b00e60f1caef9394b586cfb0a4","seq":4}

wizard-work/null/agentloop/demo-budget.jsonl

@@ -0,0 +1,9 @@
+{"hash":"9f2b12e6cea41d14dbd01bfa202219d2275bfe726637149d2552875f88139e38","kind":"user","payload":{"text":"Keep going forever."},"prev":"0000000000000000000000000000000000000000000000000000000000000000","seq":0}
+{"hash":"b9809a74113fe58c8ebd7fb896025c5a5b9cdd3834d7ecff77ea2efba6d3ba3e","kind":"assistant","payload":{"text":"still going (step 1) still going (step 1) still going (step 1) still going (step 1) still going (step 1) still going (step 1) still going (step 1) still going (step 1) still going (step 1) still going (step 1) still going (step 1) still going (step 1) ","tool_calls":[{"arguments":{"text":"tick"},"name":"echo"}]},"prev":"9f2b12e6cea41d14dbd01bfa202219d2275bfe726637149d2552875f88139e38","seq":1}
+{"hash":"8d9127924a7da47463acf8e3e2af264d9346824eca0547195f86f8f9d9298ff7","kind":"tool","payload":{"error":null,"name":"echo","ok":true,"value":"tick"},"prev":"b9809a74113fe58c8ebd7fb896025c5a5b9cdd3834d7ecff77ea2efba6d3ba3e","seq":2}
+{"hash":"ad178c6060058adb2ce55126b628886d196a86729f374d89083c2c37c6f52e6a","kind":"assistant","payload":{"text":"still going (step 2) still going (step 2) still going (step 2) still going (step 2) still going (step 2) still going (step 2) still going (step 2) still going (step 2) still going (step 2) still going (step 2) still going (step 2) still going (step 2) ","tool_calls":[{"arguments":{"text":"tick"},"name":"echo"}]},"prev":"8d9127924a7da47463acf8e3e2af264d9346824eca0547195f86f8f9d9298ff7","seq":3}
+{"hash":"a538a65f72d12b8a9d723996b7dff0ffe8a74a621da3a89bffaf1f29de1a8973","kind":"tool","payload":{"error":null,"name":"echo","ok":true,"value":"tick"},"prev":"ad178c6060058adb2ce55126b628886d196a86729f374d89083c2c37c6f52e6a","seq":4}
+{"hash":"6963d68e4cc1182d47d115ecce90ee9fd809d97cb7121b80073b0db6f2641253","kind":"assistant","payload":{"text":"still going (step 3) still going (step 3) still going (step 3) still going (step 3) still going (step 3) still going (step 3) still going (step 3) still going (step 3) still going (step 3) still going (step 3) still going (step 3) still going (step 3) ","tool_calls":[{"arguments":{"text":"tick"},"name":"echo"}]},"prev":"a538a65f72d12b8a9d723996b7dff0ffe8a74a621da3a89bffaf1f29de1a8973","seq":5}
+{"hash":"9cd3e5fe7c6630157085ae34750b16fa145d96727ff1cf3a59c035bfce3a022a","kind":"tool","payload":{"error":null,"name":"echo","ok":true,"value":"tick"},"prev":"6963d68e4cc1182d47d115ecce90ee9fd809d97cb7121b80073b0db6f2641253","seq":6}
+{"hash":"2f3e40832cc8d2eeb3cb45e7703e69395865419e21fe01e3bb60a68047fbdd3a","kind":"assistant","payload":{"text":"still going (step 4) still going (step 4) still going (step 4) still going (step 4) still going (step 4) still going (step 4) still going (step 4) still going (step 4) still going (step 4) still going (step 4) still going (step 4) still going (step 4) ","tool_calls":[{"arguments":{"text":"tick"},"name":"echo"}]},"prev":"9cd3e5fe7c6630157085ae34750b16fa145d96727ff1cf3a59c035bfce3a022a","seq":7}
+{"hash":"eeae79182ffb73b527092b95d57b3ac162cc70969ea2afb2e613b281636d50ec","kind":"final","payload":{"model_error":null,"steps":4,"stop_reason":"token_budget","tokens":252},"prev":"2f3e40832cc8d2eeb3cb45e7703e69395865419e21fe01e3bb60a68047fbdd3a","seq":8}

wizard-work/null/agentloop/demo-recover.jsonl

@@ -0,0 +1,11 @@
+{"hash":"d23b04b27ffcbd916ebcf850a097cfcd2132497171511d309ffaacd01d7c3150","kind":"user","payload":{"text":"Add 20 and 22, tolerating my mistakes."},"prev":"0000000000000000000000000000000000000000000000000000000000000000","seq":0}
+{"hash":"d27a02a27764ea253597c8c68c4e86750e3b6e0b592f60eefb65e70466e990c6","kind":"assistant","payload":{"text":"First try a name I made up.","tool_calls":[{"arguments":{"a":1},"name":"multiply"}]},"prev":"d23b04b27ffcbd916ebcf850a097cfcd2132497171511d309ffaacd01d7c3150","seq":1}
+{"hash":"1d030d3a1120706c6e2b11783feda547ce2cfb8d50ea90175e7836139d0481f5","kind":"tool","payload":{"error":"no tool named 'multiply' (have: add, echo, fail, greet)","name":"multiply","ok":false,"value":null},"prev":"d27a02a27764ea253597c8c68c4e86750e3b6e0b592f60eefb65e70466e990c6","seq":2}
+{"hash":"62bc277459dec0852538b75ebfceafabff7671d957ded874adb350ee20846fbf","kind":"assistant","payload":{"text":"Now with the wrong argument type.","tool_calls":[{"arguments":{"a":"2","b":3},"name":"add"}]},"prev":"1d030d3a1120706c6e2b11783feda547ce2cfb8d50ea90175e7836139d0481f5","seq":3}
+{"hash":"c18a871a8e6ee7c0d8aff95e7dd8c9aaacb3ab9d533b8715e73a8a4f19a1c5ff","kind":"tool","payload":{"error":"invalid arguments: $.add.a: expected type integer, got str","name":"add","ok":false,"value":null},"prev":"62bc277459dec0852538b75ebfceafabff7671d957ded874adb350ee20846fbf","seq":4}
+{"hash":"49fc20d82513ec36e60c2b15eb7e8476d13899ea6de4019245a10799c362603c","kind":"assistant","payload":{"text":"Now one that raises.","tool_calls":[{"arguments":{},"name":"fail"}]},"prev":"c18a871a8e6ee7c0d8aff95e7dd8c9aaacb3ab9d533b8715e73a8a4f19a1c5ff","seq":5}
+{"hash":"c173467c4892c61190cd92bfe3751f49d107c30fca2041a358eb86259a85b917","kind":"tool","payload":{"error":"tool 'fail' raised RuntimeError: the tool is broken on purpose","name":"fail","ok":false,"value":null},"prev":"49fc20d82513ec36e60c2b15eb7e8476d13899ea6de4019245a10799c362603c","seq":6}
+{"hash":"ac2001b3c4a39433bfa80fd767eb5b33f494450e556f0816c16a4d78d75ba7de","kind":"assistant","payload":{"text":"Now correctly.","tool_calls":[{"arguments":{"a":20,"b":22},"name":"add"}]},"prev":"c173467c4892c61190cd92bfe3751f49d107c30fca2041a358eb86259a85b917","seq":7}
+{"hash":"9a7693f9c98cf8661c1152417110f310d95e8f4e13636a8f46fe1ed68a76e100","kind":"tool","payload":{"error":null,"name":"add","ok":true,"value":{"sum":42}},"prev":"ac2001b3c4a39433bfa80fd767eb5b33f494450e556f0816c16a4d78d75ba7de","seq":8}
+{"hash":"c987f959f3890b98e242846c442382832b299668188c300d577cdb89fb6498c2","kind":"assistant","payload":{"text":"20 + 22 = 42. DONE","tool_calls":[]},"prev":"9a7693f9c98cf8661c1152417110f310d95e8f4e13636a8f46fe1ed68a76e100","seq":9}
+{"hash":"bd7a28422b47ae5a17a27d8688e790e2decb0edd75149f0d74abd70863788d82","kind":"final","payload":{"model_error":null,"steps":5,"stop_reason":"sentinel","tokens":26},"prev":"c987f959f3890b98e242846c442382832b299668188c300d577cdb89fb6498c2","seq":10}

wizard-work/null/agentloop/demo.py

@@ -0,0 +1,159 @@
+"""End-to-end demo: three scripted models over one toolset.
+
+These are not the test suite -- tests/ is.  They are the readable proof that the
+engine does something: a model that answers, a model that misuses its tools and
+recovers, and a model that never stops and is cut off by the budget.
+
+    python3 demo.py            # runs all three, writes demo-*.jsonl, prints a summary
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(HERE, ".."))
+
+from agentloop import AgentLoop, ModelReply, StopReason, Tool, ToolCall, ToolRegistry  # noqa: E402
+from agentloop.transcript import verify_file  # noqa: E402
+
+
+def make_registry():
+    def add(a, b):
+        return {"sum": a + b}
+
+    def greet(name, excited=False):
+        return "hello %s%s" % (name, "!" if excited else ".")
+
+    def fail():
+        raise RuntimeError("the tool is broken on purpose")
+
+    def echo(text):
+        return text
+
+    return ToolRegistry(
+        [
+            Tool("add", "Add two integers.", {
+                "type": "object",
+                "properties": {
+                    "a": {"type": "integer"},
+                    "b": {"type": "integer"},
+                },
+                "required": ["a", "b"],
+                "additionalProperties": False,
+            }, add),
+            Tool("greet", "Greet someone by name.", {
+                "type": "object",
+                "properties": {
+                    "name": {"type": "string", "minLength": 1},
+                    "excited": {"type": "boolean"},
+                },
+                "required": ["name"],
+            }, greet),
+            Tool("fail", "Always raises, to show a tool error in the transcript.", {
+                "type": "object", "properties": {}, "additionalProperties": False,
+            }, fail),
+            Tool("echo", "Return the text unchanged.", {
+                "type": "object",
+                "properties": {"text": {"type": "string"}},
+                "required": ["text"],
+            }, echo),
+        ]
+    )
+
+
+# -- the three models -----------------------------------------------------
+
+
+def answering_model(messages, manifest):
+    """Adds 2 and 3, then answers without asking for anything else."""
+    if not any(m.get("name") == "add" for m in messages):
+        return ModelReply("Let me add those.", [ToolCall("add", {"a": 2, "b": 3})])
+    return ModelReply("2 + 3 = 5.")
+
+
+def recovering_model(messages, manifest):
+    """First turn: a tool that does not exist, then a bad argument, then a
+    crashing tool, then a valid call.  Every failure is reported to the model as
+    a tool message and the loop continues."""
+    n = sum(1 for m in messages if m["role"] == "tool")
+    if n == 0:
+        return ModelReply("First try a name I made up.", [ToolCall("multiply", {"a": 1})])
+    if n == 1:
+        return ModelReply("Now with the wrong argument type.", [ToolCall("add", {"a": "2", "b": 3})])
+    if n == 2:
+        return ModelReply("Now one that raises.", [ToolCall("fail", {})])
+    if n == 3:
+        return ModelReply("Now correctly.", [ToolCall("add", {"a": 20, "b": 22})])
+    return ModelReply("20 + 22 = 42. DONE", [])
+
+
+def runaway_model(messages, manifest):
+    """Never stops on its own.  Only the budget stops it."""
+    i = sum(1 for m in messages if m["role"] == "assistant") + 1
+    return ModelReply("still going (step %d) " % i * 12, [ToolCall("echo", {"text": "tick"})])
+
+
+# -- runner ---------------------------------------------------------------
+
+
+def main():
+    out_dir = HERE
+    runs = []
+
+    loop = AgentLoop(make_registry(), max_steps=6)
+    result = loop.run("What is 2 + 3?", answering_model)
+    path = os.path.join(out_dir, "demo-answer.jsonl")
+    result.transcript.save(path)
+    runs.append(("answering", result, path))
+
+    loop = AgentLoop(make_registry(), max_steps=6, stop_sentinel="DONE")
+    result = loop.run("Add 20 and 22, tolerating my mistakes.", recovering_model)
+    path = os.path.join(out_dir, "demo-recover.jsonl")
+    result.transcript.save(path)
+    runs.append(("recovering", result, path))
+
+    loop = AgentLoop(make_registry(), max_steps=4, max_tokens=200)
+    result = loop.run("Keep going forever.", runaway_model)
+    path = os.path.join(out_dir, "demo-budget.jsonl")
+    result.transcript.save(path)
+    runs.append(("runaway", result, path))
+
+    print("demo runs")
+    for name, result, path in runs:
+        ok, violations, count = verify_file(path)
+        print(
+            "  %-11s stop=%-13s steps=%d tokens=%-4d records=%d chain=%s"
+            % (name, result.stop_reason, result.steps, result.tokens, count, "OK" if ok else violations)
+        )
+
+    errors = sum(1 for r in runs[1][1].tool_results() if not r.ok)
+    print("  recovering  tool failures surfaced to the model: %d" % errors)
+    print("  runaway     stopped by budget, not by the model: %s"
+          % (runs[2][1].stop_reason in (StopReason.STEP_BUDGET, StopReason.TOKEN_BUDGET)))
+    print("  transcripts written next to demo.py")
+
+    # Exit 0 only if every claim above actually held.
+    if not all(verify_file(p)[0] for _, _, p in runs):
+        print("FAIL: a transcript failed chain verification")
+        return 1
+    if runs[0][1].stop_reason != StopReason.NO_TOOL_CALLS:
+        print("FAIL: the answering model did not stop on no_tool_calls")
+        return 1
+    if runs[1][1].stop_reason != StopReason.SENTINEL:
+        print("FAIL: the recovering model did not stop on the sentinel")
+        return 1
+    if errors != 3:
+        print("FAIL: expected 3 surfaced tool failures, saw %d" % errors)
+        return 1
+    if runs[2][1].stop_reason not in (StopReason.STEP_BUDGET, StopReason.TOKEN_BUDGET):
+        print("FAIL: the runaway model was not cut off by a budget")
+        return 1
+    print("ALL DEMO CLAIMS HELD")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

wizard-work/null/agentloop/tests/__init__.py

@@ -0,0 +1 @@
+"""Test package for agentloop."""

wizard-work/null/agentloop/tests/test_agentloop.py

@@ -0,0 +1,605 @@
+"""Test suite for agentloop.
+
+Run it directly (``python3 -m unittest discover -s tests`` from the artifact
+root) or through check.py, which also runs an independent tamper probe.
+
+The tamper tests are the teeth: a verifier that cannot be shown to fail is not
+evidence.  Every mutation below is one a real editor or a buggy writer could
+make, and each is asserted to be caught AND to be caught at the right record.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.dirname(HERE)
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)
+
+from agentloop import (  # noqa: E402
+    AgentLoop,
+    ModelReply,
+    StopReason,
+    Tool,
+    ToolCall,
+    ToolRegistry,
+    ToolResult,
+    Transcript,
+    TranscriptError,
+    ValidationError,
+    canonical_json,
+    estimate_tokens,
+    validate,
+    verify_file,
+    verify_records,
+)
+from agentloop.transcript import GENESIS, KIND_ASSISTANT, KIND_FINAL, KIND_TOOL, KIND_USER  # noqa: E402
+
+
+def add(a, b):
+    return a + b
+
+
+def make_registry():
+    return ToolRegistry(
+        [
+            Tool("add", "Add two integers.", {
+                "type": "object",
+                "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
+                "required": ["a", "b"],
+                "additionalProperties": False,
+            }, add),
+            Tool("shout", "Uppercase a string.", {
+                "type": "object",
+                "properties": {"text": {"type": "string", "minLength": 1}},
+                "required": ["text"],
+            }, lambda text: text.upper()),
+        ]
+    )
+
+
+def write_tmp(records, raw_lines=None):
+    fd, path = tempfile.mkstemp(suffix=".jsonl")
+    os.close(fd)
+    with open(path, "wb") as fh:
+        if raw_lines is not None:
+            fh.write(("\n".join(raw_lines) + "\n").encode("utf-8"))
+        else:
+            fh.write("".join(canonical_json(r) + "\n" for r in records).encode("utf-8"))
+    return path
+
+
+# --------------------------------------------------------------------------
+# transcript: the chain
+# --------------------------------------------------------------------------
+
+
+class TestTranscriptChain(unittest.TestCase):
+    def test_chain_links_each_record_to_its_predecessor(self):
+        tr = Transcript()
+        tr.append(KIND_USER, {"text": "hi"})
+        tr.append(KIND_ASSISTANT, {"text": "yo"})
+        tr.append(KIND_TOOL, {"name": "add", "ok": True, "value": 5})
+        self.assertEqual([r["seq"] for r in tr.records], [0, 1, 2])
+        self.assertEqual(tr.records[0]["prev"], GENESIS)
+        for earlier, later in zip(tr.records, tr.records[1:]):
+            self.assertEqual(later["prev"], earlier["hash"])
+
+    def test_genesis_hash_is_the_all_zero_constant(self):
+        tr = Transcript()
+        tr.append(KIND_USER, {"text": "x"})
+        self.assertEqual(tr.records[0]["prev"], "0" * 64)
+
+    def test_bytes_are_deterministic_and_reloadable(self):
+        def build():
+            tr = Transcript()
+            tr.append(KIND_USER, {"text": "same"})
+            tr.append(KIND_ASSISTANT, {"text": "same", "tool_calls": []})
+            return tr
+
+        self.assertEqual(build().bytes(), build().bytes())
+        path = write_tmp(build().records)
+        try:
+            self.assertEqual(Transcript(path).bytes(), build().bytes())
+        finally:
+            os.unlink(path)
+
+    def test_save_then_load_round_trips(self):
+        tr = Transcript()
+        tr.append(KIND_USER, {"text": "round trip"})
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "t.jsonl")
+            tr.save(path)
+            self.assertEqual(Transcript(path).records, tr.records)
+            self.assertEqual(verify_file(path)[:2], (True, []))
+
+    def test_rejects_unknown_kind_and_bad_payload(self):
+        tr = Transcript()
+        with self.assertRaises(TranscriptError):
+            tr.append("system", {"text": "nope"})
+        with self.assertRaises(TranscriptError):
+            tr.append(KIND_USER, "not a dict")
+        with self.assertRaises(TranscriptError):
+            tr.append(KIND_USER, {"bad": object()})
+
+    def test_load_raises_on_bad_json_with_the_line_number(self):
+        path = write_tmp(None, raw_lines=['{"seq": 0}', "{not json"])
+        try:
+            with self.assertRaises(TranscriptError) as ctx:
+                Transcript(path)
+            self.assertIn("line 2", str(ctx.exception))
+        finally:
+            os.unlink(path)
+
+
+class TestTamperDetection(unittest.TestCase):
+    """The teeth.  Each mutation must be caught, and at the named record."""
+
+    def setUp(self):
+        tr = Transcript()
+        tr.append(KIND_USER, {"text": "add 2 and 3"})
+        tr.append(KIND_ASSISTANT, {"text": "calling add", "tool_calls": []})
+        tr.append(KIND_TOOL, {"name": "add", "ok": True, "value": 5})
+        tr.append(KIND_FINAL, {"stop_reason": "no_tool_calls", "steps": 1, "tokens": 4})
+        self.records = tr.records
+
+    def test_clean_chain_verifies(self):
+        ok, violations = verify_records(self.records)
+        self.assertTrue(ok, violations)
+        self.assertEqual(violations, [])
+
+    def test_payload_rewrite_is_caught_at_that_record(self):
+        tampered = json.loads(json.dumps(self.records))
+        tampered[2]["payload"]["value"] = 999
+        ok, violations = verify_records(tampered)
+        self.assertFalse(ok)
+        self.assertEqual(violations[0]["seq"], 2)
+        self.assertEqual(violations[0]["reason"], "hash_mismatch")
+
+    def test_single_character_edit_in_payload_is_caught(self):
+        tampered = json.loads(json.dumps(self.records))
+        tampered[0]["payload"]["text"] = "add 2 and 4"
+        ok, violations = verify_records(tampered)
+        self.assertFalse(ok)
+        self.assertEqual(violations[0]["seq"], 0)
+        self.assertEqual(violations[0]["reason"], "hash_mismatch")
+
+    def test_dropped_middle_record_is_caught_as_a_seq_gap(self):
+        tampered = json.loads(json.dumps(self.records))
+        del tampered[1]
+        ok, violations = verify_records(tampered)
+        self.assertFalse(ok)
+        self.assertEqual(violations[0]["seq"], 1)
+        self.assertEqual(violations[0]["reason"], "seq_gap")
+
+    def test_swapped_records_break_the_chain_at_the_first_swapped_index(self):
+        """A raw swap moves record 2's seq into slot 1, so the seq check fires
+        first.  The chain breaks and names index 1 -- which is the fault."""
+        tampered = json.loads(json.dumps(self.records))
+        tampered[1], tampered[2] = tampered[2], tampered[1]
+        ok, violations = verify_records(tampered)
+        self.assertFalse(ok)
+        self.assertEqual(violations[0]["seq"], 1)
+        self.assertEqual(violations[0]["reason"], "seq_gap")
+
+    def test_reorder_plus_renumber_is_caught_as_a_prev_mismatch(self):
+        """The harder attack: swap two records AND renumber them so the seq
+        check passes.  Only the prev link can catch this, and it does."""
+        tampered = json.loads(json.dumps(self.records))
+        tampered[1], tampered[2] = tampered[2], tampered[1]
+        tampered[1]["seq"], tampered[2]["seq"] = 1, 2
+        ok, violations = verify_records(tampered)
+        self.assertFalse(ok)
+        self.assertEqual(violations[0]["seq"], 1)
+        self.assertEqual(violations[0]["reason"], "prev_mismatch")
+
+    def test_forged_record_with_a_recomputed_hash_still_breaks_the_chain(self):
+        """An attacker who rewrites a payload AND recomputes that record's hash
+        is still caught, because the NEXT record still points at the old hash."""
+        from agentloop.transcript import _record_hash
+
+        tampered = json.loads(json.dumps(self.records))
+        tampered[2]["payload"]["value"] = 999
+        tampered[2]["hash"] = _record_hash(
+            tampered[2]["seq"], tampered[2]["kind"], tampered[2]["payload"], tampered[2]["prev"]
+        )
+        ok, violations = verify_records(tampered)
+        self.assertFalse(ok)
+        self.assertEqual(violations[0]["seq"], 3)
+        self.assertEqual(violations[0]["reason"], "prev_mismatch")
+
+    def test_hash_field_replaced_by_a_random_hex_is_caught(self):
+        tampered = json.loads(json.dumps(self.records))
+        tampered[1]["hash"] = "f" * 64
+        ok, violations = verify_records(tampered)
+        self.assertFalse(ok)
+        self.assertEqual(violations[0]["seq"], 1)
+        self.assertEqual(violations[0]["reason"], "hash_mismatch")
+
+    def test_missing_field_is_reported_by_name(self):
+        tampered = json.loads(json.dumps(self.records))
+        del tampered[0]["prev"]
+        ok, violations = verify_records(tampered)
+        self.assertFalse(ok)
+        self.assertEqual(violations[0]["reason"], "missing_fields")
+        self.assertEqual(violations[0]["detail"], ["prev"])
+
+    def test_non_object_record_is_caught(self):
+        tampered = json.loads(json.dumps(self.records))
+        tampered[1] = "i am not a record"
+        ok, violations = verify_records(tampered)
+        self.assertFalse(ok)
+        self.assertEqual(violations[0]["reason"], "not_an_object")
+
+    def test_bad_json_line_is_reported_with_its_line_number(self):
+        path = write_tmp(None, raw_lines=[canonical_json(self.records[0]), "{oops"])
+        try:
+            ok, violations, count = verify_file(path)
+            self.assertFalse(ok)
+            self.assertEqual(violations[0]["reason"], "bad_json")
+            self.assertEqual(violations[0]["line"], 2)
+            self.assertEqual(count, 1)
+        finally:
+            os.unlink(path)
+
+    def test_truncation_is_detected_only_against_an_external_count(self):
+        """The honest limit: the chain alone cannot see a missing tail."""
+        full = write_tmp(self.records)
+        cut = write_tmp(self.records[:2])
+        try:
+            ok_full, _, n_full = verify_file(full)
+            ok_cut, _, n_cut = verify_file(cut)
+            self.assertTrue(ok_full)
+            self.assertTrue(ok_cut)  # chain intact -- this is the documented limit
+            self.assertLess(n_cut, n_full)
+        finally:
+            os.unlink(full)
+            os.unlink(cut)
+
+    def test_verify_file_reports_an_unreadable_path(self):
+        ok, violations, count = verify_file("/nonexistent/nope.jsonl")
+        self.assertFalse(ok)
+        self.assertEqual(violations[0]["reason"], "unreadable")
+        self.assertEqual(count, 0)
+
+
+# --------------------------------------------------------------------------
+# tool arguments
+# --------------------------------------------------------------------------
+
+
+class TestArgumentValidation(unittest.TestCase):
+    def test_valid_arguments_pass(self):
+        validate({"a": 2, "b": 3}, {
+            "type": "object",
+            "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
+            "required": ["a", "b"],
+        })
+
+    def test_missing_required_property(self):
+        with self.assertRaises(ValidationError) as ctx:
+            validate({"a": 2}, {"type": "object", "properties": {"a": {"type": "integer"}},
+                                 "required": ["a", "b"]})
+        self.assertIn("b", str(ctx.exception))
+
+    def test_wrong_type(self):
+        with self.assertRaises(ValidationError):
+            validate({"a": "2", "b": 3}, {
+                "type": "object",
+                "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
+            })
+
+    def test_bool_is_not_an_integer(self):
+        """bool subclasses int; without the explicit guard True == 1."""
+        with self.assertRaises(ValidationError):
+            validate({"a": True, "b": 3}, {
+                "type": "object",
+                "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
+            })
+
+    def test_float_is_a_number_but_not_an_integer(self):
+        validate(1.5, {"type": "number"})
+        with self.assertRaises(ValidationError):
+            validate(1.5, {"type": "integer"})
+
+    def test_additional_properties_false_rejects_extras(self):
+        with self.assertRaises(ValidationError) as ctx:
+            validate({"a": 1, "b": 2, "c": 3}, {
+                "type": "object",
+                "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
+                "additionalProperties": False,
+            })
+        self.assertIn("c", str(ctx.exception))
+
+    def test_enum(self):
+        validate("red", {"enum": ["red", "green"]})
+        with self.assertRaises(ValidationError):
+            validate("blue", {"enum": ["red", "green"]})
+
+    def test_array_items_are_validated_recursively(self):
+        schema = {"type": "array", "items": {"type": "integer"}, "minItems": 1}
+        validate([1, 2, 3], schema)
+        with self.assertRaises(ValidationError) as ctx:
+            validate([1, "two"], schema)
+        self.assertIn("[1]", str(ctx.exception))
+
+    def test_min_items_and_max_items(self):
+        with self.assertRaises(ValidationError):
+            validate([], {"type": "array", "minItems": 1})
+        with self.assertRaises(ValidationError):
+            validate([1, 2], {"type": "array", "maxItems": 1})
+
+    def test_string_bounds_and_pattern(self):
+        with self.assertRaises(ValidationError):
+            validate("", {"type": "string", "minLength": 1})
+        with self.assertRaises(ValidationError):
+            validate("toolong", {"type": "string", "maxLength": 3})
+        with self.assertRaises(ValidationError):
+            validate("abc", {"type": "string", "pattern": "^[0-9]+$"})
+        validate("123", {"type": "string", "pattern": "^[0-9]+$"})
+
+    def test_numeric_bounds(self):
+        validate(5, {"type": "integer", "minimum": 5, "maximum": 5})
+        with self.assertRaises(ValidationError):
+            validate(4, {"type": "integer", "minimum": 5})
+        with self.assertRaises(ValidationError):
+            validate(6, {"type": "integer", "maximum": 5})
+
+    def test_nested_object_properties_are_checked(self):
+        schema = {
+            "type": "object",
+            "properties": {"outer": {"type": "object", "properties": {"n": {"type": "integer"}},
+                                       "required": ["n"]}},
+        }
+        validate({"outer": {"n": 1}}, schema)
+        with self.assertRaises(ValidationError) as ctx:
+            validate({"outer": {"n": "x"}}, schema)
+        self.assertIn("outer.n", str(ctx.exception))
+
+    def test_unknown_type_in_schema_is_a_loud_error(self):
+        with self.assertRaises(ValidationError):
+            validate(1, {"type": "flobble"})
+
+    def test_tool_call_validates_before_running(self):
+        tool = Tool("add", "add", {
+            "type": "object",
+            "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
+            "required": ["a", "b"],
+        }, add)
+        self.assertEqual(tool({"a": 2, "b": 3}), 5)
+        with self.assertRaises(ValidationError):
+            tool({"a": 2})
+
+    def test_tool_wraps_an_unexpected_exception_in_toolerror(self):
+        def boom():
+            raise RuntimeError("kaboom")
+
+        tool = Tool("boom", "booms", {"type": "object", "properties": {}}, boom)
+        with self.assertRaises(Exception) as ctx:
+            tool({})
+        self.assertIn("kaboom", str(ctx.exception))
+
+    def test_registry_rejects_duplicate_names(self):
+        reg = ToolRegistry([Tool("add", "", {"type": "object", "properties": {}}, add)])
+        with self.assertRaises(ValueError):
+            reg.add(Tool("add", "", {"type": "object", "properties": {}}, add))
+
+    def test_registry_manifest_is_sorted_and_complete(self):
+        reg = make_registry()
+        names = [m["name"] for m in reg.manifest()]
+        self.assertEqual(names, ["add", "shout"])
+        for m in reg.manifest():
+            self.assertEqual(sorted(m), ["description", "name", "parameters"])
+
+
+# --------------------------------------------------------------------------
+# the loop
+# --------------------------------------------------------------------------
+
+
+class TestLoop(unittest.TestCase):
+    def test_completes_when_the_model_stops_asking_for_tools(self):
+        def model(messages, manifest):
+            return ModelReply("done", [])
+
+        result = AgentLoop(make_registry()).run("hi", model)
+        self.assertEqual(result.stop_reason, StopReason.NO_TOOL_CALLS)
+        self.assertTrue(result.ok)
+        self.assertEqual(result.steps, 1)
+
+    def test_executes_tool_calls_in_order_and_records_them(self):
+        calls = [ToolCall("add", {"a": 2, "b": 3}), ToolCall("shout", {"text": "hi"})]
+        seen = []
+
+        def model(messages, manifest):
+            if not seen:
+                seen.append(1)
+                return ModelReply("working", calls)
+            return ModelReply("2+3=5 and HI", [])
+
+        result = AgentLoop(make_registry()).run("hi", model)
+        results = result.tool_results()
+        self.assertEqual([r.name for r in results], ["add", "shout"])
+        self.assertEqual([r.value for r in results], [5, "HI"])
+        self.assertTrue(all(r.ok for r in results))
+
+    def test_unknown_tool_is_surfaced_as_an_error_and_the_loop_continues(self):
+        def model(messages, manifest):
+            if len(messages) == 1:
+                return ModelReply("try", [ToolCall("nope", {})])
+            return ModelReply("recovered", [])
+
+        result = AgentLoop(make_registry()).run("hi", model)
+        self.assertEqual(result.stop_reason, StopReason.NO_TOOL_CALLS)
+        (only,) = result.tool_results()
+        self.assertFalse(only.ok)
+        self.assertIn("no tool named 'nope'", only.error)
+
+    def test_invalid_arguments_are_surfaced_as_an_error(self):
+        def model(messages, manifest):
+            if len(messages) == 1:
+                return ModelReply("try", [ToolCall("add", {"a": 1})])
+            return ModelReply("ok", [])
+
+        result = AgentLoop(make_registry()).run("hi", model)
+        (only,) = result.tool_results()
+        self.assertFalse(only.ok)
+        self.assertIn("invalid arguments", only.error)
+        self.assertIn("b", only.error)
+
+    def test_a_tool_that_raises_does_not_kill_the_loop(self):
+        reg = ToolRegistry([Tool("boom", "booms", {"type": "object", "properties": {}},
+                                 lambda: (_ for _ in ()).throw(RuntimeError("kaboom")))])
+
+        def model(messages, manifest):
+            if len(messages) == 1:
+                return ModelReply("go", [ToolCall("boom", {})])
+            return ModelReply("survived", [])
+
+        result = AgentLoop(reg).run("hi", model)
+        self.assertEqual(result.stop_reason, StopReason.NO_TOOL_CALLS)
+        self.assertFalse(result.tool_results()[0].ok)
+
+    def test_non_serialisable_return_value_is_refused(self):
+        reg = ToolRegistry([Tool("obj", "returns an object", {"type": "object", "properties": {}},
+                                 lambda: object())])
+
+        def model(messages, manifest):
+            if len(messages) == 1:
+                return ModelReply("go", [ToolCall("obj", {})])
+            return ModelReply("done", [])
+
+        result = AgentLoop(reg).run("hi", model)
+        (only,) = result.tool_results()
+        self.assertFalse(only.ok)
+        self.assertIn("non-serialisable", only.error)
+        # and the transcript is still writable
+        self.assertTrue(result.transcript.bytes())
+
+    def test_step_budget_caps_the_loop_and_matches_the_record_count(self):
+        def runaway(messages, manifest):
+            return ModelReply("more", [ToolCall("shout", {"text": "x"})])
+
+        result = AgentLoop(make_registry(), max_steps=4).run("hi", runaway)
+        self.assertEqual(result.stop_reason, StopReason.STEP_BUDGET)
+        self.assertEqual(result.steps, 4)
+        self.assertFalse(result.ok)
+        assistant_records = [r for r in result.transcript.records if r["kind"] == KIND_ASSISTANT]
+        self.assertEqual(len(assistant_records), 4)
+        self.assertEqual(result.steps, len(assistant_records))
+
+    def test_token_budget_stops_the_loop_before_running_tools(self):
+        def chatty(messages, manifest):
+            return ModelReply("x" * 400, [ToolCall("shout", {"text": "x"})])
+
+        result = AgentLoop(make_registry(), max_steps=50, max_tokens=20).run("hi", chatty)
+        self.assertEqual(result.stop_reason, StopReason.TOKEN_BUDGET)
+        self.assertEqual(result.tool_results(), [])
+        self.assertGreater(result.tokens, 20)
+
+    def test_sentinel_stops_the_loop(self):
+        def model(messages, manifest):
+            if len(messages) == 1:
+                return ModelReply("first", [ToolCall("add", {"a": 1, "b": 1})])
+            return ModelReply("the answer is 2 -- DONE", [])
+
+        result = AgentLoop(make_registry(), stop_sentinel="DONE").run("hi", model)
+        self.assertEqual(result.stop_reason, StopReason.SENTINEL)
+        self.assertTrue(result.ok)
+
+    def test_sentinel_wins_over_a_tool_call_in_the_same_turn(self):
+        def model(messages, manifest):
+            if len(messages) == 1:
+                return ModelReply("first", [ToolCall("add", {"a": 1, "b": 1})])
+            return ModelReply("done DONE", [ToolCall("add", {"a": 9, "b": 9})])
+
+        result = AgentLoop(make_registry(), stop_sentinel="DONE").run("hi", model)
+        self.assertEqual(result.stop_reason, StopReason.SENTINEL)
+        self.assertEqual(len(result.tool_results()), 1)
+
+    def test_model_exception_becomes_a_model_error_stop(self):
+        def model(messages, manifest):
+            raise RuntimeError("the model is down")
+
+        result = AgentLoop(make_registry()).run("hi", model)
+        self.assertEqual(result.stop_reason, StopReason.MODEL_ERROR)
+        self.assertIn("the model is down", result.model_error)
+        self.assertFalse(result.ok)
+
+    def test_model_returning_the_wrong_type_becomes_a_model_error(self):
+        result = AgentLoop(make_registry()).run("hi", lambda m, f: "not a reply")
+        self.assertEqual(result.stop_reason, StopReason.MODEL_ERROR)
+
+    def test_a_model_that_raises_mid_run_still_leaves_a_valid_transcript(self):
+        state = {"n": 0}
+
+        def model(messages, manifest):
+            state["n"] += 1
+            if state["n"] == 1:
+                return ModelReply("go", [ToolCall("add", {"a": 1, "b": 2})])
+            raise RuntimeError("died on turn 2")
+
+        result = AgentLoop(make_registry()).run("hi", model)
+        self.assertEqual(result.stop_reason, StopReason.MODEL_ERROR)
+        self.assertTrue(verify_records(result.transcript.records)[0])
+        self.assertEqual(result.transcript.records[-1]["kind"], KIND_FINAL)
+
+    def test_transcript_record_order_is_user_assistant_tool_final(self):
+        def model(messages, manifest):
+            if len(messages) == 1:
+                return ModelReply("go", [ToolCall("add", {"a": 1, "b": 1})])
+            return ModelReply("done", [])
+
+        result = AgentLoop(make_registry()).run("hi", model)
+        kinds = [r["kind"] for r in result.transcript.records]
+        self.assertEqual(kinds, [KIND_USER, KIND_ASSISTANT, KIND_TOOL, KIND_ASSISTANT, KIND_FINAL])
+
+    def test_identical_runs_produce_byte_identical_transcripts(self):
+        def model(messages, manifest):
+            if len(messages) == 1:
+                return ModelReply("go", [ToolCall("add", {"a": 40, "b": 2})])
+            return ModelReply("42", [])
+
+        a = AgentLoop(make_registry()).run("hi", model)
+        b = AgentLoop(make_registry()).run("hi", model)
+        self.assertEqual(a.transcript.bytes(), b.transcript.bytes())
+
+    def test_a_supplied_transcript_is_appended_to_not_replaced(self):
+        existing = Transcript()
+        existing.append(KIND_USER, {"text": "an earlier session"})
+
+        result = AgentLoop(make_registry()).run("hi", lambda m, f: ModelReply("done", []),
+                                             transcript=existing)
+        # the supplied record, then user + assistant + final = 4
+        self.assertEqual(len(result.transcript), 4)
+        self.assertEqual(result.transcript.records[0]["payload"]["text"], "an earlier session")
+        self.assertEqual([r["kind"] for r in result.transcript.records][0], KIND_USER)
+        self.assertTrue(verify_records(result.transcript.records)[0])
+
+    def test_budgets_must_be_positive_integers(self):
+        with self.assertRaises(ValueError):
+            AgentLoop(make_registry(), max_steps=0)
+        with self.assertRaises(ValueError):
+            AgentLoop(make_registry(), max_steps=True)
+        with self.assertRaises(ValueError):
+            AgentLoop(make_registry(), max_tokens=0)
+
+    def test_registry_must_be_a_toolregistry(self):
+        with self.assertRaises(TypeError):
+            AgentLoop({"add": add})
+
+    def test_estimate_tokens_is_a_documented_floor(self):
+        self.assertEqual(estimate_tokens(""), 0)
+        self.assertEqual(estimate_tokens("ab"), 1)   # floor 1, not 0
+        self.assertEqual(estimate_tokens("abcd"), 1)
+        self.assertEqual(estimate_tokens("abcde"), 1)
+        self.assertEqual(estimate_tokens("a" * 40), 10)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)