-
Notifications
You must be signed in to change notification settings - Fork 0
API Reference
cvsz edited this page Aug 31, 2026
·
1 revision
class LoopEngine:
def __init__(
self,
adapter: AgentAdapter,
memory: MemoryStore,
*,
enable_logging: bool = True,
) -> None: ...
def run(
self,
goal: str,
acceptance_criteria: list[str],
budgets: Budgets | None = None,
) -> LoopState: ...@dataclass
class LoopState:
loop_id: str
goal: str
acceptance_criteria: list[str]
budgets: Budgets
state: State = State.DISCOVER
iteration: int = 0
repair_attempts: int = 0
consecutive_no_progress: int = 0
usage: Usage = field(default_factory=Usage)
evidence: list[str] = field(default_factory=list)
blockers: list[str] = field(default_factory=list)
history: list[dict[str, Any]] = field(default_factory=list)
started_at: float = field(default_factory=time.time)@dataclass
class AgentResult:
status: str # "OK", "FAIL", "BLOCKED", "INCONCLUSIVE"
summary: str
evidence: list[str] = field(default_factory=list)
next_action: str | None = None
risks: list[str] = field(default_factory=list)
artifacts: list[str] = field(default_factory=list)
memory_updates: list[str] = field(default_factory=list)
usage: Usage = field(default_factory=Usage)
progress: bool = True
verification_passed: bool = False
blocking_review_findings: bool = False@dataclass
class Budgets:
max_iterations: int = 12
max_repairs: int = 4
max_consecutive_no_progress: int = 2
token_budget: int = 500_000
cost_budget: float = 25.0
wall_clock_budget_seconds: int = 7200@dataclass
class Usage:
tokens: int = 0
cost: float = 0.0class State(str, Enum):
DISCOVER = "DISCOVER"
PLAN = "PLAN"
EXECUTE = "EXECUTE"
VERIFY = "VERIFY"
REVIEW = "REVIEW"
REPAIR = "REPAIR"
SHIPPED = "SHIPPED"
HANDOFF = "HANDOFF"
BUDGET_EXHAUSTED = "BUDGET_EXHAUSTED"
POLICY_VIOLATION = "POLICY_VIOLATION"
FAILED = "FAILED"class JsonlMemoryStore:
def __init__(self, path: str = ".zloop/memory.jsonl", validate: bool = True) -> None: ...
def save(self, state: LoopState) -> None: ...class AgentAdapter(Protocol):
def run(self, role: str, state: LoopState) -> AgentResult: ...
class MemoryStore(Protocol):
def save(self, state: LoopState) -> None: ...from validation import validate_loop_state, is_valid_loop_state
from validation import validate_agent_result, is_valid_agent_result
from validation import validate_verification_result