diff --git a/config/production.example.yaml b/config/production.example.yaml new file mode 100644 index 0000000..88ba506 --- /dev/null +++ b/config/production.example.yaml @@ -0,0 +1,29 @@ +loop: + mode: closed + max_iterations: 12 + max_repairs: 4 + max_consecutive_no_progress: 2 + token_budget: 500000 + cost_budget: 25.00 + wall_clock_budget_seconds: 7200 +fleet: + max_depth: 2 + max_fanout: 4 + max_parallel_mutators: 2 +permissions: + default: READ_ONLY + require_approval: [REMOTE_MUTATION, HIGH_IMPACT] +models: + discoverer: {primary: cheap, fallbacks: [strong]} + planner: {primary: cheap, fallbacks: [strong]} + executor: {primary: strong, fallbacks: []} + verifier: {primary: strong, fallbacks: []} +persistence: + state_backend: postgresql + idempotency_backend: postgresql + artifacts: object-storage + audit: append-only +observability: + traces: otlp + metrics: enabled + secret_redaction: true diff --git a/docs/ARCHITECTURE-PRODUCTION.md b/docs/ARCHITECTURE-PRODUCTION.md new file mode 100644 index 0000000..80753f8 --- /dev/null +++ b/docs/ARCHITECTURE-PRODUCTION.md @@ -0,0 +1,23 @@ +# ZLoop Production Architecture + +## Control plane +Mission API -> Orchestrator -> Policy Engine + Budget Governor + Model Router. + +## Data plane +Isolated Executors -> Connector Adapters -> Independent Verifiers -> Repair Planner. + +## Persistence +PostgreSQL for canonical loop/checkpoint/idempotency/approval state; durable queue for worker leases/retries/cancellation; object storage for artifacts; append-only audit store for mutation events. + +## Security invariants +- deny mutation by default; +- tenant/actor/action/target binding on approvals; +- no client-side provider secrets; +- executor cannot directly mark `SHIPPED`; +- stale revision/SHA blocks mutation; +- credentials are scoped per adapter and never persisted in loop memory; +- external side effects require durable idempotency keys; +- `INCONCLUSIVE` is never success. + +## Fleet constraints +Children receive strict subsets of parent scope and remaining budget. Depth, fan-out and concurrency caps are mandatory. Mutating children use isolated workspaces. Final acceptance belongs to an independent verifier. diff --git a/docs/PRODUCTION-READINESS.md b/docs/PRODUCTION-READINESS.md new file mode 100644 index 0000000..b1513b1 --- /dev/null +++ b/docs/PRODUCTION-READINESS.md @@ -0,0 +1,44 @@ +# ZLoop Production Readiness + +ZLoop is considered **production-grade-ready** only when every gate below has implementation evidence, automated tests, and operational ownership. This document distinguishes implemented foundations from deployment-dependent controls. + +## Implemented foundation + +- bounded loop lifecycle and terminal states; +- independent verifier/reviewer contracts; +- durable SQLite checkpoint and idempotency ledger; +- durable SQLite reference queue with leases, lease-expiry reclaim, retry, cancellation and deadlines; +- checkpoint serialization/restoration helpers; +- legal state-transition validation; +- provider-neutral model routing contract and bounded fallback; +- hard global and per-profile token/cost limits with fail-closed behavior; +- structured-output validation; +- deny-by-default permission gate; +- verifier registry with fail-closed `INCONCLUSIVE` behavior; +- evidence fingerprints for repeat/no-progress detection; +- bounded fleet depth/fan-out governor; +- permission-bound command runner with timeout and reduced environment; +- GitHub PR safety boundary with stale-SHA, approval and check gates; +- audit event contract with tenant field; +- metrics/tracing-compatible span primitives; +- secret-redaction helper; +- CI, CodeQL, Dependency Review, Dependabot, concurrency and timeouts. + +## Mandatory deployment gates before production mutation + +1. PostgreSQL durable state/idempotency implementation and migration tests. +2. Production queue/worker backend with distributed leases and crash-recovery integration tests; SQLite is the local/reference backend only. +3. OS/container sandbox with CPU, memory, process and network limits; the reference command runner is not a security sandbox. +4. Capability-scoped real connector adapters with tenant/actor/action/target binding. +5. External approval authority for `REMOTE_MUTATION` and `HIGH_IMPACT` operations. +6. Durable append-only audit sink and retention policy. +7. OpenTelemetry exporter configuration, SLO dashboards and alerting. +8. Secrets-manager integration and production redaction tests. +9. Branch protection with required CI/security/review checks. +10. SBOM, provenance/attestation, signed release process and rollback runbook. +11. PostgreSQL backups, restore drills, RPO/RTO and disaster-recovery ownership. +12. Load, concurrency and chaos tests against the actual deployment topology. + +## Release criteria + +No production release may be marked ready unless all required checks have explicit passing evidence, no blocking review finding remains, mutating connectors are approval-bound, idempotency survives restarts, state recovery is crash-tested, budgets fail closed, `INCONCLUSIVE` never ships, secrets are absent from artifacts/logs, and rollback/handoff procedures are tested. diff --git a/docs/RELEASE-GATES.md b/docs/RELEASE-GATES.md new file mode 100644 index 0000000..41b65a9 --- /dev/null +++ b/docs/RELEASE-GATES.md @@ -0,0 +1,30 @@ +# Release Gates + +## Automated +- unit tests +- compile/static syntax checks +- CodeQL/security analysis +- dependency review +- secret-pattern scan +- package/build validation +- persistence migration tests when schemas change + +## Engineering +- maker/checker separation +- blocking findings resolved +- acceptance criteria mapped to evidence +- backward compatibility assessed +- rollback path documented +- mutations covered by permission + idempotency tests + +## Production-only +- required branch checks configured +- release provenance verified +- secrets-manager integration verified +- audit retention configured +- dashboards/alerts enabled +- backup restore drill passed +- capacity/load test passed +- operator runbook reviewed + +No missing or `INCONCLUSIVE` mandatory gate may be treated as PASS. diff --git a/schemas/audit-event.schema.json b/schemas/audit-event.schema.json new file mode 100644 index 0000000..95b3f77 --- /dev/null +++ b/schemas/audit-event.schema.json @@ -0,0 +1,17 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "ZLoop Audit Event", + "type": "object", + "required": ["actor", "action", "target", "permission", "result", "created_at"], + "properties": { + "actor": {"type": "string", "minLength": 1}, + "tenant_id": {"type": ["string", "null"]}, + "action": {"type": "string", "minLength": 1}, + "target": {"type": "string", "minLength": 1}, + "permission": {"enum": ["READ_ONLY", "LOCAL_MUTATION", "REMOTE_MUTATION", "HIGH_IMPACT"]}, + "result": {"type": "string", "minLength": 1}, + "idempotency_key": {"type": ["string", "null"]}, + "created_at": {"type": "number"} + }, + "additionalProperties": false +} diff --git a/src/execution.py b/src/execution.py new file mode 100644 index 0000000..9dc32c4 --- /dev/null +++ b/src/execution.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Mapping, Sequence +import os +import subprocess + +from .runtime_contracts import Permission, PermissionGate + + +@dataclass(frozen=True) +class CommandResult: + returncode: int + stdout: str + stderr: str + + +class CommandRunner: + def __init__(self, workspace: str, permission_gate: PermissionGate, timeout_seconds: int = 120) -> None: + self.workspace = Path(workspace).resolve() + self.permission_gate = permission_gate + self.timeout_seconds = timeout_seconds + + def run(self, command: Sequence[str], *, env: Mapping[str, str] | None = None) -> CommandResult: + self.permission_gate.require(Permission.LOCAL_MUTATION) + if not self.workspace.exists() or not self.workspace.is_dir(): + raise ValueError("workspace must exist") + safe_env = {"PATH": os.environ.get("PATH", ""), "LANG": "C.UTF-8"} + if env: + safe_env.update(env) + completed = subprocess.run( + list(command), cwd=self.workspace, env=safe_env, + text=True, capture_output=True, timeout=self.timeout_seconds, check=False, + ) + return CommandResult(completed.returncode, completed.stdout, completed.stderr) diff --git a/src/github_loop.py b/src/github_loop.py new file mode 100644 index 0000000..4a2cf83 --- /dev/null +++ b/src/github_loop.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Protocol, Sequence + +from .runtime_contracts import Permission, PermissionGate + + +@dataclass(frozen=True) +class PullRequestSnapshot: + number: int + head_sha: str + base_branch: str + required_checks: Sequence[str] + + +class GitHubAdapter(Protocol): + def get_pull_request(self, number: int) -> PullRequestSnapshot: ... + def push_branch(self, branch: str, expected_head_sha: str) -> str: ... + def merge_pull_request(self, number: int, expected_head_sha: str) -> str: ... + + +class SafeGitHubLoop: + def __init__(self, adapter: GitHubAdapter, permissions: PermissionGate) -> None: + self.adapter = adapter + self.permissions = permissions + + def assert_fresh(self, number: int, expected_head_sha: str) -> PullRequestSnapshot: + snapshot = self.adapter.get_pull_request(number) + if snapshot.head_sha != expected_head_sha: + raise RuntimeError("stale pull request head SHA") + return snapshot + + def merge(self, number: int, expected_head_sha: str, checks_passed: bool, approved: bool) -> str: + self.permissions.require(Permission.HIGH_IMPACT) + self.assert_fresh(number, expected_head_sha) + if not checks_passed or not approved: + raise RuntimeError("merge gate not satisfied") + return self.adapter.merge_pull_request(number, expected_head_sha) diff --git a/src/job_queue.py b/src/job_queue.py new file mode 100644 index 0000000..96bb060 --- /dev/null +++ b/src/job_queue.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any +import json +import sqlite3 +import time +import uuid + + +@dataclass(frozen=True) +class Job: + job_id: str + payload: dict[str, Any] + status: str + attempts: int + lease_owner: str | None + lease_until: float | None + deadline: float | None + + +class SQLiteJobQueue: + """Durable local queue with leases, reclaim, retry, cancellation and deadlines.""" + + def __init__(self, path: str) -> None: + self.path = path + Path(path).parent.mkdir(parents=True, exist_ok=True) + with self._connect() as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS jobs ( + job_id TEXT PRIMARY KEY, + payload_json TEXT NOT NULL, + status TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + available_at REAL NOT NULL, + lease_owner TEXT, + lease_until REAL, + deadline REAL, + updated_at REAL NOT NULL + ) + """ + ) + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.path, timeout=30, isolation_level=None) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=FULL") + return conn + + def enqueue(self, payload: dict[str, Any], *, job_id: str | None = None, deadline: float | None = None) -> str: + jid = job_id or str(uuid.uuid4()) + now = time.time() + with self._connect() as conn: + conn.execute( + "INSERT INTO jobs(job_id,payload_json,status,attempts,available_at,deadline,updated_at) VALUES(?,?,?,0,?,?,?)", + (jid, json.dumps(payload, sort_keys=True), "READY", now, deadline, now), + ) + return jid + + def lease(self, owner: str, *, lease_seconds: int = 60) -> Job | None: + now = time.time() + lease_until = now + lease_seconds + with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") + conn.execute( + "UPDATE jobs SET status='READY', lease_owner=NULL, lease_until=NULL, updated_at=? WHERE status='LEASED' AND lease_until <= ?", + (now, now), + ) + conn.execute( + "UPDATE jobs SET status='EXPIRED', updated_at=? WHERE status IN ('READY','LEASED') AND deadline IS NOT NULL AND deadline <= ?", + (now, now), + ) + row = conn.execute( + "SELECT * FROM jobs WHERE status='READY' AND available_at <= ? ORDER BY available_at, job_id LIMIT 1", + (now,), + ).fetchone() + if row is None: + conn.execute("COMMIT") + return None + conn.execute( + "UPDATE jobs SET status='LEASED', attempts=attempts+1, lease_owner=?, lease_until=?, updated_at=? WHERE job_id=?", + (owner, lease_until, now, row["job_id"]), + ) + conn.execute("COMMIT") + return self.get(row["job_id"]) + + def ack(self, job_id: str, owner: str) -> None: + now = time.time() + with self._connect() as conn: + cur = conn.execute( + "UPDATE jobs SET status='SUCCEEDED', lease_owner=NULL, lease_until=NULL, updated_at=? WHERE job_id=? AND status='LEASED' AND lease_owner=?", + (now, job_id, owner), + ) + if cur.rowcount != 1: + raise RuntimeError("ack rejected: lease ownership mismatch") + + def retry(self, job_id: str, owner: str, *, delay_seconds: float = 0) -> None: + now = time.time() + with self._connect() as conn: + cur = conn.execute( + "UPDATE jobs SET status='READY', available_at=?, lease_owner=NULL, lease_until=NULL, updated_at=? WHERE job_id=? AND status='LEASED' AND lease_owner=?", + (now + delay_seconds, now, job_id, owner), + ) + if cur.rowcount != 1: + raise RuntimeError("retry rejected: lease ownership mismatch") + + def cancel(self, job_id: str) -> None: + now = time.time() + with self._connect() as conn: + cur = conn.execute( + "UPDATE jobs SET status='CANCELLED', lease_owner=NULL, lease_until=NULL, updated_at=? WHERE job_id=? AND status NOT IN ('SUCCEEDED','CANCELLED','EXPIRED')", + (now, job_id), + ) + if cur.rowcount != 1: + current = self.get(job_id) + if current is None: + raise KeyError(job_id) + if current.status == "SUCCEEDED": + raise RuntimeError("cannot cancel succeeded job") + + def get(self, job_id: str) -> Job | None: + with self._connect() as conn: + row = conn.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone() + if row is None: + return None + return Job( + job_id=row["job_id"], payload=json.loads(row["payload_json"]), status=row["status"], + attempts=int(row["attempts"]), lease_owner=row["lease_owner"], lease_until=row["lease_until"], deadline=row["deadline"], + ) diff --git a/src/observability.py b/src/observability.py new file mode 100644 index 0000000..c0cab5f --- /dev/null +++ b/src/observability.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +from collections import defaultdict +from contextlib import contextmanager +from dataclasses import dataclass +from typing import Dict, Iterator +import time + + +@dataclass(frozen=True) +class SpanRecord: + name: str + duration_seconds: float + success: bool + + +class Metrics: + def __init__(self) -> None: + self.counters: Dict[str, int] = defaultdict(int) + self.timings: list[SpanRecord] = [] + + def increment(self, name: str, value: int = 1) -> None: + self.counters[name] += value + + @contextmanager + def span(self, name: str) -> Iterator[None]: + started = time.perf_counter() + success = False + try: + yield + success = True + finally: + self.timings.append(SpanRecord(name, time.perf_counter() - started, success)) diff --git a/src/recovery.py b/src/recovery.py new file mode 100644 index 0000000..cc7e705 --- /dev/null +++ b/src/recovery.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +from dataclasses import asdict +from typing import Any, Dict +import time + +from .loop_engine import Budgets, LoopState, State, Usage + + +LEGAL_TRANSITIONS = { + State.DISCOVER: {State.PLAN, State.HANDOFF, State.FAILED, State.BUDGET_EXHAUSTED, State.POLICY_VIOLATION}, + State.PLAN: {State.EXECUTE, State.HANDOFF, State.FAILED, State.BUDGET_EXHAUSTED, State.POLICY_VIOLATION}, + State.EXECUTE: {State.VERIFY, State.HANDOFF, State.FAILED, State.BUDGET_EXHAUSTED, State.POLICY_VIOLATION}, + State.VERIFY: {State.REVIEW, State.REPAIR, State.HANDOFF, State.BUDGET_EXHAUSTED, State.POLICY_VIOLATION}, + State.REVIEW: {State.SHIPPED, State.REPAIR, State.HANDOFF, State.BUDGET_EXHAUSTED, State.POLICY_VIOLATION}, + State.REPAIR: {State.VERIFY, State.HANDOFF, State.FAILED, State.BUDGET_EXHAUSTED, State.POLICY_VIOLATION}, +} + + +def assert_legal_transition(current: State, next_state: State) -> None: + if current in {State.SHIPPED, State.HANDOFF, State.BUDGET_EXHAUSTED, State.POLICY_VIOLATION, State.FAILED}: + raise ValueError("terminal state cannot transition") + if next_state not in LEGAL_TRANSITIONS[current]: + raise ValueError(f"illegal transition: {current.value} -> {next_state.value}") + + +def serialize_state(state: LoopState) -> Dict[str, Any]: + data = asdict(state) + data["state"] = state.state.value + return data + + +def restore_state(data: Dict[str, Any]) -> LoopState: + try: + return LoopState( + loop_id=data["loop_id"], + goal=data["goal"], + acceptance_criteria=list(data["acceptance_criteria"]), + budgets=Budgets(**data["budgets"]), + state=State(data["state"]), + iteration=int(data.get("iteration", 0)), + repair_attempts=int(data.get("repair_attempts", 0)), + consecutive_no_progress=int(data.get("consecutive_no_progress", 0)), + usage=Usage(**data.get("usage", {})), + evidence=list(data.get("evidence", [])), + blockers=list(data.get("blockers", [])), + history=list(data.get("history", [])), + started_at=float(data.get("started_at", time.time())), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError("invalid checkpoint") from exc + + +def deadline_exceeded(deadline_epoch: float | None) -> bool: + return deadline_epoch is not None and time.time() >= deadline_epoch diff --git a/src/runtime_contracts.py b/src/runtime_contracts.py new file mode 100644 index 0000000..e035080 --- /dev/null +++ b/src/runtime_contracts.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Callable, Dict, Iterable, Mapping, Protocol, Sequence +import hashlib +import json +import time + + +class Permission(str, Enum): + READ_ONLY = "READ_ONLY" + LOCAL_MUTATION = "LOCAL_MUTATION" + REMOTE_MUTATION = "REMOTE_MUTATION" + HIGH_IMPACT = "HIGH_IMPACT" + + +class Verdict(str, Enum): + PASS = "PASS" + FAIL = "FAIL" + INCONCLUSIVE = "INCONCLUSIVE" + + +@dataclass(frozen=True) +class ModelProfile: + role: str + primary: str + fallbacks: Sequence[str] = () + max_tokens: int = 32_000 + max_cost: float = 1.0 + + +@dataclass +class BudgetLedger: + token_limit: int + cost_limit: float + tokens_used: int = 0 + cost_used: float = 0.0 + + def charge(self, *, tokens: int, cost: float) -> None: + if tokens < 0 or cost < 0: + raise ValueError("usage cannot be negative") + if self.tokens_used + tokens > self.token_limit: + raise RuntimeError("token budget exceeded") + if self.cost_used + cost > self.cost_limit: + raise RuntimeError("cost budget exceeded") + self.tokens_used += tokens + self.cost_used += cost + + +class ModelProvider(Protocol): + def invoke(self, model: str, payload: Mapping[str, Any]) -> Mapping[str, Any]: ... + + +class ModelRouter: + def __init__(self, provider: ModelProvider, ledger: BudgetLedger) -> None: + self.provider = provider + self.ledger = ledger + + def run(self, profile: ModelProfile, payload: Mapping[str, Any]) -> Mapping[str, Any]: + last_provider_error: Exception | None = None + for model in (profile.primary, *profile.fallbacks): + try: + result = self.provider.invoke(model, payload) + except Exception as exc: + last_provider_error = exc + continue + + usage = result.get("usage", {}) + tokens = int(usage.get("tokens", 0)) + cost = float(usage.get("cost", 0.0)) + if tokens > profile.max_tokens: + raise RuntimeError("model profile token cap exceeded") + if cost > profile.max_cost: + raise RuntimeError("model profile cost cap exceeded") + self.ledger.charge(tokens=tokens, cost=cost) + return result + + raise RuntimeError("all model routes failed") from last_provider_error + + +def validate_structured_output(value: Mapping[str, Any], required: Iterable[str]) -> None: + missing = [key for key in required if key not in value] + if missing: + raise ValueError(f"missing structured-output fields: {', '.join(missing)}") + + +@dataclass(frozen=True) +class AuditEvent: + actor: str + action: str + target: str + permission: Permission + result: str + idempotency_key: str | None = None + tenant_id: str | None = None + created_at: float = field(default_factory=time.time) + + +class PermissionGate: + def __init__(self, allowed: Iterable[Permission]) -> None: + self.allowed = set(allowed) + + def require(self, permission: Permission) -> None: + if permission not in self.allowed: + raise PermissionError(f"permission denied: {permission.value}") + + +def evidence_fingerprint(items: Iterable[str]) -> str: + canonical = json.dumps(sorted(set(items)), separators=(",", ":"), ensure_ascii=False) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class VerificationCheck: + name: str + run: Callable[[], Verdict] + + +class VerifierRegistry: + def __init__(self) -> None: + self._checks: Dict[str, VerificationCheck] = {} + + def register(self, check: VerificationCheck) -> None: + if check.name in self._checks: + raise ValueError(f"duplicate verifier: {check.name}") + self._checks[check.name] = check + + def evaluate(self, required: Sequence[str]) -> Dict[str, Verdict]: + results: Dict[str, Verdict] = {} + for name in required: + results[name] = Verdict.INCONCLUSIVE if name not in self._checks else self._checks[name].run() + return results + + +class FleetGovernor: + def __init__(self, *, max_depth: int = 2, max_fanout: int = 4) -> None: + self.max_depth = max_depth + self.max_fanout = max_fanout + + def validate_spawn(self, *, depth: int, children: int) -> None: + if depth >= self.max_depth: + raise RuntimeError("fleet max depth reached") + if children > self.max_fanout: + raise RuntimeError("fleet max fanout exceeded") + + +def redact_secrets(text: str, secrets: Iterable[str]) -> str: + output = text + for secret in secrets: + if secret: + output = output.replace(secret, "[REDACTED]") + return output diff --git a/src/test_job_queue.py b/src/test_job_queue.py new file mode 100644 index 0000000..cfa71bb --- /dev/null +++ b/src/test_job_queue.py @@ -0,0 +1,58 @@ +import tempfile +import time +import unittest + +from src.job_queue import SQLiteJobQueue + + +class SQLiteJobQueueTests(unittest.TestCase): + def test_enqueue_lease_ack(self): + with tempfile.TemporaryDirectory() as tmp: + q = SQLiteJobQueue(f"{tmp}/queue.db") + jid = q.enqueue({"task": "x"}, job_id="job-1") + job = q.lease("worker-1", lease_seconds=30) + self.assertEqual(job.job_id, jid) + self.assertEqual(job.status, "LEASED") + self.assertEqual(job.attempts, 1) + q.ack(jid, "worker-1") + self.assertEqual(q.get(jid).status, "SUCCEEDED") + + def test_expired_lease_is_reclaimed(self): + with tempfile.TemporaryDirectory() as tmp: + q = SQLiteJobQueue(f"{tmp}/queue.db") + jid = q.enqueue({"task": "x"}) + q.lease("worker-1", lease_seconds=0) + reclaimed = q.lease("worker-2", lease_seconds=30) + self.assertEqual(reclaimed.job_id, jid) + self.assertEqual(reclaimed.lease_owner, "worker-2") + self.assertEqual(reclaimed.attempts, 2) + + def test_retry_requires_lease_owner(self): + with tempfile.TemporaryDirectory() as tmp: + q = SQLiteJobQueue(f"{tmp}/queue.db") + jid = q.enqueue({"task": "x"}) + q.lease("worker-1") + with self.assertRaises(RuntimeError): q.retry(jid, "worker-2") + q.retry(jid, "worker-1") + self.assertEqual(q.get(jid).status, "READY") + + def test_cancel_and_deadline_fail_closed(self): + with tempfile.TemporaryDirectory() as tmp: + q = SQLiteJobQueue(f"{tmp}/queue.db") + cancelled = q.enqueue({"task": "x"}) + q.cancel(cancelled) + self.assertEqual(q.get(cancelled).status, "CANCELLED") + expired = q.enqueue({"task": "y"}, deadline=time.time() - 1) + self.assertIsNone(q.lease("worker")) + self.assertEqual(q.get(expired).status, "EXPIRED") + + def test_succeeded_job_cannot_be_cancelled(self): + with tempfile.TemporaryDirectory() as tmp: + q = SQLiteJobQueue(f"{tmp}/queue.db") + jid = q.enqueue({"task": "x"}) + q.lease("worker") + q.ack(jid, "worker") + with self.assertRaises(RuntimeError): q.cancel(jid) + + +if __name__ == "__main__": unittest.main() diff --git a/src/test_runtime_contracts.py b/src/test_runtime_contracts.py new file mode 100644 index 0000000..c9901a2 --- /dev/null +++ b/src/test_runtime_contracts.py @@ -0,0 +1,100 @@ +import tempfile +import unittest + +from src.execution import CommandRunner +from src.github_loop import PullRequestSnapshot, SafeGitHubLoop +from src.loop_engine import Budgets, LoopState, State +from src.observability import Metrics +from src.recovery import assert_legal_transition, restore_state, serialize_state +from src.runtime_contracts import ( + BudgetLedger, FleetGovernor, ModelProfile, ModelRouter, Permission, + PermissionGate, VerificationCheck, VerifierRegistry, Verdict, + evidence_fingerprint, redact_secrets, validate_structured_output, +) + + +class Provider: + def __init__(self): self.calls = [] + def invoke(self, model, payload): + self.calls.append(model) + if model == "cheap": raise RuntimeError("temporary failure") + return {"ok": True, "usage": {"tokens": 10, "cost": 0.1}} + + +class ExpensiveProvider: + def __init__(self): self.calls = [] + def invoke(self, model, payload): + self.calls.append(model) + return {"ok": True, "usage": {"tokens": 1000, "cost": 9.0}} + + +class GitHubFake: + def __init__(self): self.snapshot = PullRequestSnapshot(1, "abc", "main", ["ci"]) + def get_pull_request(self, number): return self.snapshot + def push_branch(self, branch, expected_head_sha): return expected_head_sha + def merge_pull_request(self, number, expected_head_sha): return "merged" + + +class RuntimeContractTests(unittest.TestCase): + def test_router_fallback_and_budget(self): + ledger = BudgetLedger(100, 1.0); provider = Provider() + self.assertTrue(ModelRouter(provider, ledger).run(ModelProfile("executor", "cheap", ["strong"]), {})["ok"]) + self.assertEqual(provider.calls, ["cheap", "strong"]) + self.assertEqual(ledger.tokens_used, 10) + + def test_router_budget_failure_does_not_fallback(self): + ledger = BudgetLedger(100, 1.0); provider = ExpensiveProvider() + with self.assertRaises(RuntimeError): + ModelRouter(provider, ledger).run(ModelProfile("executor", "primary", ["fallback"], max_tokens=10, max_cost=0.5), {}) + self.assertEqual(provider.calls, ["primary"]) + self.assertEqual(ledger.tokens_used, 0) + + def test_budget_fails_closed(self): + with self.assertRaises(RuntimeError): BudgetLedger(5, 1.0).charge(tokens=6, cost=0.0) + + def test_permissions_default_deny(self): + gate = PermissionGate([Permission.READ_ONLY]) + with self.assertRaises(PermissionError): gate.require(Permission.REMOTE_MUTATION) + + def test_verifier_missing_is_inconclusive(self): + registry = VerifierRegistry(); registry.register(VerificationCheck("tests", lambda: Verdict.PASS)) + result = registry.evaluate(["tests", "security"]) + self.assertEqual(result["security"], Verdict.INCONCLUSIVE) + + def test_fingerprint_is_order_independent(self): + self.assertEqual(evidence_fingerprint(["a", "b"]), evidence_fingerprint(["b", "a", "a"])) + + def test_fleet_limits(self): + with self.assertRaises(RuntimeError): FleetGovernor(max_fanout=2).validate_spawn(depth=0, children=3) + + def test_structured_output_and_redaction(self): + validate_structured_output({"status": "ok"}, ["status"]) + with self.assertRaises(ValueError): validate_structured_output({}, ["status"]) + self.assertEqual(redact_secrets("token=abc", ["abc"]), "token=[REDACTED]") + + def test_checkpoint_round_trip_and_transition(self): + state = LoopState("id", "goal", ["criterion"], Budgets(), state=State.PLAN) + self.assertEqual(restore_state(serialize_state(state)).state, State.PLAN) + assert_legal_transition(State.DISCOVER, State.PLAN) + with self.assertRaises(ValueError): assert_legal_transition(State.DISCOVER, State.SHIPPED) + + def test_command_runner_is_permission_bound(self): + with tempfile.TemporaryDirectory() as d: + denied = CommandRunner(d, PermissionGate([Permission.READ_ONLY])) + with self.assertRaises(PermissionError): denied.run(["python", "-c", "print(1)"]) + allowed = CommandRunner(d, PermissionGate([Permission.LOCAL_MUTATION])) + self.assertEqual(allowed.run(["python", "-c", "print(1)"]).returncode, 0) + + def test_github_merge_requires_fresh_sha_checks_and_approval(self): + loop = SafeGitHubLoop(GitHubFake(), PermissionGate([Permission.HIGH_IMPACT])) + with self.assertRaises(RuntimeError): loop.merge(1, "stale", True, True) + with self.assertRaises(RuntimeError): loop.merge(1, "abc", False, True) + self.assertEqual(loop.merge(1, "abc", True, True), "merged") + + def test_metrics_records_success(self): + metrics = Metrics() + with metrics.span("ok"): pass + self.assertTrue(metrics.timings[-1].success) + + +if __name__ == "__main__": unittest.main()