LoopSmith: Closed-Loop AI Engineering — Autonomous /goal Execution for Self-Correcting Pipelines on Antigravity 2.0
Stop prompting the agent turn by turn. Hand it an objective, a bar to clear, and let the state machine write, run, and fix its own code until the output is verified.
This guide targets Antigravity 2.0 — the four-surface release (desktop app, agy CLI, google-antigravity SDK, and enterprise cloud) that shares one agent harness. The SDK is pre-v1.0; symbol names and CLI flags below reflect the documented API as of mid-2026. Treat the patterns as stable and re-check exact signatures against the current docs before you ship.
Table of contents
- The problem with the on-demand agent
- The architectural shift: closed-loop engineering as a state machine
- Step 1 — Initialize the project and the state machine (agy)
- Step 2 — Define the objective and trigger /goal execution mode (SDK)
- Step 3 — Implement the self-correcting loop
- Step 4 — Verification and state finalization
- Conclusion
1. The problem with the on-demand agent
Most “AI engineering” today is still conversational. You prompt; the model answers. You notice the answer is wrong; you prompt again. You paste a traceback; it apologizes and tries once more. The intelligence is real — but you are the control loop. You are the thing that runs the code, reads the error, decides whether the output is good enough, and feeds the next instruction back in. Take the human out of that seat and the whole system stops.
That’s fine for a chat window. It falls apart the moment you want an agent to produce a deliverable — a cleaned dataset, a reconciled financial report, a migration that actually compiles. Real analytical work is iterative and self-referential: you write a script, it crashes on a currency string, you fix the parse, it runs but the totals don’t reconcile, you fix the aggregation, and only then is the output trustworthy. Every one of those arrows is a decision. An on-demand agent makes you supply all of them.
There are three specific failures hiding in “just prompt it again”:
- No definition of done. A conversational agent stops when it feels finished, which is to say when it produces fluent text. Fluent is not the same as correct. “Here’s your revenue analysis!” over a script that silently dropped a third of the rows reads exactly like a correct answer. Nothing in the loop checks.
- The human is the error handler. The agent has no durable memory of what it was trying to achieve across turns, so it can’t tell whether its latest attempt is closer or further from the goal. You hold that state in your head. You are the while loop, the try/except, and the assertion — manually, turn after turn.
- No verified exit. Because there’s no explicit success criterion, there’s no moment the system can point to and say “done, and here’s the proof.” It just… trails off when you stop typing.
The fix is not a smarter model or a longer prompt. It’s an architecture: give the system the objective and the definition of done, then let it run its own control loop — generate, execute, inspect, refine — until the output clears a bar it can check for itself. That is closed-loop AI engineering, and Antigravity 2.0 makes it a first-class execution mode: /goal.
We’ll build one end to end. The running example is a workload every data team recognizes: hand an agent a messy CSV and an analytical objective, and get back a verified JSON — with zero human turns in the middle. The dataset fights back on purpose (revenue stored as “$1,299.00”, blank region cells, mixed date formats), so the agent’s first attempt fails and you get to watch it correct itself. We call the reference implementation LoopSmith.
2. The architectural shift: closed-loop engineering as a state machine
Stop thinking of the agent as a chatbot with a scrollback. Think of it as a process with states. At any instant an autonomous run is doing exactly one thing — planning, or generating code, or executing it, or inspecting the result, or refining — and the only interesting question is which state comes next. That question has a small, finite set of legal answers. Encode them, and you have a finite-state machine: the most boring, most battle-tested control structure in all of engineering, pointed at an LLM.
Here is the entire topology. Read the cycle in the middle — that loop is the self-correction engine.
- Determinism you can audit. Control flow lives in a transition table, not in the model’s whim. The run can only move along edges you drew. Every execution emits a replayable tape — plan → generate → execute → inspect → refine → execute → inspect → verified — so when something goes sideways you read the tape instead of guessing. The model supplies judgement inside a state; the machine decides where the run is allowed to go next. That division is the whole trick.
- Self-correction as a cycle, not a retry. The EXECUTE → INSPECT → REFINE → EXECUTE loop is a closed feedback circuit. INSPECT observes the actual result (exit code, stderr, whether the numbers reconcile) and emits a diagnosis; REFINE consumes that diagnosis and emits a better program; EXECUTE tests it. No human turn anywhere on that circuit. The agent isn’t “trying again” — it’s reading its own error and steering.
- A verified exit, defined in advance. The run terminates in VERIFIED only when every acceptance check passes — and those checks are declared up front, in the PLAN state, as machine-checkable predicates (exit code 0, the artifact parses, Σ(by_region) == total within a cent). “Done” stops being a vibe and becomes a boolean. If the attempt budget runs out first, the run exits FAILED — loudly, with its tape — rather than pretending.
The opposite of this is the open loop: prompt in, text out, human holds the state. Everything below is the move from open to closed. Two design rules make it work in practice, and they’re worth stating before any code:
- Split judgement from control. LLMs are extraordinary at the judgement inside a state (“this traceback means the currency column is a string; strip $ and , before casting”) and unreliable at being their own scheduler. So the state graph is plain, deterministic Python; the agents never decide what state comes next — they only decide what to do in the state they’re in.
- Verify with code, not vibes. The parts of “done” that a computer can check — exit codes, schemas, reconciliation — are checked in code, as ground truth the model may not overrule. The model’s judgement is layered on top for the things code can’t assert, never underneath.
Let’s build it.
3. Step 1 — Initialize the project and the state machine (agy)
Install the two surfaces you’ll use. The CLI (agy) is the Go-based terminal harness; the SDK (google-antigravity) is the same harness as a Python library. You prototype the loop interactively in the CLI, then codify it in Python.
# 1. Install the Antigravity CLI (macOS / Linux). It lands at ~/.local/bin/agy
curl -fsSL https://antigravity.google/cli/install.sh | bash
# Windows PowerShell: irm https://antigravity.google/cli/install.ps1 | iex
# 2. Confirm the harness is live and see which models you can reach
agy models
# 3. Scaffold the project
mkdir loopsmith && cd loopsmith
python -m venv .venv && source .venv/bin/activate
# 4. Install the SDK. IMPORTANT: install the published wheel, not a source
# checkout - the wheel ships the `localharness` binary the SDK drives.
pip install google-antigravity pydantic python-dotenv pandas
# 5. Authenticate. Either a direct Gemini key…
export GEMINI_API_KEY="your_key_here"
# …or route through Vertex AI with Application Default Credentials:
# gcloud auth application-default login # then set VERTEX=1 in .env
Before you write a single agent, write the state machine. This is the inversion that makes an autonomous run legible: the topology is defined first, in isolation, with no model anywhere near it. The seven states, the legal transitions, and a guard that refuses anything off the graph:
# loop/states.py
from dataclasses import dataclass, field
from enum import Enum
class State(str, Enum):
PLAN = "plan" # decompose the objective into an approach + checks
GENERATE = "generate" # write the first analysis script
EXECUTE = "execute" # run the script in the sandbox
INSPECT = "inspect" # judge the run against the acceptance checks
REFINE = "refine" # rewrite the script to fix what INSPECT found
VERIFIED = "verified" # terminal SUCCESS - all checks pass
FAILED = "failed" # terminal FAILURE - attempt budget exhausted
TERMINAL = frozenset({State.VERIFIED, State.FAILED})
# The transition table IS the architecture. A move not listed here is a bug.
TRANSITIONS = {
State.PLAN: frozenset({State.GENERATE}),
State.GENERATE: frozenset({State.EXECUTE}),
State.EXECUTE: frozenset({State.INSPECT}),
State.INSPECT: frozenset({State.VERIFIED, State.REFINE, State.FAILED}),
State.REFINE: frozenset({State.EXECUTE}),
}
class IllegalTransition(RuntimeError):
"""Raised when the run attempts a move the transition table forbids."""
@dataclass
class Machine:
state: State = State.PLAN
history: list[State] = field(default_factory=lambda: [State.PLAN])
def to(self, nxt: State) -> State:
allowed = TRANSITIONS.get(self.state, frozenset())
if nxt not in allowed:
raise IllegalTransition(
f"illegal transition {self.state.value!r} → {nxt.value!r} "
f"(allowed: {sorted(s.value for s in allowed) or 'none - terminal'})"
)
self.state = nxt
self.history.append(nxt)
return nxt
@property
def done(self) -> bool:
return self.state in TERMINAL
def tape(self) -> str:
return " → ".join(s.value for s in self.history)
Why this is the first thing you write, not the last:
- TRANSITIONS is the contract. Notice the cycle: EXECUTE → INSPECT → REFINE → EXECUTE. That single loop in a dictionary is the self-correction engine — the run walks it until INSPECT emits VERIFIED. And notice what’s absent: there is no edge from REFINE to GENERATE, no edge out of VERIFIED. Illegal moves aren’t handled defensively; they’re impossible.
- Machine.to() is the single choke point. Every state change in the entire system flows through this one method. That’s what guarantees history is complete and the tape is trustworthy — there’s no back door that mutates state without recording it. In Step 4 the audit hook narrates this live.
- It’s model-free, so it’s testable. You can unit-test the topology — “can a run reach VERIFIED without passing through INSPECT?” — with zero API calls and zero flakiness. The stochastic part (the agents) and the deterministic part (the graph) never contaminate each other.
Interactively, you can watch the same machine run before you’ve written any Python at all. From the agy TUI, /goal drops you into execution mode and streams the state transitions as they happen:
agy> /goal "Reconcile revenue by region in ./data/sales.csv into a verified JSON"
◆ plan → 6 acceptance checks declared
◆ generate → analysis.py (41 lines)
◆ execute → exit 1 (ValueError: could not convert string to float: '$1,299.00')
◆ inspect → NOT verified · reconciles_total blocked by crash
◆ refine → strip '$'/',' before cast
◆ execute → exit 0 · result.json written
◆ inspect → NOT verified · no_null_region FAIL (blank region leaked as key)
◆ refine → drop blank-region rows
◆ execute → exit 0
◆ inspect → VERIFIED ✓ (6/6 checks)
◆ done → tape: plan→generate→execute→inspect→refine→execute→inspect→refine→execute→inspect→verified
That’s the whole story on one screen: two self-corrections, no human turn, a verified exit. Now we build the Python that makes it reproducible.
4. Step 2 — Define the objective and trigger /goal execution mode (SDK)
/goal mode needs two inputs and nothing else: a high-level objective and a definition of done. Everything else — how many iterations, which script, which fix — the loop decides for itself.
There are two legitimate ways to invoke it, and a senior engineer should know both.
Pattern A — harness-native /goal. Hand the harness the objective and a success criterion and let it own the loop. The SDK exposes run_goal; the harness plans, generates, executes, inspects, and refines internally, and returns a GoalResult. This is the least code and the right default for exploratory work:
# the one-liner form — the harness runs the whole closed loop for you
from google.antigravity import Agent, GoalConfig
async def quick_goal():
async with Agent.for_goal(GoalConfig(model="gemini-3-pro")) as agent:
result = await agent.run_goal(
objective=(
"Analyze ./data/sales.csv and report total revenue per region, "
"the top region, and its share of total revenue as result.json."
),
success=(
"python exits 0; result.json parses; sum(by_region) == total_revenue "
"within 0.01; top_region == argmax(by_region); no blank region key."
),
max_iterations=5, # the self-correction budget
)
print(result.final_state, result.verified, result.tape)
Pattern B — the explicit state machine. When you need determinism and auditability — which a production pipeline always does — you drive the same loop yourself with the SDK primitives, so every transition, every check, and every refinement is your code. This is what the rest of the tutorial builds, because seeing the machinery is the point.
Either way, the definition of done is not prose you hope the model honors — it’s a list of acceptance checks, and the first agent’s whole job is to produce them. Structured output is what makes that possible. Every model turn in LoopSmith is constrained to a Pydantic schema via response_schema=, so the state machine receives typed objects it can branch on, not text it has to parse:
# loop/schemas.py (the contracts that flow between states)
from enum import Enum
from pydantic import BaseModel, Field
class CheckKind(str, Enum):
DETERMINISTIC = "deterministic" # enforced in code (verifier.py)
JUDGEMENT = "judgement" # delegated to the INSPECT agent
class AcceptanceCheck(BaseModel):
id: str = Field(description="stable slug, e.g. 'reconciles_total'")
description: str
kind: CheckKind
class Plan(BaseModel):
objective: str
approach: list[str]
output_contract: str # the exact JSON the script must write
checks: list[AcceptanceCheck] = Field(min_length=1)
class GeneratedCode(BaseModel): # reused by BOTH generate and refine
filename: str = "analysis.py"
language: str = "python"
source: str # the complete file contents, ready to run
notes: str = "" # what changed vs. the last version
class Inspection(BaseModel): # the branch decision, as data
verified: bool # True iff EVERY acceptance check passes
failing_checks: list[str] = Field(default_factory=list)
diagnosis: str # root cause, grounded in stderr / checks
fix_hint: str = "" # concrete change REFINE should make
Two schema decisions carry the architecture:
- Inspection.verified is a bool, so the branch is a field read. The whole INSPECT → {VERIFIED, REFINE} fork collapses to if inspection.verified:. No regex over model prose, no “did it say the word passed?” heuristics. The model fills a typed field; the machine reads it.
- GeneratedCode is shared by GENERATE and REFINE. A refinement is just the next version of the same artifact. Because both states emit the identical schema, the transition table can route both into EXECUTE with no special casing — the loop doesn’t care whether a script is the first draft or the third.
5. Step 3 — Implement the self-correcting loop
Now the substance: the four agents, the sandbox that runs their code, and the driver that walks the state machine. Every agent is minted from one factory with one set of guardrails — the Shared Agent Harness. An autonomous run that writes and executes its own code is exactly the run you want on the shortest possible leash: default-deny policy, one audit line per tool call, no network, no shell.
# loop/harness.py
import os
from typing import Callable, Sequence
from google.antigravity import LocalAgentConfig
from google.antigravity.hooks import hooks, policy
MODEL = os.getenv("LOOPSMITH_MODEL", "gemini-3-pro")
# Least privilege, encoded once, inherited by every agent in the loop.
BASE_POLICIES = [
policy.allow("view_file"),
policy.allow("write_file", when=lambda a: _within_workspace(a.get("path", ""))),
policy.deny("run_command"), # agents never shell out - the SANDBOX does
policy.deny("net_fetch"), # a self-correcting loop must stay hermetic
policy.deny("*"), # default-deny: anything not allowed is refused
]
class StateAuditHook(hooks.PostToolCallHook):
async def run(self, context, data) -> None:
print(f"[audit] state={getattr(context, 'label', '?')} "
f"tool={getattr(data, 'name', '?')} ok={getattr(data, 'ok', True)}")
def build_agent_config(*, label: str, system_instructions: str,
tools: Sequence[Callable] = ()) -> LocalAgentConfig:
kwargs = dict(
model=MODEL, label=label, system_instructions=system_instructions,
tools=list(tools), policies=BASE_POLICIES, hooks=[StateAuditHook()],
)
if os.getenv("VERTEX") == "1":
kwargs["vertex"] = True
elif os.getenv("GEMINI_API_KEY"):
kwargs["api_key"] = os.environ["GEMINI_API_KEY"]
return LocalAgentConfig(**kwargs)
def _within_workspace(path: str) -> bool:
return "workspace" in os.path.normpath(path).split(os.sep)
The critical decision here is policy.deny(“run_command”) — the agents themselves get no shell. That looks paradoxical for a loop whose whole job is to run code, until you see where execution actually happens: in a separate, deterministic sandbox the orchestrator controls, not inside a model’s tool call. The model writes the program; a governed subprocess runs it. The model can never pip install or curl its way out, because it was never handed the keys.
The four agents — judgement, no control flow
Each state’s agent is a single-responsibility config. None of them holds orchestration logic; none of them knows it’s in a loop. They reason over what they’re handed and emit a typed object — that’s it.
# loop/agents.py (abridged — full system prompts in the repo)
from .harness import build_agent_config
def planner_config(): # PLAN: objective -> approach + machine-checkable checks
return build_agent_config(label="plan", system_instructions=(
"You are a senior data analyst PLANNER. Given an objective and a preview "
"of a messy CSV, produce an approach, an exact output_contract for the "
"JSON the script must write, and a list of ACCEPTANCE CHECKS. Prefer these "
"deterministic ids when they apply: exits_clean, artifact_exists, "
"schema_ok, no_null_region, reconciles_total, top_is_argmax. The run is "
"'verified' only when EVERY check passes - do not under-specify."))
def generator_config(): # GENERATE: write the first script
return build_agent_config(label="generate", system_instructions=(
"You are a Python DATA ENGINEER. Write a COMPLETE, self-contained script "
"that reads os.environ['LOOPSMITH_DATASET'], performs the planned "
"analysis, and writes 'result.json'. Use only the standard library and "
"pandas. No network. Return the full file contents."))
def inspector_config(): # INSPECT: verified vs. refine, grounded in hard checks
return build_agent_config(label="inspect", system_instructions=(
"You are a rigorous VERIFICATION agent. You get the acceptance checks, the "
"sandbox RunResult, and the results of the DETERMINISTIC checks already "
"computed in code. You may NOT mark verified if any deterministic check "
"failed - those are ground truth. When not verified, write a diagnosis "
"rooted in the real stderr/failed checks and a concrete fix_hint."))
def refiner_config(): # REFINE: rewrite to fix exactly what INSPECT diagnosed
return build_agent_config(label="refine", system_instructions=(
"You are a DEBUGGING engineer. Given the current script, the RunResult "
"showing how it failed, and a diagnosis + fix_hint, produce the NEXT full "
"version that fixes the identified problem while preserving what worked."))
The asymmetry is the design. The planner turns a fuzzy objective into a testable bar. The inspector is explicitly forbidden from overruling the deterministic checks — its judgement sits on top of ground truth, never underneath it. The refiner receives a diagnosis, not a vague “try again,” so its rewrite is targeted. Each has a small context window and one job.
The sandbox — where generated code actually runs
This module is real, runnable Python — no SDK, no model. It is the EXECUTE state, and it’s the security boundary. It runs the agent’s script as a subprocess, in a confined working directory, with a wall-clock timeout and a stripped environment, and reports back exactly what happened.
# loop/sandbox.py (core)
import subprocess, sys, time
from pathlib import Path
from .schemas import GeneratedCode, RunResult
WORKSPACE = Path(__file__).resolve().parent.parent / "workspace"
def write_code(code: GeneratedCode) -> Path:
WORKSPACE.mkdir(parents=True, exist_ok=True)
path = WORKSPACE / code.filename
path.write_text(code.source, encoding="utf-8")
return path
def run_python(path: Path, *, dataset: Path, timeout_s: float = 30.0) -> RunResult:
before = _snapshot() # which files exist before the run
started = time.monotonic()
try:
proc = subprocess.run(
[sys.executable, str(path)],
cwd=WORKSPACE, # confine the working directory
env={"LOOPSMITH_DATASET": str(dataset), # minimal env - no secrets leak
"PATH": _safe_path(), "PYTHONUNBUFFERED": "1"},
capture_output=True, text=True, timeout=timeout_s,
)
exit_code, stdout, stderr = proc.returncode, proc.stdout, proc.stderr
timed_out = False
except subprocess.TimeoutExpired as e:
exit_code, stdout = 124, (e.stdout or "")
stderr = (e.stderr or "") + f"n[sandbox] killed after {timeout_s}s timeout"
timed_out = True
artifacts = sorted(p.name for p in _snapshot() - before) # what the run created
return RunResult(exit_code=exit_code, stdout=_clip(stdout), stderr=_clip(stderr),
duration_s=round(time.monotonic() - started, 3),
artifacts=artifacts, timed_out=timed_out)
Line by line, this is the entire EXECUTE contract:
- subprocess.run(…, cwd=WORKSPACE, timeout=timeout_s) — the generated code runs outside the agent’s process, in its own directory, and cannot run longer than the timeout. An infinite loop the model wrote gets SIGKILL, surfaces as exit_code=124, and becomes just another INSPECT input — not a hung pipeline.
- env={…} — a hand-built environment. The dataset path arrives via LOOPSMITH_DATASET; your API keys, cloud credentials, and shell aliases do not. The script sees exactly what it needs and nothing else.
- _snapshot() before and after — the sandbox diffs the workspace to report precisely which artifacts the run produced. That’s how the loop knows result.json was written, rather than trusting the model’s stdout claim that it was.
- The RunResult is the feedback signal. Exit code, stdout, stderr, artifacts — the raw truth of what happened, handed straight to INSPECT. The loop corrects itself against observed behavior, never against the model’s hopes.
The driver — walking the machine
Here is the closed loop, top to bottom. This is the explicit form of /goal: you can trace every transition, and every m.to(…) is a guarded, logged move.
# loop/orchestrator.py
from pathlib import Path
from google.antigravity import Agent
from . import dataset, sandbox, verifier
from .agents import planner_config, generator_config, inspector_config, refiner_config
from .schemas import Plan, GeneratedCode, Inspection, GoalResult
from .states import Machine, State
async def _turn(config, prompt, schema):
"""One isolated agent, one typed turn, then tear down."""
async with Agent(config) as agent:
resp = await agent.chat(prompt, response_schema=schema)
return await resp.parsed() # a validated Pydantic instance
async def run_goal(objective, *, dataset_path, max_attempts=5) -> GoalResult:
data = Path(dataset_path)
preview = dataset.preview(data)
m = Machine() # starts in PLAN
# ---- PLAN: objective -> approach + acceptance checks --------------------
plan = await _turn(planner_config(),
f"OBJECTIVE:n{objective}nnDATASET PREVIEW:n{preview}", Plan)
m.to(State.GENERATE)
# ---- GENERATE: write the first script -----------------------------------
code = await _turn(generator_config(), _generate_prompt(objective, plan, preview),
GeneratedCode)
code_path = sandbox.write_code(code)
m.to(State.EXECUTE)
attempts = 0
while True:
# ---- EXECUTE: run the current script in the sandbox -----------------
run = sandbox.run_python(code_path, dataset=data)
attempts += 1
m.to(State.INSPECT)
# ---- INSPECT: hard checks first, then agent judgement ---------------
det = verifier.evaluate(plan.checks, run, sandbox.WORKSPACE)
inspection = await _turn(inspector_config(),
_inspect_prompt(plan, run, det), Inspection)
verified = inspection.verified and verifier.all_passed(det)
if verified:
m.to(State.VERIFIED); break # terminal success
if attempts >= max_attempts:
m.to(State.FAILED); break # terminal failure - budget spent
# ---- REFINE: rewrite the script to fix what INSPECT found -----------
m.to(State.REFINE)
code = await _turn(refiner_config(),
_refine_prompt(code, run, det, inspection), GeneratedCode)
code_path = sandbox.write_code(code)
m.to(State.EXECUTE) # close the loop - back to EXECUTE
return GoalResult(
objective=objective, final_state=m.state.value,
verified=(m.state is State.VERIFIED), attempts=attempts, tape=m.tape(),
report=verifier.load_report(sandbox.WORKSPACE) if m.state is State.VERIFIED else {},
code_path=str(code_path), artifacts=run.artifacts,
)
Read that while loop as the feedback circuit it is:
- run = sandbox.run_python(…) is the plant — it produces a real observation. det = verifier.evaluate(…) and inspection are the sensor. code = await _turn(refiner_config(), …) is the controller. The loop closes when EXECUTE runs the controller’s new output. That’s not a metaphor; it’s literally a discrete feedback controller with an LLM in the actuator.
- verified = inspection.verified and verifier.all_passed(det) is the safety interlock. The model’s verified flag is necessary but not sufficient — the deterministic checks get a hard veto via the and. A model that hallucinates success cannot force a VERIFIED exit; the arithmetic won’t let it.
- if attempts >= max_attempts: m.to(State.FAILED) is the anti-livelock. A self-correcting loop must be able to give up. Without this bound, a genuinely impossible objective would burn tokens forever. With it, the run terminates honestly — FAILED, with its full tape — and a human is paged with the evidence already assembled.
- Every branch is a guarded m.to(…). The control flow you’re reading is the only control flow there is; the model never picks the next state. That is what makes an autonomous run something you can put on a schedule and trust.
Point this at the messy sales.csv and the first EXECUTE fails exactly as designed — ValueError: could not convert string to float: ‘$1,299.00’. INSPECT reads that traceback, sees reconciles_total blocked, and emits a diagnosis; REFINE strips the $ and , and re-runs; the next INSPECT catches a subtler bug (a blank region cell leaked in as a by_region key); REFINE drops it; the third run clears all six checks. No human typed a word.
6. Step 4 — Verification and state finalization
The VERIFIED exit is only meaningful if “verified” means something a machine checked. That’s the verifier — the deterministic half of INSPECT. It’s real, runnable Python, and the INSPECT agent is forbidden from overruling it.
# loop/verifier.py (core checks)
import json
from pathlib import Path
from .schemas import AcceptanceCheck, CheckKind, CheckResult, RunResult
RESULT_FILE = "result.json"
TOLERANCE = 0.01 # one cent
def evaluate(checks, run, workspace) -> list[CheckResult]:
result = _load_result(workspace) # parse the artifact once
out = []
for check in checks:
if check.kind is not CheckKind.DETERMINISTIC:
continue # judgement checks belong to INSPECT
out.append(_run_check(check, run, result))
return out
def _reconciles_total(_run, result): # the check that matters most
by_region, total = result["by_region"], result["total_revenue"]
summed = sum(float(v) for v in by_region.values())
delta = abs(summed - float(total))
return delta <= TOLERANCE, f"Σby_region={summed:.2f} vs total={total:.2f} (Δ={delta:.4f})"
def _no_null_region(_run, result): # catches the blank-cell leak
bad = [k for k in result.get("by_region", {}) if k in (None, "", "nan", "None")]
return (not bad), (f"leaked empty region key(s): {bad}" if bad else "clean")
def _top_is_argmax(_run, result): # the claim must match the data
by = result["by_region"]
argmax = max(by, key=lambda k: float(by[k]))
return argmax == result["top_region"], f"argmax={argmax!r} vs claim={result['top_region']!r}"
Three things make this a finalization mechanism and not just a linter:
- Checks fail closed. An unknown check id, a crashing check, a missing artifact — all resolve to False, never True. The loop can only exit VERIFIED by affirmatively satisfying every predicate. Ambiguity refines; it never passes. That’s the correct default for autonomy: silence is not consent.
- Ground truth outranks the model. _reconciles_total re-computes the sum from the artifact the script wrote and compares it to the script’s own claimed total. If the code dropped rows, the numbers diverge and the check fails — regardless of how confidently the model narrated success. Reconciliation is not a matter of opinion.
- The verified artifact is the deliverable. On a VERIFIED exit, verifier.load_report() reads back the proven result.json and stamps it into the GoalResult. The run doesn’t return a description of the analysis — it returns the analysis, already checked:
{
"objective": "…revenue per region, top region, and its share…",
"final_state": "verified",
"verified": true,
"attempts": 3,
"tape": "plan → generate → execute → inspect → refine → execute → inspect → refine → execute → inspect → verified",
"report": {
"by_region": {"North": 10749.90, "South": 8552.50, "East": 7212.25, "West": 8631.00},
"total_revenue": 35145.65,
"top_region": "North",
"top_region_share": 0.3059
}
}
The tape field is the finalization proof: you can read the entire life of the run — three EXECUTE attempts, two self-corrections, a verified exit — as a single string. And because the machine only reaches VERIFIED by clearing every check, that report is safe to hand downstream without a human re-checking it. That is the whole promise of closed-loop engineering: not “the agent produced something,” but “the agent produced something, and here is the machine-checked proof it’s right.”
To run the whole thing as a governed build step — headless, worktree-isolated, pipeable — that’s the agy CLI in /goal mode:
agy goal
--max-iterations 5
--success "python main.py exits 0 and workspace/result.json reconciles:
sum(by_region) == total_revenue (±0.01), top_region == argmax,
and no blank region key."
--output-format json
-p "Run `python main.py`. If the acceptance checks fail, inspect the
traceback and refine the analysis script until every check passes.
Then commit result.json and open a PR '[goal] verified analysis'."
| jq '{state: .final_state, verified, attempts, tape}'
–success declares the bar; –max-iterations bounds the budget; –output-format json makes the run a pipe stage. Each scheduled run gets its own git worktree, so the loop can rewrite its own scripts all it likes without ever touching your live checkout. The CLI and the SDK are two faces of one harness — prototype the loop interactively with /goal, then codify it in Python for CI.
7. Conclusion
The on-demand agent was a useful stage — it proved the model could reason. But reasoning isn’t engineering. Engineering is a loop: produce, test, observe, correct, and stop only when the output clears a bar you set in advance. For thirty years we’ve built that loop out of CI pipelines and assertions and retries. The shift Antigravity 2.0 makes possible is to put an LLM inside that loop as the generator and the debugger, while keeping the loop itself deterministic, bounded, and auditable.
That’s the mental model to leave with. The state machine is not boilerplate around the intelligence — it is the engineering. The model supplies judgement inside each state; the machine supplies the control flow, the verification, and the verified exit. /goal gives you the objective-in, proof-out contract; structured output makes each transition a field read instead of a guess; the sandbox lets the agent run its own code without you trusting it; and the verifier makes “done” a boolean a machine can defend.
Build the machine once. Then hand it an objective and walk away — and come back to a result with a receipt.
Acknowledgement
Google Cloud Credits are provided for this project. #AgenticArchitectSprint #Antigravity
Further reading
- Google Antigravity SDK (Python): https://github.com/google-antigravity/antigravity-sdk-python
- Antigravity CLI (agy): https://github.com/google-antigravity/antigravity-cli
- /goal execution mode: https://antigravity.google/docs/goal-mode
- Companion tutorial — Multi-Agent Dispatch (concurrent subagents): the fan-out counterpart to this fan-forward loop. LoopSmith: Closed-Loop AI Engineering — Autonomous /goal Execution for Self-Correcting Pipelines on Antigravity 2.0
LoopSmith: Closed-Loop AI Engineering — Autonomous /goal Execution for Self-Correcting Pipelines on… was originally published in Google Developer Experts on Medium, where people are continuing the conversation by highlighting and responding to this story.
