initial: zubble agentic harness — extracted from llm_benchmark
Some checks are pending
release / release (push) Waiting to run

Package layout (zubble/), vendored benchlib.stream_chat, LocalEnv,
minimal  CLI, 38 pure-logic tests, and a tag-triggered
Forgejo Actions release pipeline that builds a single-file PyInstaller
executable.
This commit is contained in:
pch 2026-10-24 09:37:10 -04:00
commit b005fb1ef6
Signed by untrusted user who does not match committer: pch
GPG key ID: 8F0826ECF7302C63
19 changed files with 4177 additions and 0 deletions

View file

@ -0,0 +1,54 @@
name: release
# Cut a release ONLY when a v* tag is pushed. Regular pushes do nothing.
# The tag name becomes the release version and the binary name suffix.
on:
push:
tags:
- 'v*'
jobs:
release:
permissions:
contents: write
runs-on: codeberg-small
steps:
# Full history so make_release.py can build changelog since the last tag.
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Install package + dev deps
run: |
python -m pip install --upgrade pip
pip install -e .[dev]
- name: Run tests
run: pytest -q
- name: Build single-file executable
# --onefile bundles the interpreter: a true standalone binary.
# zubble+benchlib are pure stdlib, so the bundle is small and
# there is no fragile third-party dependency packing to go wrong.
run: pyinstaller --onefile --name zubble zubble/cli.py
- name: Smoke test the binary
# Validates the executable actually launches and parses args —
# never ship a build that can't even print --version.
run: |
./dist/zubble --version
./dist/zubble --help | head -1
- name: Create release and upload binary
env:
FORGEJO_TOKEN: ${{ secrets.GITHUB_TOKEN }}
FORGEJO_API: ${{ github.server_url }}/api/v1
REPO: ${{ github.repository }}
TAG: ${{ github.ref_name }}
run: python scripts/make_release.py dist/zubble

18
.gitignore vendored Normal file
View file

@ -0,0 +1,18 @@
# Python
__pycache__/
*.py[cod]
*.egg-info/
.eggs/
build/
dist/
.pytest_cache/
# Runtime data (live sqlite journals — never belong in version control)
*.sqlite3
*.sqlite3-shm
*.sqlite3-wal
# Editor
.idea/
.vscode/
*.swp

571
DESIGN.md Normal file
View file

