What this post covers
The big changes in this post are:
- Add more static validators, like
tflintandcheckov - Add a structured output type that nudges the agent towards self-reporting
- Build an eval set and use
pydantic-evalsto run it
For any agentic system the cheapest way to get better results is if you can add inexpensive static validators the agent can run itself, and that we re-run on submission. This is particularly true for this agent that produces files on disk. The tools the agent runs and the tools that grade its submission are the same set. We just layer deterministic formatting on top.
How strict the validators should be is a trade-off. checkov ships policies that are far stricter than many projects need, and that context would need to be configured, which we do not want to add right now. So making it slightly less strict is a good way to manage cost (default settings ballooned cost for some models by 100x).
pydantic-evals is the foundation for anything we add later, and eval cases for a real (non-blog example) scenario would need to significantly extend it and add cases that start from an existing project. Currently we have five simple test cases, which does not give us a lot of confidence statistically. Take the results presented with a pinch of salt: they are directionally useful and enough to rule a model out, but they do not predict performance on new and more complex cases.
The work in this post turned up a few gotchas that forced changes to earlier code. With more validation tools the total run size increased, eventually breaching Firehose's record cap (we use it to store audit traces). So we had to add batching for the audit copy. We also cap the model's max_tokens and the run's request_limit, so the message history cannot grow without a ceiling. That does not bound what a tool returns, though MAX_REPORT_CHARS caps the one that mattered, checkov's report. Beyond that we would have to trim or offload the history itself.
From validators to evals
The additional validators we added in this post for the agent answer whether the terraform code is correct, reasonably secure according to static rules, and in line with our lint style. They say nothing about whether it did the task it was assigned. For example if it was tasked to create an AWS S3 bucket and did not, then every one of those checks would still pass.
That is where the evals come in. The structure in pydantic-evals is a small vocabulary. A case is a fixed input plus the metadata that says what should come out of it. The agent runs and results are passed to evaluators. These can be boolean, which gives a pass rate, categorical, which classifies runs and gives a ratio per label, or numeric, which averages into a score. Evaluators can also use OTEL span information meaning even tool calls can be evaluated if tool usage is expected. In addition, token usage and pydantic-ai's genai-prices integration, which we extend with prices it has not shipped yet, give us run cost and cost per pass.
Building an eval harness is the only real way to decide on a model or make an informed change to the setup (like system prompts or tooling). LLM agents are non-deterministic, so without a way to evaluate the results we cannot make any decisions. Collecting stats like token usage, pass rate, and cost is also what lets you run a model to a budget. If your budget is $0.20 per successful run, then you need to know how often you land under it. There is no 100% safety as your evals do not cover everything, which is why you should also do live monitoring for this - the initial premise of our series.
Our eval-case structure is currently basic. We have the inputs (the task, what the agent should do) and some metadata which we use to configure the validators (for example expected terraform resources).
evals/cases.yaml
cases:
- name: logs-bucket
inputs: >-
Create an S3 bucket for storing application logs. Make sure the
configuration blocks public access.
metadata:
expected_resources:
- type: aws_s3_bucket
- type: aws_s3_bucket_public_access_block
The result
We did run the eval harness against a few models:
-
GLM 5.2 (
glm5p2), Z.ai's 743B mixture-of-experts model with a 1M-token context, built for long-horizon coding. We reach it through Fireworks. -
Mistral Medium (
mistral-medium), Mistral's mid-tier generalist, which their own docs position for agentic and coding work. -
Haiku 4.5 (
haiku), Anthropic's fastest model, and the only Bedrock entry in our registry. -
Mistral Large (
mistral-large), Mistral's largest general-purpose model. -
Codestral (
codestral), Mistral's code-completion specialist (not a good choice for the task, which you will see).
Now plotting pass rate against cost per task gives us a decent model selection picture. First of all haiku drops out on cost: it passes as often as mistral-medium and costs more than five times as much per task. glm5p2 is the only model that passed all fifteen runs, but mistral-medium is cheaper per task and its single miss was a run that hit the request limit rather than bad Terraform, so it still comes out ahead on cost per pass. I would expect this to change once we add more complex validation checks and cases, but for now with the eval data we have mistral-medium would be the pick if you can absorb the occasional failure with a retry, and glm5p2 if you want the consistency without one.
See this chart full-size on the original post.
The fuller numbers (n=15: 5 cases x 3 repeats), including the tool-error score:
| Model | Pass rate | Errored | Eff $/M tok | Cost / task | Cost / pass | Tool-error |
|---|---|---|---|---|---|---|
| glm5p2 | 100% | 0 | $0.68 | $0.036 | $0.036 | 0.83 |
| mistral-medium | 93% | 1 | $0.45 | $0.028 | $0.030 | 0.66 |
| haiku | 93% | 1 | $1.25 | $0.156 | $0.167 | 0.60 |
| mistral-large | 87% | 1 | $2.16 | $0.084 | $0.097 | 0.61 |
| codestral | 33% | 10 | $0.33 | $0.008 | $0.025 | 0.68 |
Tool-error is a score between 0 and 1, not a boolean like the evaluators behind the pass rate. It starts at 1 and for each tool call that needs to be retried the score is reduced by 0.1. So you can see that mistral-medium has more tool errors than glm5p2. The causes differ, and some would yield to better instructions. For example in a particular eval case for the log bucket the agent ran terraform init as instructed, then wrote a file adding the first aws provider resource, and then failed on terraform validate with Error: Missing required provider. An instruction covering this edge case would probably have caught it (in real scenarios we should have already aws provider resources in the project). Others are just plain mechanical validations like tflint's terraform "required_version" attribute is required (terraform_required_version). glm5p2 avoids most of these, while mistral-medium got confused enough in one run that it never finished inside the request limit.
What is nice if logfire is configured is the ability to drill into a particular eval case and look at the spans which include the agent conversation and the tool retry errors reducing our score.
We only have one case where an assertion failed, meaning the model did not fully meet the case's expectations. Most other failures are mechanical, runs that never finished. The 50-request limit is what stops a confused run from spiralling in cost.
See this chart full-size on the original post.
Difficulty is not spread evenly either. queue-with-dlq and sessions-table account for most of the misses, while logs-bucket and lambda-exec-role were passed by every model except codestral, which missed one run of each.
See this chart full-size on the original post.
The code
New to the series? Tooling, AWS access, and project setup are covered in Part 1 (linked above).
The final tree. + is new in post 5, ~ extends a post 4 file, blank carries unchanged. Click any changed or new file to read it; the download below fast-forwards to this state if you want to walk through the post against the finished code.
The following files are new at this checkpoint:
| File | What it does |
|---|---|
agent/tools/ |
Tools become a package: filesystem.py tracks changed files, validators.py adds tflint and checkov
|
evals/ |
The offline harness: cases.yaml and its generated schema, evaluators.py, run.py. Never ships in the image |
agent/prices.py |
The glm5p2 price genai-prices has not shipped yet. In agent/, so a live run is priced like a sweep |
tests/test_evaluators.py, tests/fixtures/plan.json
|
The plan-comparison logic against a canned plan document |
tests/test_run.py |
The harness points model resolution at the eval registry |
These carry forward from post 4 with changes:
| File | What changed |
|---|---|
agent/core.py |
TaskResult output type and output validator; execute() takes an optional workspace, and the run states its own ceilings |
agent/lambda_entry.py |
The response output becomes the serialized TaskResult
|
agent/models.py |
Resolves whichever registry MODELS_PARAMETER names, and silences the Mistral SDK's duplicate chat spans |
agent/observability.py |
One Firehose record per span, EMF metrics matched on span scope as well as the metadata attribute, and the price registered |
tests/ |
test_core.py, test_lambda_entry.py, test_tools.py, test_observability.py and test_validators.py cover the above |
Build and infrastructure:
| File | What changed |
|---|---|
infra/models.tf |
A second registry for sweeps, whose Bedrock entry points at its own profile tagged Purpose=eval
|
infra/lambda.tf |
The Lambda role can read the Fireworks key, so a live run can reach glm5p2 and not just an eval can |
infra/variables.tf |
default_model moves to mistral-medium, the model the sweep below argues for |
Dockerfile |
Two more pinned stages: the tflint binary, checksum-verified, and a checkov venv |
pyproject.toml |
The dev group gains pydantic-evals and typer
|
Two rows in those tables have a story behind them. The audit copy from post 2 shipped a whole trace as one Firehose record, which a long validation run outgrew, so it now ships one record per span, and the run states its own ceilings so no single span can outgrow the cap either.
The other is cost. An uncapped checkov report is not paid for once: it joins the message history and is charged again on every later turn, so validators.py caps what a failed check hands back to the model.
terraform-pr-agent/
agent/
tools/
+ __init__.py
+ filesystem.py
+ validators.py
__init__.py
~ core.py
env.py
~ lambda_entry.py
memory.py
~ models.py
~ observability.py
+ prices.py
runs.py
ssm.py
evals/
+ __init__.py
+ cases_schema.json
+ cases.yaml
+ evaluators.py
+ run.py
infra/
placeholder/
Dockerfile
handler.py
alerts.tf
audit-bucket.tf
bedrock.tf
cloudwatch.tf
ecr.tf
firehose.tf
iam.tf
kms.tf
~ lambda.tf
logfire.tf
main.tf
~ models.tf
runs-bucket.tf
~ variables.tf
scripts/
build-lambda.sh
chat.py
queries.sql
traces.sql
tests/
fixtures/
+ plan.json
+ clean.tf
conftest.py
~ test_core.py
test_env.py
+ test_evaluators.py
~ test_lambda_entry.py
+ test_observability.py
+ test_run.py
~ test_tools.py
+ test_validators.py
.dockerignore
.envrc
.envrc.local
.gitignore
AGENTS.md
~ Dockerfile
~ pyproject.toml
README.md
Browse these files interactively on the original post.
Fast-forward to the final code of this post:
mkdir -p ~/projects
cd ~/projects
curl -fsSL https://andreaslang.dev/terraform-pr-agent/terraform-pr-agent-05.tar.gz | tar xz
More validation tools
We extend the validation suite well beyond Post 4's terraform_validate. It is nice to tell the agent "make sure the S3 bucket you create follows security best practices", but it is better to use checkov to check that the bucket is encrypted and has a lifecycle policy that deletes old versions and tflint to check that nothing uses deprecated syntax or known bad practices.
We do not want to give the agent raw access to the shell and all features of terraform, so we create a wrapper class that standardises how each shell command runs and how its output or failure is reported.
agent/tools/validators.py
class CommandResult(BaseModel):
success: bool
stdout: str
stderr: str
def format_error_for_agent(self) -> str:
"""The failure as the model sees it, capped. Callers keep the full output."""
if not self.success:
return f"failed:\n{_capped(self.stdout)}\n{_capped(self.stderr)}"
raise RuntimeError("CommandResult.format_error_for_agent() called on success")
class Command(BaseModel):
name: str
commands: list[str]
@property
def installs_providers(self) -> bool:
return "init" in self.commands
def run(self, path: Path) -> CommandResult:
if self.installs_providers:
with _INSTALL_LOCK:
return self._run(path)
return self._run(path)
def _run(self, path: Path) -> CommandResult:
# Inside the lock, so the span measures the command and not the wait.
with track_memory(self.name):
result = subprocess.run(self.commands, cwd=path, capture_output=True, text=True)
return CommandResult(
success=result.returncode == 0, stdout=result.stdout, stderr=result.stderr
)
def run_root(self, ctx: RunContext[WorkspaceDeps]) -> CommandResult:
return self.run(ctx.deps.root)
def run_in_tool(self, path: Path) -> str:
result = self.run(path)
if not result.success:
raise ModelRetry(f"{self.name} {result.format_error_for_agent()}")
return f"OK: {self.name} passed."
The first runs with the new tools went badly, and logfire showed it straight away. Token usage was rising quickly and I could see a lot of failed tool calls for checkov. Failed here means the checks did not pass, not that the tool crashed. A quick dive into the tool parameters showed that checkov was returning a very long report where:
- All checks, even the successful ones, were listed, filling the agent's context and confusing it
- The actual failed tests were extremely strict, stricter than you would expect in a normal setup
For example KMS over AES256 is a good rule, but irrelevant outside a compliance setting. Similarly, every S3 bucket is told it needs its own access log bucket. We disable a short list of rules, but only when the workspace ships no .checkov.yaml of its own.
So the fix? Add --quiet --compact to checkov, so we do not fill the agent's context with irrelevant information. I would rather read the short version as a human too. Together with excluding a few rules, the results improved immediately. We went from timing out after 300s and 2M tokens input to 94s and less than 20k tokens input. Significant cost difference!
A failed check is not paid for once, though. Its output joins the message history and is billed again on every later turn, so we also cap what a failure hands back:
agent/tools/validators.py
# Every failure report becomes a tool result the model carries for the rest of
# the run, so an uncapped one is paid for on every later turn and lands in the
# audit record. checkov's report is the offender; its head carries the
# failures that matter, and the agent re-runs the check anyway.
MAX_REPORT_CHARS = 8_000
def _capped(text: str) -> str:
if len(text) <= MAX_REPORT_CHARS:
return text
return f"{text[:MAX_REPORT_CHARS]}\n[{len(text) - MAX_REPORT_CHARS} more characters cut]"
agent/tools/validators.py
TERRAFORM_INIT = Command(
name="terraform_init",
commands=["terraform", "init", "-backend=false", "-input=false", "-no-color"],
)
TERRAFORM_VALIDATE = Command(
name="terraform_validate",
commands=["terraform", "validate", "-no-color"],
)
TERRAFORM_FMT = Command(
name="terraform_fmt",
commands=["terraform", "fmt", "-recursive"],
)
TFLINT = Command(
name="tflint",
commands=["tflint", "--format", "compact"],
)
# Checks skipped by default: cost or architecture posture decisions, not
# security baselines, and routinely disabled in real projects. Genuine
# baselines (public access blocks, encryption at rest, IAM wildcards) stay on.
_CHECKOV_DEFAULT_SKIPS = [
"CKV_AWS_18", # S3 access logging on every bucket
"CKV_AWS_144", # S3 cross-region replication
"CKV_AWS_145", # S3 must use KMS; SSE-S3 (CKV_AWS_19) still enforced
"CKV2_AWS_61", # S3 lifecycle configuration on every bucket
"CKV2_AWS_62", # S3 event notifications on every bucket
"CKV_AWS_50", # Lambda X-Ray tracing
"CKV_AWS_115", # Lambda reserved concurrency
"CKV_AWS_116", # Lambda dead-letter queue
"CKV_AWS_117", # Lambda attached to a VPC
"CKV_AWS_272", # Lambda code signing
"CKV_AWS_338", # CloudWatch log retention of at least a year
]
def _checkov_command(root: Path) -> Command:
"""Default skips apply only when the workspace brings no config of its own:
checkov auto-discovers .checkov.yaml in the scanned directory, and a project
that states its policy wins over ours.
"""
commands = ["checkov", "-d", ".", "--quiet", "--compact"]
if not any((root / name).exists() for name in (".checkov.yaml", ".checkov.yml")):
commands += ["--skip-check", ",".join(_CHECKOV_DEFAULT_SKIPS)]
return Command(name="checkov", commands=commands)
Wrapping them as tools is then trivial. The command objects never mutate their state, so holding them as module-level singletons carries no concurrency risk.
agent/tools/validators.py
def terraform_init(ctx: RunContext[WorkspaceDeps]) -> str:
"""Run ``terraform init`` in the workspace.
Required once before the first ``terraform_validate`` and again after
provider or module requirements change.
"""
return TERRAFORM_INIT.run_in_tool(ctx.deps.root)
def terraform_validate(ctx: RunContext[WorkspaceDeps]) -> str:
"""Run ``terraform validate`` in the workspace and return its output."""
return TERRAFORM_VALIDATE.run_in_tool(ctx.deps.root)
def tflint(ctx: RunContext[WorkspaceDeps]) -> str:
return TFLINT.run_in_tool(ctx.deps.root)
def checkov(ctx: RunContext[WorkspaceDeps]) -> str:
return _checkov_command(ctx.deps.root).run_in_tool(ctx.deps.root)
validate_workspace is different: we hook it up as an output validator. In the previous post we ran this check ourselves and passed the messages back into a new run with the retry prompt. Then I discovered pydantic-ai covers this out of the box, which let us delete most of that code.
agent/tools/validators.py
def validate_workspace(path: Path) -> CommandResult:
# Normalize formatting before gating; a fmt failure (unparsable HCL) is
# ignored here because terraform validate reports it better one step later.
TERRAFORM_FMT.run(path)
for command in [TERRAFORM_VALIDATE, TFLINT, _checkov_command(path)]:
result = command.run(path)
if not result.success:
break
return result
We extend the system prompt so the agent reaches for the new validators:
Run tflint and checkov and clear all findings before reporting done.
TaskResult: structured self-report
Making the agent declare what it did, knowing the claim gets checked, changes its behaviour more than I expected. So in this post we also added a structured output type:
agent/core.py
class TaskResult(BaseModel):
"""The agent's structured self-report on a finished run.
Each required field pushes the agent to consider that dimension of its
work; the tool-call spans in the trace remain the ground truth that
exposes any embellishment. Because this is the agent's output type,
ending a run now requires calling the final_result output tool, so a
text-only reply can no longer end a run silently.
"""
model_config = ConfigDict(frozen=True)
summary: str
"""One line on what was done."""
solution_description: str
"""How the problem was solved, including architectural choices."""
validations_run: list[str]
"""Which validation tools were invoked during the run."""
issues_addressed: list[str]
"""Security or correctness problems identified and fixed."""
known_limitations: list[str]
"""What was not handled; surfaces for human review."""
ready_for_review: bool
"""The agent's self-assessment that the workspace is PR-ready."""
Reporting done is not the same as being done, so the same validate_workspace gate runs as an output validator: the agent only escapes the loop when the checks it is told to run actually pass.
agent/core.py
_RETRY_PROMPT = (
"terraform validate still reports errors after you reported done. "
"Fix them and validate again.\n\n{output}"
)
@agent.output_validator
def _validate_final_workspace(ctx: RunContext[WorkspaceDeps], output: TaskResult) -> TaskResult:
# The deliverable is the workspace, not this object: a run that changed
# nothing produced nothing, however plausible its self-report reads.
if not ctx.deps.files_changed:
raise ModelRetry(
"You reported done but made no changes to the workspace. "
"Use the file tools to implement the request, validate, then report done again."
)
validation_result = validate_workspace(ctx.deps.root)
if not validation_result.success:
raise ModelRetry(_RETRY_PROMPT.format(output=validation_result.format_error_for_agent()))
return output
To be clear this does not entirely prevent the agent from making things up and filling things in it did not do, but it does push more runs the right way, because each output field forces the agent to account for that dimension.
In the eval runs I compared different models with and without the structured output. In particular weaker ones like Haiku (vs glm5p2) did better if they were forced to justify themselves in the structured output. glm5p2 showed no difference. For Haiku the structured output runs were cleaner with tool calls 27.0 down to 22.4, validator calls 13.0 down to 10.8, validate-pass 93% up to 100%, errored runs 6.7% down to 0%.
The output type itself is enforced, the agent has to call a tool with this enforced schema to end the run. We also have a max limit of turns and errors to avoid infinite unsuccessful runs.
This structured schema is also what will let us deterministically create PRs. This is what we set out to do in the first place after all.
The eval harness: the validators are the conditions
pydantic_evals is built around a few concepts:
- Datasets: contain cases for evaluation and which Evaluators to apply - the central abstraction validation runs against
- Cases: inputs, expected outputs, and metadata
- Evaluators: Logic to evaluate which can be based on expected outputs and metadata
It offers a few built-in evaluators, but the more complex checks we need, actually inspecting the code the agent produced, required building our own.
-
WorkspaceValidates- uses the existingvalidate_workspaceoutput validator to check the workspace is valid. Technically a slight double validation as the agent cannot submit the result without this. -
PlanMatchesGraph- runsterraform plan -outandterraform show -jsonto compare the workspace to the expected graph (coming from metadata). Configuring the AWS provider authenticates against STS, so the plan would need real credentials. It runs on a copy of the workspace with a*_override.tfthat supplies mock ones, which keeps the sweep offline and leaves the graded workspace untouched. -
SelfReportAccurate- the agent has to say which checks it did run in itsTaskResult, we validate this against actual tool calls by going through recorded spans. -
ToolErrorMetricEvaluator- slightly different from the others: here we give a numeric score from 0.0 to 1.0 to indicate how many tool calls failed. Every time a tool call fails 0.1 is subtracted from the score until it reaches 0.0.
One slight practical hiccup was when the agent decided to add variables without a default, which made it difficult to create a plan. Therefore we instruct the agent to default every variable it declares and fail the evaluation if it does not.
WorkspaceValidates is the one place the policy tools and the harness meet, in eight lines: it calls the same validate_workspace() the agent had to satisfy at runtime, so a case clears that check for the same reason the agent was allowed to stop. The other three evaluators ask questions the validators cannot answer.
evals/evaluators.py
@dataclass
class WorkspaceValidates(Evaluator[TASK_INPUTS, EvalOutput, CaseMetadata]):
"""fmt + terraform validate + tflint + checkov, verbatim from the agent."""
def evaluate(
self, ctx: EvaluatorContext[TASK_INPUTS, EvalOutput, CaseMetadata]
) -> EvaluationReason:
result = validate_workspace(ctx.output.workspace)
if result.success:
return EvaluationReason(value=True)
return EvaluationReason(value=False, reason=result.format_error_for_agent())
Take SelfReportAccurate, one of the more interesting evaluators here: you can see that the context passed to the evaluator contains a ctx.span_tree object with functions to find spans we need for evals. The downside is that it does not carry non-genai attributes, which is a problem if you are looking for error information that is recorded in an OTEL-specific field. Not a problem here, but was an issue for ToolErrorMetricEvaluator as the errors were harder to identify consistently.
evals/evaluators.py
@dataclass
class SelfReportAccurate(Evaluator[TASK_INPUTS, EvalOutput, CaseMetadata]):
"""The TaskResult agrees with what the case knows.
The tool-call spans in the trace stay the ground truth. This checks only the
self-report claims that a case can verify cheaply.
"""
def evaluate(
self, ctx: EvaluatorContext[TASK_INPUTS, EvalOutput, CaseMetadata]
) -> dict[str, EvaluatorOutput]:
report = ctx.output.result
validations = " ".join(report.validations_run).lower()
tool_spans = ctx.span_tree.find(
predicate=SpanQuery(
name_contains="execute_tool", has_attribute_keys=["gen_ai.tool.name"]
)
)
actual_tools = {span.attributes.get("gen_ai.tool.name") for span in tool_spans}
return {
"ready_for_review": EvaluationReason(
value=report.ready_for_review,
reason="Agent should mark ready for review.",
),
"reported_linters": EvaluationReason(
value=(
"tflint" in validations
and "checkov" in validations
and "tflint" in actual_tools
and "checkov" in actual_tools
),
reason=(
f"Mismatch in validations: {validations} and {actual_tools}, "
"should both contain tflint and checkov."
),
),
}
Running the sweep
Now you can run all eval cases across the full suite of models. The command below will run 5 models and repeat each model three times. LLMs are non-deterministic, so three repeats give enough spread to tell a real failure from a bad roll.
Sweep the registry
uv run python -m evals.run \
--model glm5p2 --model haiku --model mistral-large \
--model mistral-medium --model codestral --repeat 3
You get one experiment per model, each with its own pass rate.
Once you open a particular eval run you can see more detailed results, including our assertion/boolean validators and the tool error metric score. Further down are operational metrics like cost and average task duration.
You can also interact with the logfire MCP to create ad-hoc summary tables, or, if you will run it repeatedly, have a model write a script against the API. That is where the numbers at the top of this post come from.
End state
In this post we gave the agent a stricter definition of correct Terraform. The same fmt, validate, tflint and checkov steer it while it works and gate what it submits. None of that says whether it did the task it was asked to do, which is the question the eval harness answers.
We implemented an eval harness that lets us weigh different models against each other and allows us to make a choice based on numbers instead of gut-feel.
In the next post we will integrate with github and ship the PR flow, that means human in the loop via github and deciding how we manage state across the whole PR lifecycle.






Top comments (0)