Examples

Examples

Coding Loop

Implement one bounded feature or bug fix with full verification.

from zloop import LoopEngine, JsonlMemoryStore, Budgets, State

class CodingAdapter:
    def run(self, role, state):
        handlers = {
            "discoverer": self._discover,
            "planner": self._plan,
            "executor": self._execute,
            "verifier": self._verify,
            "reviewer": self._review,
            "repairer": self._repair,
        }
        return handlers.get(role, self._default)(state)

    def _discover(self, state):
        return AgentResult(status="OK", summary="Found 3 modules")

    def _plan(self, state):
        return AgentResult(status="OK", summary="Plan: add validation")

    def _execute(self, state):
        return AgentResult(status="OK", summary="Implemented", artifacts=["src/handler.py"])

    def _verify(self, state):
        return AgentResult(status="OK", summary="Tests pass", verification_passed=True)

    def _review(self, state):
        return AgentResult(status="OK", summary="Review clean", blocking_review_findings=False)

    def _repair(self, state):
        return AgentResult(status="OK", summary="Fixed issue", progress=True)

budgets = Budgets(max_iterations=10, max_repairs=3)
engine = LoopEngine(CodingAdapter(), JsonlMemoryStore())
result = engine.run(
    goal="Add input validation",
    acceptance_criteria=["tests pass", "lint clean"],
    budgets=budgets,
)

Research Loop

Produce an evidence-backed synthesis with verified claims.

budgets = Budgets(max_iterations=8, token_budget=100_000)
engine = LoopEngine(ResearchAdapter(), JsonlMemoryStore())
result = engine.run(
    goal="Synthesize findings on topic X",
    acceptance_criteria=[
        "all claims have traceable evidence",
        "conflicts explicitly disclosed",
    ],
    budgets=budgets,
)

Content Loop

Create publishable content against a defined rubric.

budgets = Budgets(max_iterations=6, max_repairs=3)
engine = LoopEngine(ContentAdapter(), JsonlMemoryStore())
result = engine.run(
    goal="Write technical blog post",
    acceptance_criteria=[
        "addresses stated objective",
        "tone matches audience",
        "meets length requirement",
    ],
    budgets=budgets,
)

Fleet Loop

Coordinate multiple specialist agents working in parallel.

class FleetOrchestrator:
    def create_worktree(self, agent_id):
        subprocess.run(["git", "worktree", "add", f".worktrees/{agent_id}", "-b", f"agent/{agent_id}"])

    def run_specialist(self, agent_id, adapter, goal, criteria, budgets):
        worktree = self.create_worktree(agent_id)
        os.chdir(worktree)
        engine = LoopEngine(adapter, JsonlMemoryStore(".zloop/memory.jsonl"))
        return engine.run(goal, criteria, budgets)

fleet = FleetOrchestrator()
research = fleet.run_specialist("research-001", ResearchAdapter(), "Research topic", [...], budgets)
engineer = fleet.run_specialist("engineer-001", EngineerAdapter(), "Implement feature", [...], budgets)
qa = fleet.run_specialist("qa-001", QAAdapter(), "Verify implementation", [...], budgets)