@ -0,0 +1,571 @@
# zubble — A Minimal Agentic Harness
> A hackable, extensible agent loop that combines the simplicity of
> [mini-swe-agent](https://github.com/SWE-agent/mini-swe-agent) with the
> anti-loop discipline of [Ralph Wiggum](https://github.com/fstandhartinger/ralph-wiggum),
> the doom-loop detection of [ZeroStack](https://github.com/gi-dellav/zerostack),
> and hard-won lessons from FestBench's ScenarioBench harness.
---
## Design Principles
1. **Readable.** The core agent loop is ~200 lines of Python. A new
contributor should understand the entire control flow in one sitting.
No magic, no metaprogramming, no framework tax.
2. **Extensible.** Three pluggable interfaces — `Model`, `Environment`,
`Observer` — cover every extension point. New providers, sandboxes, or
grading hooks are subclasses, not config gymnastics.
3. **Sharable.** MIT licensed. No FestBench dependency. Runs standalone
against any OpenAI-compatible endpoint, hosted API, or local CLI.
Pip-installable. Single-file deployable.
4. **Pythonic.** Type-hinted, dataclass configs, stdlib-first. Pydantic
for config validation (same as mini-swe-agent). No Rust, no Go, no
Node — the language the research community already speaks.
5. **Honest.** Every failure mode is recorded, not hidden. Degenerate
loops are detected and logged, not silently retried. Token costs are
tracked per call, not estimated after the fact.
---
## Why "zubble"?
Named after the world's best dog. A good agentic harness is Man's best
friend — loyal, dependable, always ready to help you fix what's broken.
---
## Origin: What We Learned from FestBench
FestBench's `scenariobench.py` gave a model a broken Kubernetes cluster,
a wall-clock budget, and a free-form text protocol. At xhigh thinking
levels, the model generated 79K characters of degenerate conversation
simulation — emitting both `User:` and `Assistant:` turns in a single
response, 7,303 repetitions of the word "truncated" before timing out.
Five root causes were identified:
1. **No stop sequences** — the API call had no `stop=["User:"]`, so
generation continued through simulated conversation turns.
2. **Flat-text rendering** — the harness flattened the message history
into a single string with `User:`/`Assistant:` markers, then sent it
as one user message. The model saw the pattern and copied it.
3. **No in-stream repetition detection** — the livelock detector only
checked extracted commands between turns, not repeated text within
a single generation.
4. **Unbounded maxtok** — xhigh's 32K token budget at 41.5 tok/s = 770s,
longer than the 480s timeout. One degenerate turn ate the entire
scenario budget.
5. **No context reset** — when the model got stuck, the accumulated
context (including the 79K-char degenerate response) poisoned every
subsequent turn.
`zubble` is designed to make all five of these failure modes structurally
impossible.
---
## Architecture
```
stream_chat() ← shared HTTP client (unchanged, no ledger awareness)
↑
├── softbench.py → calls stream_chat once per task, scores output
├── infrabench.py → calls stream_chat once per task, validates output
├── prosebench.py → calls stream_chat once per task, scores output
├── psychbench.py → calls stream_chat once per task, scores output
└── scenariobench.py → calls agent_loop() which calls stream_chat N times
↑
agent_loop (zubble)
├── protocol.py ← JSON schema, system prompt, parse-retry
├── ledger.py ← plan state machine (short-term memory)
└── repetition.py ← in-stream repetition detection
```
The plan-state ledger is the model's short-term memory. It tracks not
just what was attempted, but the full causal chain: what the model
planned to do, which steps are done vs pending, where execution diverged
from the plan, and why previous approaches were abandoned.
### Protocol
The model communicates in structured JSON, not free-form text:
```json
{
"action": "plan" | "run" | "report" | "submit" | "ask",
"command": "<shell command, for action=run>",
"plan_step_id": <int, references the plan>,
"reasoning": "<brief explanation>",
"steps": ["<step 1>", "<step 2>", ...], // for action=plan only
"report": "<short submission summary>", // for action=report/submit only
"question": "<clarification>", // for action=ask only
"revised_plan": ["<new steps>"] // when abandoning current plan
}
```
Turn 1: model emits `action: "plan"` with an ordered list of steps.
Turns 2..N: model emits `action: "run"` with `plan_step_id` referencing
which step this action fulfills. The harness checks each action against
the plan state machine. If the action doesn't match the next pending
step, a divergence warning is injected before the command executes.
`submit` is the only verification path. A rejected submission tells the
model to keep working and submit again; the harness owns termination through
the deadline, turn cap, or doom-loop safeguards. The full post-mortem belongs
in the scenario report file, while the JSON `report` field is a short,
single-line summary.
JSON enforcement combines endpoint `response_format: json_schema`, strict
single-object parsing, and one targeted repair generation on malformed output.
After a failed repair, the scenario exits as `parse_error` with the parser
reason rather than spending its remaining budget in an invalid-JSON loop.
### Plan-state ledger
```
PLAN (captured at turn 1):
Step 1: Diagnose bad cert → DONE (found expired cert on node-b)
Step 2: Create new cert manifest → DONE (cert-fix.yaml created)
Step 3: Apply manifest → PENDING
Step 4: Verify cert renewed → PENDING
ABANDONED APPROACHES (with causal failure reasons):
- "restart kubelet" → failed because kubelet was already running;
cert expiry was root cause, not service state
CURRENT STATE:
Next expected step: Step 3 (apply manifest)
Last action: kubectl create -f cert-fix.yaml (step 2, succeeded)
Divergence flag: NONE
```
Injected into every prompt as "## SESSION MEMORY" so the model sees its
plan, progress, and abandoned approaches before choosing its next action.
### Repetition detection (Phase 2)
Rolling 10-gram CRC32 fingerprint over the last 200 generated tokens.
Graduated thresholds prevent false-positive kills of valid long-running
processes:
| Threshold | Action | Rationale |
|-----------|--------|-----------|
| 50 | Soft nudge injected into stream | Low-cost course correction |
| 500 | Parallel self-check (diagnostic) | First probe, no action taken |
| 1000 | "STOP! Restart with fresh context or you will be terminated." | Warning, NOT a kill |
| 1500 | Self-confirmation gate: model reviews progress, answers true/false. TRUE→kill, FALSE→continue | Confirmation gate prevents false kills |
| 3000 | Unconditional kill | Final safety net |
Self-check failure or timeout → continue, not kill. The cost of a false
kill (killing a valid process making real progress) is higher than the
cost of letting a stuck process run until the next threshold.
### Component contracts
**Model** — `query(messages) -> response`
Wraps any LLM endpoint. Handles streaming, token accounting, thinking-level
control, stop sequences. Subclasses: `OpenAIModel` (llama.cpp/vLLM/Ollama),
`LitellmModel` (hosted APIs), `CliModel` (claude-cli, gemini, copilot).
**Environment** — `execute(action) -> observation`
Runs the agent's commands in a sandbox. Subclasses: `LocalEnv`,
`DockerEnv`, `KubectlEnv`, `SandboxedEnv` (bubblewrap).
**Observer** — `check(messages, response) -> Optional[Violation]`
Inspects the conversation for degenerate patterns. Returns a `Violation`
enum: `DOOM_LOOP`, `TEXT_REPETITION`, `FORMAT_ERROR`, `CONTEXT_PRESSURE`.
The agent decides what to do based on the violation type.
**Journal** — `record(command, exit_code, output)`, `summary() -> str`
Disk-persisted record of what the agent tried, what worked, and what
didn't. Survives context resets. Fed to the model at the start of each
fresh-context iteration.
---
## Feature Catalog
### From mini-swe-agent (the base)
| Feature | Description |
|---|---|
| Linear history | Every step appends to `messages`. The trajectory IS the message list. No hidden state, trivially debuggable. |
| Stateless execution | `subprocess.run` per action, not a persistent shell. No stateful session can accumulate and go wrong. |
| `max_consecutive_format_errors` | Exit after N format errors in a row. Catches degenerate output structurally — garbage doesn't parse. |
| `step_limit` | Hard stop on number of LLM calls. |
| `cost_limit` | Hard stop on accumulated API cost. |
| `wall_time_limit_seconds` | Hard stop on elapsed time. |
| `role: "exit"` protocol | The loop breaks on an exit message. Model can signal completion, or any limit/error forces one. |
| Config via Pydantic | Type-safe config with defaults, validation, and YAML/CLI/env merging. |
| Trajectory serialization | Full message history + metadata saved as JSON for debugging and fine-tuning. |
### From Ralph Wiggum (anti-loop methodology)
| Feature | Description |
|---|---|
| Fresh-context reset | On degenerate loop detection, kill the context and restart with a fresh message list + journal summary. The model can't loop on identical context if the context is always fresh. |
| Disk-persisted state | The `Journal` lives on disk, not in the model's context. State survives resets without context bloat. |
| Completion signal | Structured `<done>` tag (not free-form SUBMIT) that the harness greps for. Reliable, unambiguous. |
| `max_retries` tracking | After N attempts on a task, flag as "stuck" and abandon. Task-level repetition detection. |
| Consecutive failure detection | 3 iterations without completion → warning to simplify the task or fix blockers. |
### From ZeroStack (loop detection)
| Feature | Description |
|---|---|
| Doom-loop detection | 3 identical tool calls in a row → `DOOM_LOOP` violation. Catches the 7,303-repetition case after 3, not 7,303. |
| Mid-turn compaction | Compact context at 80% pressure during a turn, not just between turns. Prevents context overflow mid-generation. |
| Sandbox mode | Bubblewrap/network-isolated execution. The agent's commands cannot reach the host. |
### From FestBench analysis (hard-won fixes)
| Feature | Description |
|---|---|
| Stop sequences | `stop=["User:", "\nUser:", "User:\n"]` on every API call. Prevents conversation simulation at the API level. |
| In-stream repetition detection | Rolling 50-token window; if the same sequence appears >5 times, abort the HTTP connection mid-stream. Catches degenerate output within a single API call. |
| Proper messages array | Send `{role, content}` messages directly — never flatten to plain text with `User:`/`Assistant:` markers. The model never sees the pattern it would copy. |
| Thinking level control | `chat_template_kwargs` with `enable_thinking` and `reasoning_effort`. The core measurement dimension for reasoning models. |
| Per-call maxtok by level | off:4000, low:8000, medium:16000, xhigh:12000. Prevents one degenerate turn from eating the entire budget. |
| Response truncation | Stored assistant responses capped at 2000 chars after action extraction. Prevents context poisoning from degenerate output. |
| Temperature flag | Default 0 for reproducibility; `--temperature` flag for testing loop-breaking with sampling randomness. |
### New: Trial Journal (stored memory)
The Journal is `zubble`'s distinctive feature — a persistent, structured
record of what the agent tried, what worked, and what didn't. This
prevents the agent from repeating failed approaches after a context
reset, and builds an evolving knowledge base across runs.
**Per-entry fields:**
- `command` — the shell command attempted
- `exit_code` — 0 = worked, non-zero = failed
- `result_summary` — first 200 chars of output (enough to recognize, not enough to bloat)
- `category` — `diagnostic` | `fix_attempt` | `verification` (auto-classified)
- `model_note` — optional note the model can add about why it failed
- `timestamp` — when the command was run
**Storage:** `{work_dir}/.zubble/journal.json` — one file per task, cleared on new task.
**Fed to model on context reset:**
```
[Previous attempts in this task — try a different approach]
DIAGNOSTIC:
✗ kubectl get pods -A → connection refused
✗ kubectl get nodes → connection refused
✓ kubectl --kubeconfig=/etc/kubernetes/admin.conf get nodes → Ready
FIX_ATTEMPT:
✗ kubectl patch deployment ... → forbidden (insufficient RBAC)
✓ kubectl edit deployment ... → updated
VERIFICATION:
✓ kubectl rollout status deployment/xavier-ocr → deployment successfully rolled out
[Note: direct API access requires the admin kubeconfig at /etc/kubernetes/admin.conf]
```
**Cross-run memory (optional):** If `--memory-dir` is set, the journal
is also saved to a shared directory keyed by task type. Future runs of
similar tasks can load prior journals as a starting knowledge base. This
creates an evolving "what worked for this type of problem" library.
---
## Module Layout
```
zubble/
├── LICENSE # MIT
├── README.md
├── DESIGN.md # This document
├── pyproject.toml # pip install zubble-agent
├── zubble/
│ ├── __init__.py
│ ├── agent.py # ~200 lines: the core loop
│ ├── config.py # Pydantic config models
│ ├── exceptions.py # FormatError, LimitsExceeded, etc.
│ │
│ ├── models/ # Pluggable LLM backends
│ │ ├── __init__.py # Model base class
│ │ ├── openai_model.py # OpenAI-compatible (llama.cpp, vLLM, Ollama)
│ │ ├── litellm_model.py # Hosted APIs (Anthropic, OpenAI, etc.)
│ │ └── cli_model.py # Local CLIs (claude, gemini, copilot)
│ │
│ ├── environments/ # Pluggable execution sandboxes
│ │ ├── __init__.py # Environment base class
│ │ ├── local_env.py # subprocess.run on local machine
│ │ ├── docker_env.py # docker exec
│ │ ├── kubectl_env.py # kubectl exec into a pod
│ │ └── sandboxed_env.py # bubblewrap isolation
│ │
│ ├── observers/ # Pluggable loop detectors
│ │ ├── __init__.py # Observer base class, Violation enum
│ │ ├── doom_loop.py # 3 identical commands = trigger
│ │ ├── text_repetition.py # In-stream and cross-turn text repetition
│ │ └── format_errors.py # max_consecutive_format_errors
│ │
│ ├── journal.py # TrialJournal: disk-persisted memory
│ │
│ ├── protocols/ # Task-specific message templates
│ │ ├── __init__.py
│ │ ├── repair.py # Infrastructure repair (FestBench-style)
│ │ ├── swe.py # SWE-bench-style coding tasks
│ │ └── custom.py # User-defined task protocol
│ │
│ └── run.py # CLI entry point: `zubble run --task ...`
│
└── tests/
├── test_agent.py
├── test_observers.py
├── test_journal.py
└── test_models.py
```
---
## Agent Control Flow
```python
class Agent:
def run(self, task: str) -> RunResult:
self.messages = self.protocol.initial_messages(task, self.journal)
self.journal.start(task)
while True:
# 1. Check hard limits
self._check_limits() # step, cost, wall_time
# 2. Query the model (with stop sequences, thinking control)
response = self.model.query(
self.messages,
stop=self.protocol.stop_sequences,
max_tokens=self.protocol.maxtok_for_level(self.level),
chat_template_kwargs=self.level_kwargs,
temperature=self.temperature,
)
self.messages.append(response.message)
self.cost += response.cost
self.n_calls += 1
# 3. Check for completion signal
if self.protocol.is_complete(response.text):
return RunResult(status="completed", ...)
# 4. Extract and execute actions
actions = self.protocol.parse_actions(response.text)
if not actions:
self.n_consecutive_format_errors += 1
if self.n_consecutive_format_errors >= self.config.max_consecutive_format_errors:
raise FormatError("too many format errors in a row")
continue
self.n_consecutive_format_errors = 0
for action in actions:
obs = self.env.execute(action)
self.journal.record(action, obs)
self.messages.append(self.protocol.format_observation(obs))
# 5. Run observers (loop detection)
violation = self.observer.check(self.messages, response)
if violation:
if violation.type == Violation.DOOM_LOOP:
self._handle_doom_loop(violation)
elif violation.type == Violation.TEXT_REPETITION:
self._handle_context_reset(violation)
elif violation.type == Violation.FORMAT_ERROR:
self._handle_format_error(violation)
# 6. Truncate stored response to prevent context poisoning
self._truncate_last_response(max_chars=2000)
def _handle_context_reset(self, violation):
"""Ralph Wiggum-style: fresh context, preserved journal."""
self.context_resets += 1
if self.context_resets > self.config.max_context_resets:
raise LimitsExceeded("exhausted context resets")
journal_summary = self.journal.summary()
self.messages = self.protocol.reset_messages(
task=self.task,
journal=journal_summary,
reason=str(violation),
)
# Reset per-loop tracking but keep the journal
self.n_consecutive_format_errors = 0
```
---
## CLI Usage (standalone, no FestBench)
```bash
# Install
pip install zubble-agent
# Run against any OpenAI-compatible endpoint
zubble run \
--endpoint http://192.168.5.117:8000 \
--model nemotron-3.5-bolt-74b \
--task "fix the kubelet config on node-1" \
--env docker \
--env-container bastion-03 \
--level xhigh \
--minutes 15 \
--stop-sequences "User:" "\nUser:" \
--temperature 0
# Run against a hosted API
zubble run \
--model anthropic/claude-sonnet-4 \
--task "refactor the auth module" \
--env local \
--protocol swe
# Run with trial journal + cross-run memory
zubble run \
--task "debug the cert_expiry_triage scenario" \
--env kubectl \
--memory-dir ~/.zubble/memory \
--journal
# Run as a Ralph Wiggum loop (fresh context each iteration)
zubble loop \
--task spec.md \
--max-iterations 20 \
--fresh-context \
--completion-signal "<done>"
```
---
## Integration with FestBench
`zubble` is designed to run independently, but FestBench can use it as a
drop-in replacement for `scenariobench.py`'s `make_backend`:
```python
# In scenariobench.py (future integration)
from zubble import Agent, OpenAIModel, DockerEnv, RepairProtocol, Journal
def make_backend(via, model, base, container=None):
if via == "zubble":
env = DockerEnv(container=container)
mdl = OpenAIModel(endpoint=base, model=model)
protocol = RepairProtocol(report_path="/findings/report.md")
journal = Journal(work_dir=f"/tmp/festbench_{container}")
def gen(messages, timeout, kwargs=None, maxtok=4000, stop=None):
agent = Agent(mdl, env, protocol, journal, level_kwargs=kwargs)
result = agent.run_task(messages, timeout=timeout, maxtok=maxtok)
return result.text, None, result.perf
return gen
```
FestBench's scenario orchestration (kind clusters, oracles, CWD tracking)
stays in `scenariobench.py`. `zubble` handles only the agent loop — the part
that was broken.
---
## Comparison
The four harnesses below cover the corners of the industry design space
`zubble` distances itself from. **OpenHands** is the heavyweight multi-agent
platform — web UI, Agent-Client Protocol, Docker sandboxes, automations,
and a multi-repo TypeScript/Python codebase. **Aider** is the terminal
pair-programmer: git-integrated, litellm-backed, with an architect/editor
split for reasoning-then-edit workflows. **opencode** is the
permission-driven CLI agent — MCP support, subagents, and a built-in
`doom_loop` permission that triggers when the same tool call repeats three
times. **pi.dev** is a minimal, extensible TypeScript harness with skills,
extensions, packages, MCP and ACP support; it is also the harness FestBench
currently runs via `pi-harness:0.85.1`, so the `pi.dev` column shows
exactly what `zubble` adds beyond the status quo.
| Feature | mini-swe-agent | Ralph Wiggum | ZeroStack | OpenHands | Aider | opencode | pi.dev | **zubble** |
|---|---|---|---|---|---|---|---|---|
| Core LoC | ~200 | ~400 (bash) | ~30K (Rust) | ~100K+ (TS/Py) | ~50K (Py) | ~100K (TS) | ~50K (TS) | **~200 (Py)** |
| License | MIT | MIT | GPL-3.0 | MIT | Apache-2.0 | MIT | MIT | **MIT** |
| Stop sequences | partial | N/A | no | no | no | no | no | **yes** |
| In-stream repetition detection | no | no | no | no | no | no | no | **yes** |
| Proper messages array | yes | N/A | yes | yes | yes | yes | yes | **yes** |
| Doom-loop detection (3x) | no | no | yes | partial | partial | yes (built-in `doom_loop` permission) | no | **yes** |
| Fresh-context reset | no | yes (core) | yes | no | no | partial (compaction) | partial (auto-compaction + branch summary) | **yes** |
| Trial journal | no | no | no | no | no (git commits) | no (snapshots) | no (sessions JSONL is telemetry, not fed back) | **yes** |
| Cross-run memory | no | no | no | no | no | yes (memories + skills) | yes (skills, extensions, packages) | **yes** (optional) |
| Thinking level control | partial | N/A | no | yes | yes | no | yes (per-model overrides) | **yes** |
| Per-call maxtok by level | no | N/A | no | no | no | no | no | **yes** |
| Wall-time / step / cost limits | yes | yes | yes | yes | partial | no | partial (overflow recovery, context budget) | **yes** |
| Stateless execution | yes | N/A | yes | no | no | no | no | **yes** |
| Sandbox isolation | partial | N/A | yes (bubblewrap) | yes (Docker) | no (host fs) | partial (permissions) | no (no built-in sandbox) | **yes** |
| Pluggable environments | yes | no | no | yes (backends) | no | no | no | **yes** |
| Pluggable models | yes (litellm) | no (CLIs only) | yes | yes | yes (litellm) | yes (providers) | yes (providers, llama.cpp, custom) | **yes** |
| FestBench integration | no | no | no | no | no | no | yes (`pi-harness:0.85.1`, `--via pi`) | **yes** (optional) |
| Standalone operation | yes | yes | yes | yes | yes | yes | yes | **yes** |
| GUI / TUI | CLI | CLI | CLI | Web UI | TUI | TUI + desktop | TUI (+ print/JSON/RPC/SDK) | CLI |
| MCP / ACP support | no | no | no | ACP | no | MCP | both | no |
| Multi-agent orchestration | no | no | no | yes (automations + agent server) | partial (architect/editor split) | yes (subagents, `subagent_depth`) | no (explicit; skips sub-agents and plan mode) | no |
---
## Roadmap
### Phase 0: Skeleton (this document)
- [x] `zubble/` directory created in llm_benchmark repo
- [x] LICENSE (MIT)
- [x] DESIGN.md (this file)
### Phase 1: Structured protocol + plan-state ledger
- [x] `protocol.py` — JSON schema, system prompt template, parse-retry wrapper
- [x] `ledger.py` — plan state machine, attempted actions, abandoned approaches
with causal failure reasons, divergence detection, summary injection
- [x] `stream_chat` in benchlib.py — accepts optional `system_prompt` and
`response_format` (defaults None; existing callers unaffected)
- [x] `scenariobench.py` integration — `--structured` flag enables JSON protocol
+ ledger injection; existing text protocol preserved as default
- [ ] Tests: re-run 5 xhigh DepthSweep scenarios, compare against 7303-rep baseline
### Phase 2: In-stream repetition detection
- [ ] `repetition.py` — rolling 10-gram CRC32 fingerprint, last-200-token window
- [ ] Graduated thresholds:
50 reps → soft nudge injected into stream
500 reps → parallel self-check (diagnostic, no action)
1000 reps → "STOP! Restart with fresh context or you will be terminated."
(warning only, does NOT kill)
1500 reps → self-confirmation gate: model reviews progress, answers
true/false. TRUE → kill. FALSE → continue.
3000 reps → unconditional kill (final safety net)
- [ ] Self-check failure/timeout → continue, not kill
- [ ] Tests for repetition detection at each threshold
### Phase 3: Ledger hierarchical summarization
- [ ] Category buckets (network/compute/storage/auth/rbac/cert)
- [ ] Collapse oldest entries into summary when `attempted` > 20
- [ ] Context reset with ledger summary on 1500-rep kill
### Phase 4: CLI + Polish
- [ ] `run.py` — CLI entry point
- [ ] `pyproject.toml` — pip installable
- [ ] `README.md` with quickstart
- [ ] Integration test: run one FestBench scenario end-to-end
### Phase 5 (deferred): Web search
- [ ] Per-scenario opt-in flag in ScenarioBench YAML
- [ ] DuckDuckGo/Brave/SearXNG fetch + model-extracted relevant steps
- [ ] Measure search+apply vs pure-reasoning separately
---
## Attribution
`zubble` draws heavily from:
- [mini-swe-agent](https://github.com/SWE-agent/mini-swe-agent) by Princeton/Stanford (MIT)
- [Ralph Wiggum](https://github.com/fstandhartinger/ralph-wiggum) by fstandhartinger (MIT), based on Geoffrey Huntley's methodology
- [ZeroStack](https://github.com/gi-dellav/zerostack) by gi-dellav (GPL-3.0) — doom-loop and compaction patterns
- FestBench's `scenariobench.py` — the hard-won lessons from degenerate loops

21
LICENSE Normal file
View file

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 pch
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

88
README.md Normal file
View file

@ -0,0 +1,88 @@
# zubble
A minimal, hackable agentic harness for LLM-driven repair and coding tasks.
`zubble` combines the simplicity of [mini-swe-agent](https://github.com/SWE-agent/mini-swe-agent),
the anti-loop discipline of [Ralph Wiggum](https://github.com/fstandhartinger/ralph-wiggum),
the doom-loop detection of [ZeroStack](https://github.com/gi-dellav/zerostack),
and hard-won lessons from FestBench's ScenarioBench — in ~200 lines of readable Python.
## Why?
Heavyweight agent platforms (Goose, Crush, OpenHands) are powerful but opaque.
`zubble` is the opposite: a loop you can read in one sitting, extend with a
subclass, and run against any endpoint — no framework tax.
It was designed to make five failure modes structurally impossible:
1. **Conversation simulation** — stop sequences + proper messages array
2. **Degenerate repetition** — in-stream repetition detection + doom-loop detector
3. **Context poisoning** — response truncation + fresh-context reset
4. **Budget exhaustion** — per-call maxtok caps by thinking level
5. **Amnesia** — a disk-persisted trial journal that survives context resets
## Quickstart
```bash
# Install (when published)
pip install zubble-agent
# Run against any OpenAI-compatible endpoint (llama.cpp, vLLM, Ollama)
zubble run \
--endpoint http://192.168.5.117:8000 \
--task "fix the kubelet config on node-1" \
--env docker --env-container bastion-03 \
--level xhigh --minutes 15
# Run against a hosted API
zubble run --model anthropic/claude-sonnet-4 --task "refactor auth" --env local
# Run as a Ralph Wiggum loop (fresh context each iteration)
zubble loop --task spec.md --max-iterations 20 --fresh-context
# Run with trial journal + cross-run memory
zubble run --task "debug cert_expiry" --env kubectl --memory-dir ~/.zubble/memory
```
## Key features
- **Stop sequences** on every API call — prevents conversation simulation
- **In-stream repetition detection** — aborts degenerate output mid-generation
- **Proper messages array** — never flattens to `User:`/`Assistant:` text
- **Doom-loop detection** — 3 identical commands triggers a warning
- **Fresh-context reset** — on degenerate loop, restart with journal summary
- **Trial journal** — disk-persisted record of what worked and what didn't
- **Thinking level control** — `chat_template_kwargs` for reasoning models
- **Per-call maxtok by level** — off/low/medium/xhigh budgets
- **Temperature flag** — default 0 for reproducibility, overridable
- **Pluggable** — `Model`, `Environment`, `Observer`, `Protocol` are interfaces
- **MIT licensed** — no copyleft, no vendor lock-in
## Architecture
Four pluggable components:
| Component | Role | Examples |
|---|---|---|
| `Model` | LLM API client | `OpenAIModel`, `LitellmModel`, `CliModel` |
| `Environment` | Command execution sandbox | `LocalEnv`, `DockerEnv`, `KubectlEnv` |
| `Observer` | Loop/degeneration detector | `DoomLoopObserver`, `TextRepetitionObserver` |
| `Journal` | Disk-persisted trial memory | `TrialJournal` |
Plus `Protocol` classes that define task-specific message templates
(repair, SWE-bench, custom).
## Design document
See [DESIGN.md](DESIGN.md) for the full architecture, control flow,
feature catalog, comparison table, and roadmap.
## License
MIT — see [LICENSE](LICENSE).
## Attribution
`zubble` draws from mini-swe-agent (MIT, Princeton/Stanford), Ralph Wiggum
(MIT, fstandhartinger / Geoffrey Huntley), ZeroStack (GPL-3.0, gi-dellav),
and FestBench's ScenarioBench harness.

31
pyproject.toml Normal file
View file

@ -0,0 +1,31 @@
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "zubble-agent"
version = "0.1.0"
description = "A minimal, hackable agentic harness for LLM-driven repair and coding tasks."
readme = "README.md"
license = { text = "MIT" }
requires-python = ">=3.9"
authors = [{ name = "pch" }]
keywords = ["llm", "agent", "harness", "swe-agent", "doom-loop"]
classifiers = [
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Topic :: Scientific/Engineering :: Artificial Intelligence",
]
dependencies = []
[project.optional-dependencies]
dev = ["pytest>=7", "pyinstaller>=6"]
[project.scripts]
zubble = "zubble.cli:main"
[tool.setuptools.packages.find]
include = ["zubble*"]
[tool.pytest.ini_options]
testpaths = ["tests"]

146
scripts/make_release.py Normal file
View file

@ -0,0 +1,146 @@
#!/usr/bin/env python3
"""Create a Forgejo release for the current tag and attach a binary asset.
Invoked by .forgejo/workflows/release.yml after the single-file executable
is built. Uses only the stdlib (urllib) so it runs in the bare CI Python
with no extra dependencies.
Environment:
FORGEJO_TOKEN API token with contents:write (the workflow's GITHUB_TOKEN)
FORGEJO_API base API URL, e.g. https://build.diganthrope.com/api/v1
REPO owner/name, e.g. pch/zubble
TAG release tag, e.g. v0.1.0
GITHUB_WORKSPACE checkout root (for git log). Defaults to cwd.
Usage:
python scripts/make_release.py <asset_path>
"""
import json
import os
import subprocess
import sys
import urllib.error
import urllib.request
from pathlib import Path
def _notes(repo_root: str, tag: str) -> str:
"""Release notes = commit subjects since the previous tag (or full log)."""
prev = _previous_tag(repo_root, tag)
if prev:
rng = f"{prev}..{tag}"
header = f"### Changes since {prev}\n\n"
else:
rng = tag
header = "### Changes\n\n"
try:
out = subprocess.run(
["git", "log", "--pretty=format:- %s", rng],
cwd=repo_root, capture_output=True, text=True, check=True,
).stdout.strip()
except Exception:
return header.strip()
return header + out
def _previous_tag(repo_root: str, tag: str) -> str | None:
try:
out = subprocess.run(
["git", "tag", "--sort=-v:refname", "--list", "v*"],
cwd=repo_root, capture_output=True, text=True, check=True,
).stdout.split()
except Exception:
return None
# tags strictly before this one, newest first
if tag in out:
out = out[out.index(tag) + 1:]
return out[0] if out else None
def _api(method, path, token, api_base, body=None, multipart=None):
url = f"{api_base}/{path.lstrip('/')}"
headers = {"Authorization": f"token {token}"}
data = None
if body is not None:
data = json.dumps(body).encode()
headers["Content-Type"] = "application/json"
elif multipart is not None:
data, boundary = multipart
headers["Content-Type"] = f"multipart/form-data; boundary={boundary}"
req = urllib.request.Request(url, data=data, method=method, headers=headers)
try:
with urllib.request.urlopen(req) as r:
raw = r.read()
return r.status, json.loads(raw) if raw else {}
except urllib.error.HTTPError as e:
return e.code, e.read().decode(errors="replace")
def _upload_asset(release_id, asset_path, token, api_base, repo):
"""Upload one file as a release attachment (multipart form-data)."""
boundary = "----zubble-release-boundary"
path = Path(asset_path)
fields = {"name": path.name}
parts = []
for k, v in fields.items():
parts.append(f"--{boundary}\r\n"
f'Content-Disposition: form-data; name="{k}"\r\n\r\n'
f"{v}\r\n")
body = "".join(parts).encode()
body += (f"--{boundary}\r\n"
'Content-Disposition: form-data; name="attachment"; '
f'filename="{path.name}"\r\n'
"Content-Type: application/octet-stream\r\n\r\n").encode()
body += path.read_bytes()
body += f"\r\n--{boundary}--\r\n".encode()
status, resp = _api(
"POST",
f"repos/{repo}/releases/{release_id}/assets",
token, api_base, multipart=(body, boundary),
)
if status not in (200, 201):
print(f"asset upload failed ({status}): {resp}", file=sys.stderr)
sys.exit(1)
print(f" attached {path.name} ({path.stat().st_size} bytes)")
def main():
if len(sys.argv) != 2:
print("usage: make_release.py <asset_path>", file=sys.stderr)
sys.exit(2)
asset = sys.argv[1]
token = os.environ["FORGEJO_TOKEN"]
api = os.environ["FORGEJO_API"]
repo = os.environ["REPO"]
tag = os.environ["TAG"]
root = os.environ.get("GITHUB_WORKSPACE", os.getcwd())
notes = _notes(root, tag)
status, resp = _api(
"POST", f"repos/{repo}/releases", token, api,
body={"tag_name": tag, "name": tag, "body": notes,
"draft": False, "prerelease": False},
)
if status == 409:
# Release already exists for this tag — fetch it.
status, resp = _api(
"GET", f"repos/{repo}/releases/tags/{tag}", token, api,
)
if status != 200:
print(f"release exists but fetch failed ({status}): {resp}",
file=sys.stderr)
sys.exit(1)
elif status not in (200, 201):
print(f"release create failed ({status}): {resp}", file=sys.stderr)
sys.exit(1)
else:
print(f"created release {tag}")
release_id = resp["id"]
print(f" release id: {release_id}")
_upload_asset(release_id, asset, token, api, repo)
print(f"release {tag} published with {Path(asset).name}")
if __name__ == "__main__":
main()

View file

@ -0,0 +1,142 @@
#!/usr/bin/env python3
"""Tests for stream_chat watcher abort hook.
The watcher lets a caller abort generation mid-stream when an in-stream
repetition detector fires. These tests stub only HTTP I/O (the SSE
response), never the logic that processes deltas or manages the abort.
python3 -m unittest discover -s tests -v
"""
import os
import sys
import json
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from zubble import benchlib # noqa: E402
from zubble.doom import StreamRepetitionDetector, Level # noqa: E402
class _SSEStub:
"""A minimal file-like that yields SSE data frames then [DONE]."""
def __init__(self, frames):
self._data = b""
for payload in frames:
self._data += f"data: {json.dumps(payload)}\n\n".encode()
self._data += b"data: [DONE]\n\n"
self._pos = 0
def read(self, n=4096):
chunk = self._data[self._pos:self._pos + n]
self._pos += len(chunk)
return chunk
def __iter__(self):
for line in self._data.split(b"\n"):
yield line + b"\n"
def __enter__(self):
return self
def __exit__(self, *a):
return False
def _content_frames(text, n, finish="stop", total=None):
"""Build n content-delta frames + finish + usage."""
frames = []
for _ in range(n):
frames.append({"choices": [{"delta": {"content": text}}]})
frames.append({"choices": [{"finish_reason": finish}]})
frames.append({"usage": {"completion_tokens": total or n,
"prompt_tokens": 10}})
return frames
class StreamChatWatcherTest(unittest.TestCase):
def setUp(self):
self._urlopen = benchlib.urllib.request.urlopen
os.environ["BENCH_MODEL"] = "test-model"
def tearDown(self):
benchlib.urllib.request.urlopen = self._urlopen
os.environ.pop("BENCH_MODEL", None)
def _stub_urlopen(self, frames):
def _urlopen(req, timeout=2400):
return _SSEStub(frames)
benchlib.urllib.request.urlopen = _urlopen
def test_watcher_aborts_mid_stream(self):
self._stub_urlopen(_content_frames("AB ", 100))
count = [0]
def watcher(kind, text):
count[0] += 1
return count[0] >= 30
content, reasoning, perf = benchlib.stream_chat(
"http://x", "prompt", timeout=10, watcher=watcher)
self.assertEqual(perf["finish"], "aborted")
self.assertEqual(count[0], 30)
self.assertEqual(len(content), 30 * 3)
def test_no_watcher_no_abort(self):
self._stub_urlopen(_content_frames("hello", 5, total=7))
content, reasoning, perf = benchlib.stream_chat(
"http://x", "prompt", timeout=10)
self.assertNotEqual(perf["finish"], "aborted")
self.assertEqual(content, "hello" * 5)
def test_watcher_receives_reasoning_kind(self):
frames = []
for _ in range(50):
frames.append({"choices": [{"delta": {"reasoning_content": "x "}}]})
for _ in range(50):
frames.append({"choices": [{"delta": {"content": "y "}}]})
frames.append({"choices": [{"finish_reason": "stop"}]})
frames.append({"usage": {"completion_tokens": 100, "prompt_tokens": 10}})
self._stub_urlopen(frames)
kinds = []
def watcher(kind, text):
kinds.append(kind)
return len(kinds) >= 60
content, reasoning, perf = benchlib.stream_chat(
"http://x", "prompt", timeout=10, watcher=watcher)
self.assertEqual(perf["finish"], "aborted")
self.assertEqual(kinds[:50], ["reasoning"] * 50)
self.assertIn("content", kinds)
def test_watcher_false_does_not_abort(self):
self._stub_urlopen(_content_frames("X", 10, total=12))
def watcher(kind, text):
return False
content, reasoning, perf = benchlib.stream_chat(
"http://x", "prompt", timeout=10, watcher=watcher)
self.assertNotEqual(perf["finish"], "aborted")
self.assertEqual(content, "X" * 10)
def test_integration_with_stream_repetition_detector(self):
detector = StreamRepetitionDetector()
def watcher(kind, text):
result = detector.feed(kind, text)
return result != Level.NONE
self._stub_urlopen(_content_frames("x ", 51 * 50))
content, reasoning, perf = benchlib.stream_chat(
"http://x", "prompt", timeout=10, watcher=watcher)
self.assertEqual(perf["finish"], "aborted")
self.assertEqual(detector.level, Level.L1)
if __name__ == "__main__":
unittest.main()

247
tests/test_zubble_agent.py Normal file
View file

@ -0,0 +1,247 @@
#!/usr/bin/env python3
"""Tests for zubble agent loop orchestration.
These tests mock only I/O (stream_chat HTTP calls, env.exec/verify),
never the orchestration logic that wires detectors, protocol, ledger,
and abort handling together.
python3 -m unittest discover -s tests -v
"""
import os
import sys
import json
import time
import unittest
from unittest.mock import MagicMock, patch
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from zubble import benchlib # noqa: E402
from zubble.agent import run_agent_loop # noqa: E402
from zubble import protocol # noqa: E402
class _MockEnv:
"""Minimal env stub matching scenariobench's interface."""
intro = "You have a root shell on a host."
report_path = "/root/report.md"
def __init__(self, exec_results=None, verify_result=(1, "unhealthy")):
self._exec_results = exec_results or []
self._exec_idx = 0
self._verify_result = verify_result
def exec(self, cmd, timeout=90):
if self._exec_idx < len(self._exec_results):
r = self._exec_results[self._exec_idx]
self._exec_idx += 1
return r
return (0, "")
def verify(self):
return self._verify_result
def setup(self):
return True, "ok"
def teardown(self):
pass
def motd(self):
return "welcome"
def report(self):
return ""
def _make_stream_response(content):
"""Build a minimal perf dict + content for a mock stream_chat return."""
perf = dict(ttft=0.01, t_answer=0.01, wall=0.01,
reasoning=0, visible=len(content), total=len(content),
decode=0.0, user_tps=0.0, prompt_tokens=10,
prefill=0.0, finish="stop")
return content, "", perf
class AgentLoopActionKillTest(unittest.TestCase):
"""Verify that repeated identical actions trigger kill → fresh context
→ reset cap → doom_loop."""
def test_action_repetition_leads_to_doom_loop(self):
repeated_action = json.dumps({
"action": "run",
"command": "kubectl get pods",
"plan_step_id": 1,
"reasoning": "checking pods",
})
def mock_stream_chat(*args, **kwargs):
return _make_stream_response(repeated_action)
with patch.object(benchlib, "stream_chat", side_effect=mock_stream_chat):
env = _MockEnv(exec_results=[(0, "NAME READY STATUS\npod-a Running")] * 100)
deadline = time.monotonic() + 10
result = run_agent_loop(
"http://x", env, "Fix the cluster", "Connected.",
deadline, "off", None, max_turns=200, maxtok=4000, name="test")
self.assertEqual(result["outcome"], "doom_loop")
self.assertTrue(result["turns"] > 0)
def test_plan_then_run_then_submit_passes(self):
responses = [
json.dumps({"action": "plan", "steps": ["diagnose", "fix", "verify"],
"reasoning": "3-step plan"}),
json.dumps({"action": "run", "command": "kubectl get pods",
"plan_step_id": 1, "reasoning": "step 1"}),
json.dumps({"action": "run", "command": "kubectl apply -f fix.yaml",
"plan_step_id": 2, "reasoning": "step 2"}),
json.dumps({"action": "submit", "report": "Fixed the cert.",
"reasoning": "done"}),
]
idx = [0]
def mock_stream_chat(*args, **kwargs):
r = _make_stream_response(responses[idx[0]])
idx[0] += 1
return r
with patch.object(benchlib, "stream_chat", side_effect=mock_stream_chat):
env = _MockEnv(
exec_results=[(0, "pods running"), (0, "applied")],
verify_result=(0, "cluster healthy"),
)
deadline = time.monotonic() + 10
result = run_agent_loop(
"http://x", env, "Fix the cluster", "Connected.",
deadline, "off", None, max_turns=200, maxtok=4000, name="test")
self.assertTrue(result["passed"])
self.assertEqual(result["outcome"], "passed")
self.assertEqual(result["submissions"], 1)
self.assertEqual(result["commands"], 2)
def test_agent_requests_constrained_json_schema(self):
response = json.dumps({"action": "submit", "report": "fixed",
"reasoning": "verified"})
with patch.object(benchlib, "stream_chat",
return_value=_make_stream_response(response)) as stream_chat:
env = _MockEnv(verify_result=(0, "cluster healthy"))
result = run_agent_loop(
"http://x", env, "Fix the cluster", "Connected.",
time.monotonic() + 10, "off", None,
max_turns=200, maxtok=4000, name="test")
self.assertTrue(result["passed"])
self.assertEqual(stream_chat.call_args.kwargs["response_format"],
protocol.RESPONSE_FORMAT)
def test_rejected_submit_continues_until_later_pass(self):
responses = [
json.dumps({"action": "plan", "steps": ["check"], "reasoning": "plan"}),
json.dumps({"action": "run", "command": "kubectl get pods",
"plan_step_id": 1, "reasoning": "check"}),
json.dumps({"action": "submit", "report": "first attempt",
"reasoning": "ready to verify"}),
json.dumps({"action": "run", "command": "kubectl apply -f fix.yaml",
"plan_step_id": 1, "reasoning": "apply correction"}),
json.dumps({"action": "submit", "report": "fix applied",
"reasoning": "ready to verify again"}),
]
idx = [0]
verify_calls = [0]
def mock_stream_chat(*args, **kwargs):
r = _make_stream_response(responses[idx[0]])
idx[0] += 1
return r
def verify():
verify_calls[0] += 1
return ((1, "still unhealthy") if verify_calls[0] == 1
else (0, "cluster healthy"))
with patch.object(benchlib, "stream_chat", side_effect=mock_stream_chat):
env = _MockEnv(exec_results=[(0, "pods running"), (0, "applied")])
env.verify = verify
deadline = time.monotonic() + 10
result = run_agent_loop(
"http://x", env, "Fix the cluster", "Connected.",
deadline, "off", None, max_turns=200, maxtok=4000, name="test")
self.assertTrue(result["passed"])
self.assertEqual(result["outcome"], "passed")
self.assertEqual(result["submissions"], 2)
self.assertEqual(result["rejected"], 1)
self.assertEqual(result["commands"], 2)
def test_malformed_response_then_valid_repair_executes_action(self):
responses = [
'{"action":"finished","reasoning":"done"}',
json.dumps({"action": "submit", "report": "fixed",
"reasoning": "ready to verify"}),
]
idx = [0]
def mock_stream_chat(*args, **kwargs):
r = _make_stream_response(responses[idx[0]])
idx[0] += 1
return r
with patch.object(benchlib, "stream_chat", side_effect=mock_stream_chat):
env = _MockEnv(verify_result=(0, "cluster healthy"))
result = run_agent_loop(
"http://x", env, "Fix the cluster", "Connected.",
time.monotonic() + 10, "off", None,
max_turns=200, maxtok=4000, name="test")
self.assertTrue(result["passed"])
self.assertEqual(result["submissions"], 1)
self.assertEqual(result["parse_failures"], 1)
def test_malformed_response_and_repair_end_as_parse_error(self):
def mock_stream_chat(*args, **kwargs):
return _make_stream_response('{"action":"submit","report":"bad\nreport"}')
with patch.object(benchlib, "stream_chat", side_effect=mock_stream_chat):
env = _MockEnv()
result = run_agent_loop(
"http://x", env, "Fix the cluster", "Connected.",
time.monotonic() + 10, "off", None,
max_turns=200, maxtok=4000, name="test")
self.assertFalse(result["passed"])
self.assertEqual(result["outcome"], "parse_error")
self.assertEqual(result["turns"], 1)
self.assertEqual(result["parse_failures"], 2)
self.assertEqual(result["submissions"], 0)
class ProtocolTest(unittest.TestCase):
def test_final_and_finished_are_rejected_with_submit_guidance(self):
for action in ("final", "finished"):
with self.subTest(action=action):
with self.assertRaisesRegex(protocol.ProtocolError,
"use 'submit'"):
protocol.parse_response(json.dumps({
"action": action, "report": "done", "reasoning": "done",
}))
def test_trailing_content_is_rejected(self):
with self.assertRaisesRegex(protocol.ProtocolError, "trailing content"):
protocol.parse_response(
'{"action":"submit","report":"done","reasoning":"done"} trailing')
def test_long_escaped_report_is_valid(self):
response = json.dumps({
"action": "submit",
"report": "line one\nline two with {braces} and \\\"quotes\\\"",
"reasoning": "verified",
})
self.assertEqual(protocol.parse_response(response)["action"], "submit")
if __name__ == "__main__":
unittest.main()

238
tests/test_zubble_doom.py Normal file
View file

@ -0,0 +1,238 @@
#!/usr/bin/env python3
"""Unit tests for zubble doom-loop escape hatch detectors.
Pure-logic tests — no I/O, no mocking. The detectors are deterministic
state machines; we feed them synthetic token streams and action
sequences and verify threshold crossings, resets, and the accepted
period-37 blind spot.
python3 -m unittest discover -s tests -v
"""
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from zubble.doom import ( # noqa: E402
ActionRepetitionDetector,
ActionSignal,
StreamRepetitionDetector,
Level,
InsuranceCounter,
ResetCap,
canonical_form,
)
# ── Helpers ────────────────────────────────────────────────────────
def _feed_blocks(detector, text, n_blocks, block_size=50):
"""Feed n_blocks identical 50-delta blocks to the detector."""
for _ in range(n_blocks):
for _ in range(block_size):
detector.feed("content", text)
def _feed_cycle(detector, cycle_len, n_cycles, block_size=50):
"""Feed a repeating pattern with period cycle_len. Each position in
the cycle gets a unique token so blocks only match if the period
divides the block size."""
for _ in range(n_cycles):
for i in range(cycle_len):
detector.feed("content", f"t{i} ")
# ── ActionRepetitionDetector ───────────────────────────────────────
class ActionRepetitionTest(unittest.TestCase):
def test_two_identical_silent(self):
d = ActionRepetitionDetector()
a = {"action": "run", "command": "kubectl get pods", "reasoning": "r", "plan_step_id": 1}
self.assertEqual(d.feed(a), ActionSignal.NONE)
self.assertEqual(d.feed(a), ActionSignal.NONE)
def test_third_identical_notes(self):
d = ActionRepetitionDetector()
a = {"action": "run", "command": "kubectl get pods", "reasoning": "r", "plan_step_id": 1}
d.feed(a)
d.feed(a)
self.assertEqual(d.feed(a), ActionSignal.NOTE)
def test_fourth_identical_kills(self):
d = ActionRepetitionDetector()
a = {"action": "run", "command": "kubectl get pods", "reasoning": "r", "plan_step_id": 1}
d.feed(a)
d.feed(a)
d.feed(a)
self.assertEqual(d.feed(a), ActionSignal.KILL)
def test_different_action_resets(self):
d = ActionRepetitionDetector()
a = {"action": "run", "command": "kubectl get pods", "reasoning": "r", "plan_step_id": 1}
b = {"action": "run", "command": "kubectl get nodes", "reasoning": "r2", "plan_step_id": 2}
d.feed(a)
d.feed(a)
d.feed(a) # count=3 → NOTE
self.assertEqual(d.feed(b), ActionSignal.NONE) # reset
self.assertEqual(d.feed(b), ActionSignal.NONE) # count=2
self.assertEqual(d.feed(b), ActionSignal.NOTE) # count=3
def test_reasoning_excluded_from_canonical(self):
d = ActionRepetitionDetector()
a1 = {"action": "run", "command": "ls", "reasoning": "thinking A", "plan_step_id": 1}
a2 = {"action": "run", "command": "ls", "reasoning": "thinking B", "plan_step_id": 1}
self.assertEqual(d.feed(a1), ActionSignal.NONE)
self.assertEqual(d.feed(a2), ActionSignal.NONE) # still count=2, different reasoning
self.assertEqual(d.feed(a1), ActionSignal.NOTE) # count=3
def test_reset_clears_state(self):
d = ActionRepetitionDetector()
a = {"action": "run", "command": "ls", "reasoning": "r", "plan_step_id": 1}
d.feed(a)
d.feed(a)
d.feed(a) # count=3 → NOTE
d.reset()
self.assertEqual(d.feed(a), ActionSignal.NONE) # count=1 after reset
def test_canonical_form_excludes_reasoning(self):
a1 = {"action": "run", "command": "ls", "reasoning": "A", "plan_step_id": 1}
a2 = {"action": "run", "command": "ls", "reasoning": "B", "plan_step_id": 1}
self.assertEqual(canonical_form(a1), canonical_form(a2))
def test_canonical_form_includes_other_fields(self):
a1 = {"action": "run", "command": "ls", "reasoning": "A", "plan_step_id": 1}
a2 = {"action": "run", "command": "ls", "reasoning": "A", "plan_step_id": 2}
self.assertNotEqual(canonical_form(a1), canonical_form(a2))
# ── StreamRepetitionDetector ───────────────────────────────────────
class StreamRepetitionTest(unittest.TestCase):
def test_nonrepeating_stream_never_fires(self):
s = StreamRepetitionDetector()
for i in range(200):
s.feed("content", f"unique{i} ")
self.assertEqual(s.level, Level.NONE)
self.assertEqual(s.streak, 0)
def test_l1_fires_at_streak_50(self):
s = StreamRepetitionDetector()
_feed_blocks(s, "hello ", n_blocks=51) # 51 blocks → streak=50
self.assertEqual(s.level, Level.L1)
def test_l2_fires_at_streak_500(self):
s = StreamRepetitionDetector()
_feed_blocks(s, "x", n_blocks=501) # 501 blocks → streak=500
self.assertEqual(s.level, Level.L2)
def test_l3_fires_at_streak_1500(self):
s = StreamRepetitionDetector()
_feed_blocks(s, "x", n_blocks=1501) # 1501 blocks → streak=1500
self.assertEqual(s.level, Level.L3)
def test_level_escalates_l1_to_l2_to_l3(self):
s = StreamRepetitionDetector()
_feed_blocks(s, "x", n_blocks=51)
self.assertEqual(s.level, Level.L1)
_feed_blocks(s, "x", n_blocks=450) # total streak=500
self.assertEqual(s.level, Level.L2)
_feed_blocks(s, "x", n_blocks=1000) # total streak=1500
self.assertEqual(s.level, Level.L3)
def test_different_block_resets_streak(self):
s = StreamRepetitionDetector()
_feed_blocks(s, "A", n_blocks=10) # streak=9
_feed_blocks(s, "B", n_blocks=1) # different block → streak=0
self.assertEqual(s.streak, 0)
self.assertEqual(s.level, Level.NONE)
def test_reasoning_and_content_combined(self):
s = StreamRepetitionDetector()
seg = "tok "
for _ in range(51):
for _ in range(50):
s.feed("reasoning", seg) # pure reasoning stream
self.assertEqual(s.level, Level.L1)
def test_mixed_kinds_combined_in_order(self):
s = StreamRepetitionDetector()
seg = "z "
for _ in range(51):
for i in range(50):
kind = "reasoning" if i % 2 == 0 else "content"
s.feed(kind, seg)
self.assertEqual(s.level, Level.L1)
def test_period_37_blind_spot(self):
"""A 37-delta cycle evades 50-delta block detection because
gcd(37, 50)=1 — each block starts at a different cycle position.
This is an accepted blind spot, pinned here as a regression."""
s = StreamRepetitionDetector()
_feed_cycle(s, cycle_len=37, n_cycles=200) # 7400 deltas, 148 blocks
self.assertEqual(s.streak, 0)
self.assertEqual(s.level, Level.NONE)
def test_period_25_detected(self):
"""A 25-delta cycle IS detected because 25 divides 50 — every
block is identical."""
s = StreamRepetitionDetector()
_feed_cycle(s, cycle_len=25, n_cycles=200) # 5000 deltas, 100 blocks
self.assertEqual(s.level, Level.L1) # streak=99 ≥ 50
# ── InsuranceCounter ───────────────────────────────────────────────
class InsuranceCounterTest(unittest.TestCase):
def test_nine_aborts_no_kill(self):
ic = InsuranceCounter()
for _ in range(9):
self.assertFalse(ic.register_abort())
self.assertEqual(ic.count, 9)
def test_tenth_abort_kills(self):
ic = InsuranceCounter()
for _ in range(9):
ic.register_abort()
self.assertTrue(ic.register_abort()) # 10th
def test_eleventh_still_kills(self):
ic = InsuranceCounter()
for _ in range(10):
ic.register_abort()
self.assertTrue(ic.register_abort()) # 11th, still >= 10
# ── ResetCap ───────────────────────────────────────────────────────
class ResetCapTest(unittest.TestCase):
def test_three_resets_ok(self):
rc = ResetCap()
for _ in range(3):
self.assertFalse(rc.register_reset())
self.assertEqual(rc.count, 3)
self.assertEqual(rc.remaining, 0)
def test_fourth_reset_exceeds_cap(self):
rc = ResetCap()
for _ in range(3):
rc.register_reset()
self.assertTrue(rc.register_reset()) # 4th → doom_loop
def test_remaining_decrements(self):
rc = ResetCap()
self.assertEqual(rc.remaining, 3)
rc.register_reset()
self.assertEqual(rc.remaining, 2)
rc.register_reset()
self.assertEqual(rc.remaining, 1)
rc.register_reset()
self.assertEqual(rc.remaining, 0)
if __name__ == "__main__":
unittest.main()

11
zubble/__init__.py Normal file
View file

@ -0,0 +1,11 @@
"""zubble — a minimal, hackable agentic harness for LLM-driven repair and coding tasks.
Public surface:
from zubble.agent import run_agent_loop
from zubble.env import LocalEnv
from zubble import protocol
from zubble.doom import StreamRepetitionDetector, Level
"""
__version__ = "0.1.0"

6
zubble/__main__.py Normal file
View file

@ -0,0 +1,6 @@
import sys
from zubble.cli import main
if __name__ == "__main__":
sys.exit(main())

339
zubble/agent.py Normal file
View file

@ -0,0 +1,339 @@
#!/usr/bin/env python3
"""agent — structured turn loop for the zubble agent.
Integrates protocol.py (JSON schema), ledger.py (plan-state memory),
doom.py (escape-hatch detectors), and benchlib.stream_chat (streaming
HTTP client) into a turn loop that replaces scenariobench's flat-text
rendering.
The loop:
1. Build prompt: system_prompt (protocol) + extra_messages (history)
+ per-turn user message (session memory + time remaining)
2. Call stream_chat with StreamRepetitionDetector as watcher
3. Handle in-stream aborts (L1 keep+note, L2 retry, L3 kill)
4. Parse JSON response via protocol.parse_response (1 repair attempt)
5. Feed parsed action to ActionRepetitionDetector (3→note, 4→kill)
6. Execute action via env (run/submit/ask/plan)
7. Record in ledger, append harness response to messages
8. On kill: fresh conversation (reset messages, keep ledger),
enforce reset cap (3 → doom_loop)
"""
import time
from zubble import benchlib
from zubble import protocol
from zubble.ledger import Ledger
from zubble.doom import (
ActionRepetitionDetector, ActionSignal,
StreamRepetitionDetector, Level,
InsuranceCounter, ResetCap,
NOTE_ACTION_REPEAT, NOTE_STREAM_L1, NOTE_STREAM_L2,
NOTE_KILL_RESET, NOTE_DOOM_LOOP,
)
_PER_CALL_TIMEOUT = {4000: 240, 8000: 240, 16000: 480, 32000: 480}
def run_agent_loop(base, env, brief, login, deadline, level, level_kwargs,
max_turns=200, maxtok=32000, show=False, name="",
proto=None):
"""Run the structured agent loop for one scenario.
``proto`` is a ``protocol.Protocol`` instance supplying the
scenario-specific framing (role, opening text, deliverable wording,
verifier feedback). Defaults to ``protocol.REPAIR`` (the original
Kubernetes repair scenario) for backward compatibility.
Returns a telemetry dict with the same keys as scenariobench.run_one's
inner loop: turns, commands, submissions, rejected, repeats, passed,
outcome, detail, gen_tok, vis_tok, rea_tok, t_first_sub, n_first_sub,
t_pass, n_pass, verdicts, messages, exchange.
"""
if proto is None:
proto = protocol.REPAIR
ledger = Ledger()
action_det = ActionRepetitionDetector()
insurance = InsuranceCounter()
reset_cap = ResetCap()
initial_content = proto.initial_message(
env.intro, brief, login, env.report_path)
messages = [{"role": "user", "content": initial_content}]
turns = commands = submissions = rejected = repeats = 0
parse_failures = 0
gen_tok = vis_tok = rea_tok = 0
passed = False
outcome = "timeout"
detail = "ran out of time"
t_first_sub = n_first_sub = t_pass = n_pass = None
verdicts = []
exchange = []
t0 = time.monotonic()
nudged = False
l2_retried = False
per_call_timeout = _PER_CALL_TIMEOUT.get(maxtok, 480)
def _fresh_context(reason):
"""Reset messages for a fresh conversation, preserving the ledger."""
nonlocal l2_retried
messages.clear()
messages.append({"role": "user", "content": proto.reset_message(
env.intro, brief, login, env.report_path,
NOTE_KILL_RESET, ledger.render_summary())})
action_det.reset()
l2_retried = False
while time.monotonic() < deadline:
if turns >= max_turns:
detail = f"hit the {max_turns} turn cap"
outcome = "turn_cap"
break
turns += 1
left = int(deadline - time.monotonic())
if not nudged and (left < (deadline - t0) * 0.3 or turns > max_turns * 0.7):
nudged = True
messages.append({"role": "user", "content":
f"[{left}s remaining] {proto.time_nudge}"})
session_memory = ledger.render_summary()
prompt = f"## SESSION MEMORY\n{session_memory}\n\n[{left}s remaining]"
stream_det = StreamRepetitionDetector()
def watcher(kind, text):
return stream_det.feed(kind, text) != Level.NONE
timeout = max(30, min(left, per_call_timeout))
try:
text, _reasoning, perf = benchlib.stream_chat(
base, prompt, level_kwargs,
maxtok=maxtok, timeout=timeout,
system_prompt=proto.system_prompt(),
response_format=protocol.RESPONSE_FORMAT,
extra_messages=messages,
watcher=watcher,
)
except Exception as e:
detail = f"{type(e).__name__}: {e}"
outcome = "error"
break
if time.monotonic() >= deadline:
detail = "ran out of time mid-turn"
outcome = "timeout"
break
if perf:
gen_tok += perf.get("total", 0) or 0
vis_tok += perf.get("visible", 0) or 0
rea_tok += perf.get("reasoning", 0) or 0
if show:
print(f" --- turn {turns} ---\n{text[:600]}", flush=True)
# ── In-stream abort handling ────────────────────────────────
if perf.get("finish") == "aborted":
lvl = stream_det.level
streak = stream_det.streak
if lvl == Level.L3:
if reset_cap.register_reset():
outcome = "doom_loop"
detail = NOTE_DOOM_LOOP.format(resets=reset_cap.count)
print(f" [{name}] doom_loop: {reset_cap.count} resets exceeded")
break
ledger.record_abandoned(
"in-stream doom loop (L3)",
reason=f"repeated 50-token block {streak} times")
_fresh_context("L3 kill")
print(f" [{name}] fresh context (L3 kill, streak={streak})")
continue
if lvl == Level.L2 and not l2_retried:
if insurance.register_abort():
if reset_cap.register_reset():
outcome = "doom_loop"
detail = NOTE_DOOM_LOOP.format(resets=reset_cap.count)
break
ledger.record_abandoned(
"in-stream doom loop (insurance)",
reason=f"insurance tripwire: {insurance.count} aborts")
_fresh_context("insurance kill")
continue
l2_retried = True
messages.append({"role": "user", "content":
NOTE_STREAM_L2.format(streak=streak)})
print(f" [{name}] L2 abort (streak={streak}), retrying turn")
continue
# L1, or L2 when already retried → keep partial, inject note
if insurance.register_abort():
if reset_cap.register_reset():
outcome = "doom_loop"
detail = NOTE_DOOM_LOOP.format(resets=reset_cap.count)
break
ledger.record_abandoned(
"in-stream doom loop (insurance)",
reason=f"insurance tripwire: {insurance.count} aborts")
_fresh_context("insurance kill")
continue
messages.append({"role": "assistant", "content": text})
messages.append({"role": "user", "content":
NOTE_STREAM_L1.format(streak=streak)})
l2_retried = False
print(f" [{name}] L1 abort (streak={streak}), keeping prefix")
continue
# ── Parse JSON response ─────────────────────────────────────
try:
obj = protocol.parse_response(text)
except protocol.ProtocolError as e:
parse_failures += 1
print(f" [{name}] PARSE FAIL (turn {turns}): {e}", flush=True)
messages.append({"role": "assistant", "content": text})
messages.append({"role": "user", "content": proto.retry_nudge(e)})
try:
text2, _r2, perf2 = benchlib.stream_chat(
base, prompt, level_kwargs,
maxtok=maxtok, timeout=timeout,
system_prompt=proto.system_prompt(),
response_format=protocol.RESPONSE_FORMAT,
extra_messages=messages,
)
if perf2:
gen_tok += perf2.get("total", 0) or 0
vis_tok += perf2.get("visible", 0) or 0
rea_tok += perf2.get("reasoning", 0) or 0
obj = protocol.parse_response(text2)
text = text2
except protocol.ProtocolError as e2:
parse_failures += 1
print(f" [{name}] PARSE FAIL repair failed: {e2}", flush=True)
messages.append({"role": "assistant", "content": text2 if text2 else ""})
outcome = "parse_error"
detail = f"JSON parse failed after one repair: {e2}"
break
except Exception as e2:
outcome = "error"
detail = f"parse repair {type(e2).__name__}: {e2}"
break
messages.append({"role": "assistant", "content": text})
l2_retried = False
# ── Action-level repetition detection ───────────────────────
sig = action_det.feed(obj)
if sig == ActionSignal.KILL:
if reset_cap.register_reset():
outcome = "doom_loop"
detail = NOTE_DOOM_LOOP.format(resets=reset_cap.count)
print(f" [{name}] doom_loop: action kill, cap exceeded")
break
canonical = repr({k: v for k, v in obj.items() if k != "reasoning"})
ledger.record_abandoned(
f"action doom loop: {obj.get('action', '?')} {obj.get('command', '')[:80]}",
reason=f"repeated identically {action_det.count} times")
_fresh_context("action kill")
print(f" [{name}] fresh context (action kill, count={action_det.count})")
continue
action_note = ""
if sig == ActionSignal.NOTE:
action_note = NOTE_ACTION_REPEAT.format(n=action_det.count)
print(f" [{name}] action repeat note (count={action_det.count})")
# ── Execute the action ──────────────────────────────────────
action = obj.get("action")
if action == "plan":
steps = obj.get("steps", [])
ledger.set_plan(steps)
harness_resp = (
f"Plan recorded with {len(steps)} steps. "
f"Execute step 1 now."
)
if action_note:
harness_resp += f"\n{action_note}"
messages.append({"role": "user", "content": harness_resp})
continue
if action == "run":
cmd = obj.get("command", "")
step_id = obj.get("plan_step_id")
div = ledger.check_divergence(step_id, cmd)
if div:
action_note = (action_note + "\n" + div).strip()
crc, cout = env.exec(cmd, timeout=90)
commands += 1
exchange.append({"turn": turns, "cmd": cmd, "exit": crc,
"out": cout[:8000]})
ledger.record_action(step_id, cmd, crc, cout)
harness_resp = proto.render_action_result(
"run", crc, cout, note=action_note)
messages.append({"role": "user", "content": harness_resp})
continue
if action == "submit":
submissions += 1
if t_first_sub is None:
t_first_sub = round(time.monotonic() - t0, 1)
n_first_sub = turns
report_text = obj.get("report", "")
vrc, vout = env.verify()
verdicts.append({
"n": submissions,
"at": round(time.monotonic() - t0, 1),
"accepted": vrc == 0,
"oracle": " ".join(vout.split())[:1200],
})
print(f" [{name}] submission {submissions} @ "
f"{int(time.monotonic()-t0)}s -> "
f"{'ACCEPTED' if vrc == 0 else 'REJECTED'}", flush=True)
if vrc == 0:
passed = True
outcome = "passed"
t_pass = round(time.monotonic() - t0, 1)
n_pass = turns
detail = f"passed on submission {submissions}"
break
rejected += 1
harness_resp = proto.render_action_result(
"submit", vrc, vout, note=action_note)
messages.append({"role": "user", "content": harness_resp})
continue
if action == "ask":
harness_resp = proto.render_action_result(
"ask", 0, "", note=action_note)
messages.append({"role": "user", "content": harness_resp})
continue
if action == "report":
harness_resp = (
"Report noted. Submit when you are ready for verification."
)
if action_note:
harness_resp += f"\n{action_note}"
messages.append({"role": "user", "content": harness_resp})
continue
secs = time.monotonic() - t0
return dict(
turns=turns, commands=commands, submissions=submissions,
rejected=rejected, repeats=repeats, passed=passed,
outcome=outcome, detail=detail,
gen_tok=gen_tok, vis_tok=vis_tok, rea_tok=rea_tok,
t_first_sub=t_first_sub, n_first_sub=n_first_sub,
t_pass=t_pass, n_pass=n_pass, verdicts=verdicts,
messages=messages, exchange=exchange, seconds=round(secs, 1),
parse_failures=parse_failures,
)

1331
zubble/benchlib.py Normal file

File diff suppressed because it is too large Load diff

101
zubble/cli.py Normal file
View file

@ -0,0 +1,101 @@
#!/usr/bin/env python3
"""cli — command-line entry point for the zubble agent.
Implements the ``zubble run`` interface advertised in the README:
zubble run \
--endpoint http://192.168.5.117:8000 \
--task "fix the kubelet config on node-1" \
--level xhigh --minutes 15
Only the ``local`` environment is wired here (LocalEnv — each command runs
in a fresh subprocess on the host). Docker/kubectl environments are
pluggable but not bundled; see DESIGN.md.
"""
import argparse
import os
import sys
import time
from zubble import __version__, protocol
from zubble.agent import run_agent_loop
from zubble.env import LocalEnv
def _level_kwargs(level):
"""Build chat_template_kwargs for a thinking level.
Mirrors benchlib.resolve_levels' explicit-external-model path: ``off``
disables thinking; any other name enables it and sets reasoning_effort.
"""
if level == "off":
return {"enable_thinking": False}
return {"enable_thinking": True, "reasoning_effort": level}
def _build_parser():
p = argparse.ArgumentParser(
prog="zubble",
description="A minimal, hackable agentic harness for LLM-driven "
"repair and coding tasks.",
)
p.add_argument("--version", action="version",
version=f"zubble {__version__}")
sub = p.add_subparsers(dest="command")
sub.required = True
run = sub.add_parser("run", help="run the agent loop against one task")
run.add_argument("--endpoint", required=True,
help="OpenAI-compatible endpoint URL "
"(llama.cpp, vLLM, Ollama, ...)")
run.add_argument("--task", required=True,
help="task brief / scenario description")
run.add_argument("--env", default="local", choices=["local"],
help="execution environment (only 'local' is bundled)")
run.add_argument("--report-path", default="/work/output.md",
help="file path the agent writes its deliverable to")
run.add_argument("--level", default="off",
help="thinking level (off/low/medium/high/xhigh)")
run.add_argument("--minutes", type=float, default=15.0,
help="time budget in minutes")
run.add_argument("--max-turns", type=int, default=200,
help="hard cap on agent turns")
run.add_argument("--maxtok", type=int, default=32000,
help="per-call max output tokens")
run.add_argument("--show", action="store_true",
help="print each turn's raw model output")
run.add_argument("--name", default="zubble",
help="label used in progress logs")
return p
def main(argv=None):
args = _build_parser().parse_args(argv)
if args.command == "run":
env = LocalEnv(report_path=args.report_path)
deadline = time.monotonic() + args.minutes * 60
login = f"Endpoint: {args.endpoint}"
result = run_agent_loop(
args.endpoint, env, args.task, login, deadline,
args.level, _level_kwargs(args.level),
max_turns=args.max_turns, maxtok=args.maxtok,
show=args.show, name=args.name,
proto=protocol.REPAIR,
)
print(f"\n--- result ---")
print(f"outcome: {result['outcome']}")
print(f"detail: {result['detail']}")
print(f"turns: {result['turns']}")
print(f"commands: {result['commands']}")
print(f"submissions:{result['submissions']}")
print(f"passed: {result['passed']}")
print(f"seconds: {result['seconds']}")
print(f"gen_tok: {result['gen_tok']}")
return 0 if result["outcome"] in ("passed", "timeout", "turn_cap") else 1
return 2
if __name__ == "__main__":
sys.exit(main())

242
zubble/doom.py Normal file
View file

@ -0,0 +1,242 @@
#!/usr/bin/env python3
"""doom — doom-loop escape hatch for the zubble agent loop.
Two detectors, an insurance counter, and a reset cap:
Detector A (action-level, across turns):
Tracks consecutive identical actions by canonical JSON (all fields
except 'reasoning'). 3rd consecutive identical → NOTE. 4th → KILL.
Detector B (in-stream, within one generation):
Watches the combined reasoning + content token stream. Buffers
50-delta blocks; when the current block equals the previous block,
a streak increments. Thresholds:
streak >= 50 → L1: abort, keep prefix, inject cut-note
streak >= 500 → L2: abort, discard turn, retry once
streak >= 1500 → L3: kill → fresh conversation
Insurance counter:
10 total aborts (L1 + L2) in a scenario → kill → fresh conversation.
Reset cap:
3 fresh-context resets per scenario; the 4th doom → outcome="doom_loop".
Known blind spot (accepted): repetition periods that don't divide 50
(e.g. 37-delta loops) evade Detector B. Exact-match only — no fuzzy
fingerprinting, no decay. Review findings in practice before adjusting.
"""
import hashlib
import json
from enum import IntEnum
# ── Signals ────────────────────────────────────────────────────────
class Level(IntEnum):
"""In-stream repetition severity (Detector B)."""
NONE = 0
L1 = 1 # abort, keep prefix, inject cut-note, continue
L2 = 2 # abort, discard turn, retry once with corrective summary
L3 = 3 # kill → fresh conversation
class ActionSignal(IntEnum):
"""Action-level repetition severity (Detector A)."""
NONE = 0
NOTE = 1 # corrective note injected
KILL = 2 # kill → fresh conversation
# ── Intervention texts ─────────────────────────────────────────────
NOTE_ACTION_REPEAT = (
"You have repeated the same action with identical input {n} times "
"in a row. Use the result from the previous execution or change "
"your approach."
)
NOTE_STREAM_L1 = (
"[GENERATION ABORTED: the same 50-token segment repeated {streak} "
"times. Here is what you produced before the abort. Do not repeat "
"it — continue with a different approach.]"
)
NOTE_STREAM_L2 = (
"Your previous generation was discarded because it repeated the "
"same content extensively ({streak} repeats). Think of a different "
"approach. Previous session memory follows."
)
NOTE_KILL_RESET = (
"The previous conversation was reset because the agent entered a "
"doom loop. Summary of what was already attempted follows. "
"Do not repeat the same actions."
)
NOTE_DOOM_LOOP = (
"Scenario abandoned: the agent entered a doom loop that could not "
"be resolved after {resets} fresh-context resets."
)
# ── Detector A: action-level consecutive identical ─────────────────
def canonical_form(obj):
"""Canonical JSON of a parsed action, excluding 'reasoning'.
Reasoning varies even at temperature 0 and is not an operative
field — it is the model's internal monologue, not the tool input.
This mirrors OpenCode's approach of comparing tool name + input,
not the model's thought process.
"""
stripped = {k: v for k, v in obj.items() if k != "reasoning"}
return json.dumps(stripped, sort_keys=True)
class ActionRepetitionDetector:
"""Tracks consecutive identical actions across turns.
feed(parsed_action) → ActionSignal
3rd consecutive identical → NOTE
4th consecutive identical → KILL
Any different action resets the counter.
"""
NOTE_THRESHOLD = 3
KILL_THRESHOLD = 4
def __init__(self):
self._last_canonical = None
self._count = 0
def feed(self, parsed_action):
canonical = canonical_form(parsed_action)
if canonical == self._last_canonical:
self._count += 1
else:
self._count = 1
self._last_canonical = canonical
if self._count >= self.KILL_THRESHOLD:
return ActionSignal.KILL
if self._count == self.NOTE_THRESHOLD:
return ActionSignal.NOTE
return ActionSignal.NONE
def reset(self):
"""Clear state (used on fresh-context reset)."""
self._last_canonical = None
self._count = 0
@property
def count(self):
return self._count
# ── Detector B: in-stream exact-segment repetition ─────────────────
class StreamRepetitionDetector:
"""Watches the combined reasoning + content stream for exact
50-delta block repetition.
feed(kind, text_delta) → Level
A new instance should be created for each generation call.
The ``level`` attribute holds the highest level triggered.
"""
BLOCK_SIZE = 50
L1_THRESHOLD = 50
L2_THRESHOLD = 500
L3_THRESHOLD = 1500
def __init__(self):
self._buf = []
self._prev_hash = None
self._streak = 0
self.level = Level.NONE
def feed(self, kind, text):
"""Process one delta. kind is 'reasoning' or 'content' (ignored
— both streams are combined in emission order). Returns the
current Level, re-evaluated at each 50-delta block boundary."""
self._buf.append(text)
if len(self._buf) < self.BLOCK_SIZE:
return Level.NONE
block = "".join(self._buf)
self._buf = []
block_hash = hashlib.md5(block.encode()).hexdigest()
if block_hash == self._prev_hash:
self._streak += 1
else:
self._streak = 0
self._prev_hash = block_hash
return self._evaluate()
def _evaluate(self):
if self._streak >= self.L3_THRESHOLD:
self.level = Level.L3
return Level.L3
if self._streak >= self.L2_THRESHOLD:
if self.level < Level.L2:
self.level = Level.L2
return Level.L2
if self._streak >= self.L1_THRESHOLD:
if self.level < Level.L1:
self.level = Level.L1
return Level.L1
return Level.NONE
@property
def streak(self):
return self._streak
# ── Insurance counter ──────────────────────────────────────────────
class InsuranceCounter:
"""Global tripwire: 10 total mid-stream aborts (L1 + L2) in a
scenario → kill → fresh conversation."""
MAX_ABORTS = 10
def __init__(self):
self._aborts = 0
def register_abort(self):
"""Call after any L1 or L2 abort. Returns True if the
insurance tripwire fired (kill → fresh conversation)."""
self._aborts += 1
return self._aborts >= self.MAX_ABORTS
@property
def count(self):
return self._aborts
# ── Reset cap ──────────────────────────────────────────────────────
class ResetCap:
"""Caps fresh-context resets at 3 per scenario. The 4th doom
→ abandon scenario with outcome='doom_loop'."""
MAX_RESETS = 3
def __init__(self):
self._resets = 0
def register_reset(self):
"""Call on each fresh-context reset. Returns True if the
cap is exceeded (scenario should be abandoned)."""
self._resets += 1
return self._resets > self.MAX_RESETS
@property
def count(self):
return self._resets
@property
def remaining(self):
return max(0, self.MAX_RESETS - self._resets)

67
zubble/env.py Normal file
View file

@ -0,0 +1,67 @@
#!/usr/bin/env python3
"""env — environment abstractions for the zubble agent loop.
An Environment provides the surface the agent loop needs to interact
with the world:
intro str — shell/environment description shown to the model
report_path str — where the agent writes its deliverable
exec(cmd, timeout) -> (exit_code, output)
verify() -> (exit_code, output) # oracle / grader
LocalEnv is the simplest concrete Environment: each ``exec()`` call
runs in a fresh subprocess on the local machine. The scenario supplies
``intro``, ``report_path``, and a ``verify_fn`` callable at construction
time. This is the building block for container-based scenarios — a
container variant would override ``exec`` to shell into the container.
"""
import subprocess
class LocalEnv:
"""Execute commands via local subprocess, stateless between calls.
Parameters
----------
intro : str
Description of the environment shown to the model as the first
line of the opening message.
report_path : str
File path the agent is told to write its deliverable to.
verify_fn : callable, optional
``verify_fn() -> (exit_code, output)``. If omitted, verify()
always returns ``(0, "accepted")`` — useful for tests.
"""
def __init__(self, intro="You have a shell on this machine.",
report_path="/work/output.md", verify_fn=None):
self.intro = intro
self.report_path = report_path
self._verify_fn = verify_fn
def exec(self, cmd, timeout=90):
"""Run *cmd* in a fresh subprocess shell.
Returns (exit_code, combined_output). Output is truncated to
8000 chars to match the scenariobench convention.
"""
try:
r = subprocess.run(
cmd, shell=True, capture_output=True, text=True,
timeout=timeout,
)
out = (r.stdout or "") + (r.stderr or "")
return r.returncode, out[:8000]
except subprocess.TimeoutExpired:
return 124, f"[command timed out after {timeout}s]"
except Exception as e:
return 1, f"[exec error: {e}]"
def verify(self):
"""Call the scenario's verification oracle.
Returns (exit_code, output). Exit code 0 means accepted.
"""
if self._verify_fn:
return self._verify_fn()
return (0, "accepted")

245
zubble/ledger.py Normal file
View file

@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""ledger — plan-state machine and short-term memory for the zubble agent loop.
The ledger is the model's short-term memory. It tracks not just what was
attempted, but the full causal chain: what the model planned to do, which
steps are done vs pending, where execution diverged from the plan, and
why previous approaches were abandoned with causal failure reasons.
This prevents two failure modes that a simple command log cannot:
1. Plan divergence — the model creates a multi-step plan, executes steps
1-2, then at step 3 does something completely different (e.g., creates
a manifest in step 2 but doesn't apply it in step 3, tries to delete
the node instead). The ledger tracks the plan as a state machine and
detects when an action doesn't match the next pending step.
2. Approach cycling — the model retries the same failed approach 10 turns
later because it forgot why it failed. The ledger stores abandoned
approaches with causal failure reasons, not just "tried X, failed."
"""
import json
import time
class Ledger:
"""Plan state machine + session memory.
Lifecycle:
ledger = Ledger()
ledger.set_plan(["diagnose", "create manifest", "apply manifest", "verify"])
ledger.record_action(plan_step_id=1, command="kubectl get certs",
exit_code=0, output="...", outcome="found expired cert")
ledger.record_action(plan_step_id=2, command="kubectl create -f cert-fix.yaml",
exit_code=0, output="...", outcome="manifest created")
divergence = ledger.check_divergence(plan_step_id=3, command="kubectl delete node-b")
# divergence = "expected: apply manifest (step 3), got: kubectl delete node-b"
ledger.record_abandoned("restart kubelet",
reason="kubelet was already running; cert expiry is root cause")
summary = ledger.render_summary() # inject into every prompt
"""
def __init__(self):
self.plan_steps = [] # [{"id": 1, "description": "...", "status": "pending|done|failed|skipped", "outcome": "..."}]
self.attempted = [] # [{"turn": N, "action": "run", "command": "...", "plan_step_id": N, "exit_code": N, "outcome": "...", "ts": float}]
self.abandoned = [] # [{"approach": "restart kubelet", "reason": "kubelet already running; cert expiry is root cause", "turn": N}]
self.divergence_log = [] # [{"turn": N, "expected": "step 3: apply manifest", "got": "kubectl delete node-b", "resolved": False}]
self.turn = 0
# ── Plan management ────────────────────────────────────────────
def set_plan(self, steps):
"""Capture the plan from the model's first turn."""
self.plan_steps = [
{"id": i + 1, "description": desc, "status": "pending", "outcome": None}
for i, desc in enumerate(steps)
]
def revise_plan(self, new_steps, reason):
"""Replace the current plan when the model abandons it."""
for step in self.plan_steps:
if step["status"] == "pending":
self.record_abandoned(step["description"],
reason=f"plan revised: {reason}")
self.plan_steps = [
{"id": i + 1, "description": desc, "status": "pending", "outcome": None}
for i, desc in enumerate(new_steps)
]
def mark_step_done(self, step_id, outcome):
for step in self.plan_steps:
if step["id"] == step_id:
step["status"] = "done"
step["outcome"] = outcome
return
def mark_step_failed(self, step_id, reason):
for step in self.plan_steps:
if step["id"] == step_id:
step["status"] = "failed"
step["outcome"] = reason
return
def next_pending_step(self):
for step in self.plan_steps:
if step["status"] == "pending":
return step
return None
# ── Action recording ───────────────────────────────────────────
def record_action(self, plan_step_id, command, exit_code, output, outcome=""):
self.turn += 1
entry = {
"turn": self.turn,
"plan_step_id": plan_step_id,
"command": command,
"exit_code": exit_code,
"outcome": outcome or self._summarize_output(exit_code, output),
"ts": time.time(),
}
self.attempted.append(entry)
if plan_step_id is not None:
if exit_code == 0:
self.mark_step_done(plan_step_id, entry["outcome"])
# Don't mark failed on non-zero exit — the model may retry
# the same step with a different command. Only mark failed
# when the model explicitly abandons the approach.
def record_abandoned(self, approach, reason, turn=None):
self.abandoned.append({
"approach": approach,
"reason": reason,
"turn": turn or self.turn,
})
# ── Divergence detection ───────────────────────────────────────
def check_divergence(self, plan_step_id, command):
"""Check if an action matches the next pending plan step.
Returns a warning string if divergent, None if aligned.
"""
if not self.plan_steps:
return None
if plan_step_id is None:
return (
f"Your action does not reference any plan step. "
f"Your next planned step is: {self._step_desc(self.next_pending_step())}."
)
pending = self.next_pending_step()
if pending is None:
return None
if plan_step_id == pending["id"]:
return None
expected = self._step_desc(pending)
got = command or "(no command)"
# Check if the step the model referenced is already done
referenced = None
for step in self.plan_steps:
if step["id"] == plan_step_id:
referenced = step
break
if referenced and referenced["status"] == "done":
warning = (
f"WARNING: Plan step {plan_step_id} ('{referenced['description']}') "
f"is already DONE. Your next pending step is step {pending['id']} "
f"('{pending['description']}'). You may be repeating completed work."
)
else:
warning = (
f"WARNING: Your action '{got}' references plan step {plan_step_id}, "
f"but your next pending step is step {pending['id']} "
f"('{pending['description']}'). Either execute step {pending['id']} "
f"as planned, or emit a revised_plan explaining why you're "
f"abandoning this approach."
)
self.divergence_log.append({
"turn": self.turn + 1,
"expected": expected,
"got": got,
"resolved": False,
})
return warning
# ── Summary rendering (injected into every prompt) ─────────────
def render_summary(self):
"""Render the session memory block for prompt injection.
This is what the model sees at the top of every turn — its plan,
progress, abandoned approaches, and any divergence warnings.
"""
lines = ["## SESSION MEMORY"]
# Plan status
if self.plan_steps:
plan_parts = []
for step in self.plan_steps:
mark = {"pending": "→", "done": "✓", "failed": "✗", "skipped": "⊘"}.get(
step["status"], "?"
)
plan_parts.append(f"[{step['id']}: {step['description']} {mark}]")
lines.append("Plan: " + " ".join(plan_parts))
else:
lines.append("Plan: (not yet created — emit action='plan' first)")
# Last action
if self.attempted:
last = self.attempted[-1]
lines.append(
f"Last action: {last['command']} "
f"(step {last['plan_step_id']}, exit={last['exit_code']}, "
f"{last['outcome'][:80]})"
)
# Next expected
pending = self.next_pending_step()
if pending:
lines.append(f"Next expected: Step {pending['id']} — {pending['description']}")
# Abandoned approaches
if self.abandoned:
lines.append("Abandoned approaches:")
for a in self.abandoned[-5:]: # last 5
lines.append(f" - {a['approach']} (failed: {a['reason']})")
# Unresolved divergence
unresolved = [d for d in self.divergence_log if not d["resolved"]]
if unresolved:
last_div = unresolved[-1]
lines.append(
f"⚠ DIVERGENCE: expected '{last_div['expected']}', "
f"got '{last_div['got']}'. Resolve this before proceeding."
)
return "\n".join(lines)
# ── Serialization ──────────────────────────────────────────────
def to_dict(self):
return {
"plan_steps": self.plan_steps,
"attempted": self.attempted,
"abandoned": self.abandoned,
"divergence_log": self.divergence_log,
"turn": self.turn,
}
def to_json(self):
return json.dumps(self.to_dict(), indent=2)
# ── Internal helpers ───────────────────────────────────────────
@staticmethod
def _step_desc(step):
if step is None:
return "(no pending steps — all done)"
return f"step {step['id']}: {step['description']}"
@staticmethod
def _summarize_output(exit_code, output):
if not output:
return f"exit={exit_code}, no output"
first_line = output.strip().split("\n")[0][:100]
return f"exit={exit_code}, {first_line}"

279
zubble/protocol.py Normal file
View file

@ -0,0 +1,279 @@
#!/usr/bin/env python3
"""protocol — structured JSON protocol for the zubble agent loop.
Replaces scenariobench's flat-text "User:/Assistant:" rendering with a
typed JSON contract. The model emits structured actions (plan, run,
submit) instead of free-form text, which prevents the
conversation-simulation loops that caused the 7,303-repetition case.
JSON enforcement combines constrained decoding, strict parsing, and one
targeted repair generation when a response is malformed.
Scenario-specific text (role, framing, deliverable wording, verifier
feedback) is encapsulated in the ``Protocol`` dataclass. The JSON
schema, action enum, parser, and repair nudge are shared across all
protocols and remain module-level.
"""
import json
from dataclasses import dataclass
# ── Schema ──────────────────────────────────────────────────────────
ACTIONS = ("plan", "run", "report", "submit", "ask")
# The endpoint constrains syntax and the action enum; _validate keeps the
# action-specific field requirements because this compact schema is shared by
# all action shapes.
RESPONSE_FORMAT = {
"type": "json_schema",
"json_schema": {
"name": "zubble_action",
"schema": {
"type": "object",
"properties": {
"action": {"type": "string", "enum": list(ACTIONS)},
"command": {"type": "string"},
"plan_step_id": {"type": "integer"},
"reasoning": {"type": "string"},
"steps": {"type": "array", "items": {"type": "string"}},
"report": {"type": "string"},
"question": {"type": "string"},
"revised_plan": {"type": "array", "items": {"type": "string"}},
},
"required": ["action", "reasoning"],
"additionalProperties": False,
},
},
}
# ── Protocol: scenario-specific framing ─────────────────────────────
_SYSTEM_PROMPT_TEMPLATE = """\
You are {role}.
You communicate ONLY in structured JSON. Every response must be a single \
JSON object — no prose before or after, no markdown fences, no commentary.
The JSON schema:
{{"action": "plan" | "run" | "report" | "submit" | "ask",
"command": "<shell command>", // required for action="run"
"plan_step_id": <integer>, // required for action="run": which plan step
"reasoning": "<brief explanation>", // always required
"steps": ["<step 1>", "<step 2>", ...], // required for action="plan"
"report": "<short summary>", // required for action="report" or "submit"
"question": "<clarification>", // required for action="ask"
"revised_plan": ["<new steps>"]}} // when abandoning the current plan
Rules:
1. Turn 1: emit {{"action": "plan", "steps": [...], "reasoning": "..."}}.
List every step you intend to take, in order.
2. Turns 2+: emit {{"action": "run", "command": "...", "plan_step_id": N, \
"reasoning": "..."}} where N is the step number you are executing.
3. When all steps are done: write the full {deliverable} to the required file,
then emit {{"action": "submit", "report": "<one-line summary>"}}.
The report value must be a short, single-line JSON string; do not put
markdown or literal newlines in it.
4. If you need to abandon your plan: emit {{"action": "run", \
"revised_plan": ["<new steps>"], "plan_step_id": 1, "command": "...", \
"reasoning": "abandoning previous plan because ..."}}.
5. If you need clarification: emit {{"action": "ask", "question": "..."}}.
6. A rejected submission is not terminal. Keep working and submit again.
The harness, not you, decides when the scenario ends.
7. NEVER emit raw text, markdown, or conversation turns. ONLY JSON.
"""
@dataclass
class Protocol:
"""Scenario-specific framing for the structured agent loop.
Holds only the text that varies between task families: the agent's
role description, the opening framing, the deliverable noun, and the
wording of verifier feedback. The JSON schema, action enum, parser,
and repair nudge are shared and stay module-level.
Create a custom Protocol for a new scenario::
spongebob = Protocol(
role="an autonomous business agent for the Chum Bucket restaurant",
framing="Your competitor is running a successful online storefront.",
deliverable="new menu item",
report_path="/work/new-menu.md",
submit_accepted="Menu item received. Thank you.",
submit_rejected="That submission is incomplete. Keep working.",
)
"""
role: str = "a Kubernetes repair agent working inside a time-boxed scenario"
framing: str = (
"A PagerDuty alert has fired against a critical Kubernetes cluster. "
"Identify the issue, fix it, and validate the fix."
)
deliverable: str = "post-mortem"
open_instruction: str = "Begin by emitting a plan."
submit_accepted: str = "The verifier accepted your fix. Scenario complete."
submit_rejected: str = (
"The verifier reports the cluster is still unhealthy. "
"Keep working and submit again."
)
ask_response: str = (
"No clarification is available. Diagnose the issue from the "
"shell output you have and proceed with your best approach."
)
time_nudge: str = (
"Time is nearly up. If you have a fix, write your report and submit now."
)
def system_prompt(self) -> str:
"""The full system prompt: role line + generic JSON schema and rules."""
return _SYSTEM_PROMPT_TEMPLATE.format(
role=self.role, deliverable=self.deliverable
)
def initial_message(self, env_intro, brief, login, report_path) -> str:
"""The opening user message that starts the agent loop."""
return (
f"{env_intro}\n\n"
f"{self.framing}\n\n"
f"## Brief\n{brief}\n\n"
f"## Login\n{login}\n\n"
f"Write your {self.deliverable} to {report_path} when done.\n"
f"{self.open_instruction}"
)
def reset_message(self, env_intro, brief, login, report_path,
kill_note, ledger_summary) -> str:
"""The fresh-context user message after a doom-loop kill."""
return (
f"{env_intro}\n\n"
f"## Brief\n{brief}\n\n"
f"## Login\n{login}\n\n"
f"Write your {self.deliverable} to {report_path} when done.\n\n"
f"{kill_note}\n\n"
f"{ledger_summary}"
)
def render_action_result(self, action, exit_code, output, note="") -> str:
"""Render the harness response to a model action, as user-message text."""
parts = []
if action == "run":
parts.append(f"exit={exit_code}")
parts.append(f"```\n{output[:4000]}\n```")
elif action == "submit":
parts.append(
self.submit_rejected if exit_code != 0 else self.submit_accepted
)
elif action == "ask":
parts.append(self.ask_response)
if note:
parts.append(note)
return "\n".join(parts)
def retry_nudge(self, error):
"""Return a repair request for malformed JSON, using generic wording."""
return (
f"Your last response was rejected: {error}. Respond once with a single "
"valid JSON object and no prose, markdown, or fences. Use only the "
f"supported actions: {', '.join(ACTIONS)}. If submitting, keep report "
f"to one short line; the full {self.deliverable} belongs in the report file."
)
# Backward-compatible default: the original Kubernetes repair scenario.
REPAIR = Protocol()
# Module-level aliases for backward compatibility.
SYSTEM_PROMPT = REPAIR.system_prompt()
# ── Parse + retry ──────────────────────────────────────────────────
_RETRIES = 1
class ProtocolError(Exception):
"""Raised when the model's response cannot be parsed as valid JSON."""
def __init__(self, message, raw):
self.raw = raw
super().__init__(message)
def parse_response(text):
"""Parse a model response into a dict.
A response must be one JSON object with no surrounding prose. ``raw_decode``
gives a precise error for trailing content without ambiguous regex recovery.
"""
raw = text.strip()
try:
obj, end = json.JSONDecoder().raw_decode(raw)
except json.JSONDecodeError as e:
raise ProtocolError(f"invalid JSON: {e.msg} at character {e.pos}", raw) from e
if raw[end:].strip():
raise ProtocolError("trailing content after JSON object", raw)
return _validate(obj, raw)
def _validate(obj, raw):
"""Check required fields and action-specific constraints."""
if not isinstance(obj, dict):
raise ProtocolError(f"expected JSON object, got {type(obj).__name__}", raw)
action = obj.get("action")
if action not in ACTIONS:
if action in ("final", "finished"):
raise ProtocolError(
f"action={action!r} is not supported; use 'submit'", raw
)
raise ProtocolError(
f"action must be one of {ACTIONS}, got {action!r}", raw
)
if "reasoning" not in obj:
raise ProtocolError(f"missing required field 'reasoning'", raw)
if action == "plan" and "steps" not in obj:
raise ProtocolError("action='plan' requires 'steps' field", raw)
if action == "run" and "command" not in obj:
raise ProtocolError("action='run' requires 'command' field", raw)
if action in ("report", "submit") and "report" not in obj:
raise ProtocolError(f"action='{action}' requires 'report' field", raw)
if action == "ask" and "question" not in obj:
raise ProtocolError("action='ask' requires 'question' field", raw)
return obj
def retry_nudge(error):
"""Return a bounded, action-oriented repair request for malformed JSON.
Delegates to the default REPAIR protocol for backward compatibility.
New code should call ``proto.retry_nudge(error)`` instead.
"""
return REPAIR.retry_nudge(error)
def parse_with_retry(text, gen_fn, max_retries=_RETRIES):
"""Parse a response, retrying once on failure.
gen_fn(retry_prompt) -> (text, reasoning, perf) is called on retry.
Returns (parsed_dict, raw_text, perf, retries_used).
"""
try:
obj = parse_response(text)
return obj, text, None, 0
except ProtocolError as e:
if max_retries <= 0:
raise
retry_text, retry_reasoning, retry_perf = gen_fn(retry_nudge(e))
try:
obj = parse_response(retry_text)
return obj, retry_text, retry_perf, 1
except ProtocolError as e2:
raise ProtocolError(
f"retry also failed: {e2}", retry_text
) from e
# ── Rendering ──────────────────────────────────────────────────────
def render_action_result(action, exit_code, output, note=""):
"""Render the harness response to a model action, as user-message text.
Delegates to the default REPAIR protocol for backward compatibility.
New code should call ``proto.render_action_result(...)`` instead.
"""
return REPAIR.render_action_result(action, exit_code, output, note=note)