Contracts and validators

A contract is a file plus the schema it must satisfy and the validators it must pass. An agent task’s output contract comes from agent_task(output=..., validators=[...]); a Python task’s from its return annotation.

Output models

Outputs are pydantic models. Define them at the top level of an importable module, not in the script you run and not inside a function: the check inside the agent’s session imports them (contraagent check says when it cannot). Their field descriptions reach the agent in the call file’s JSON Schema. Each model’s examples (its example) is what a stub run writes in place of the agent; give it one that passes the validators (Checking and testing).

class Prepared(BaseModel):
    model_config = ConfigDict(json_schema_extra={"examples": [{"unit": "alpha", "path": "inputs/alpha.md", "words": 12}]})

    unit: str
    path: str = Field(description="the document to summarize, relative to the working directory")
    words: int


class Summary(BaseModel):
    model_config = ConfigDict(
        json_schema_extra={"examples": [{"unit": "alpha", "points": ["parses a config file", "validates it", "reports errors"]}]}
    )

    unit: str = Field(description="the unit this summary is for, copied from the call file")
    points: list[str] = Field(description="the main points, one short sentence each")


class Score(BaseModel):
    model_config = ConfigDict(json_schema_extra={"examples": [{"points": 3, "words": 8}]})

    points: int
    words: int


class Verdict(BaseModel):
    model_config = ConfigDict(json_schema_extra={"examples": [{"approved": True, "note": "Accurate."}]})

    approved: bool
    note: str = Field(description="one sentence: what to change, when not approved")


#: A text output with an example: the note starts with a heading and names its unit.
Note = Annotated[str, Field(examples=["# alpha\n\nThe alpha component parses a config file.\n"])]

Rows

An output may be a list that grows as the agent works: the findings of a review, the items of a long analysis. output=Rows[Finding] declares a JSON Lines file whose every line is one Finding, a row. The prompt asks the agent to add each row to the file as soon as it has it, one line at a time, and the call file says the format (output.format is jsonl, and output.json_schema is the schema of one row). The file is {kind}/{unit}.jsonl by default ({kind}/{kind}.jsonl for a task with no unit).

class Finding(BaseModel):
    model_config = ConfigDict(json_schema_extra={"examples": [{"document": "docs/{unit}.md", "line": 1, "note": "a broken link"}]})

    document: str = Field(description="the document's path")
    line: int = Field(ge=1, description="the line the finding is on")
    note: str = Field(description="what is wrong, in one sentence")


class Tally(BaseModel):
    findings: int
    documents: int


def distinct(path: Path, findings: Rows[Finding]) -> list[str]:
    """No two findings on the same line of the same document."""
    seen: set[tuple[str, int]] = set()
    errors = []
    for n, f in enumerate(findings, 1):
        if (f.document, f.line) in seen:
            errors.append(f"row {n}: {f.document}:{f.line} has a finding already; give one finding per line")
        seen.add((f.document, f.line))
    return errors


review = agent_task(
    "review",
    instructions="Review every document under docs/ for unit $unit, and write one finding per problem you find.",
    output=Rows[Finding],
    validators=[distinct],
    partial_on=["timeout", "stalled"],
    max_turns=200,
    timeout_s=3600,
    stall_s=600,
)

Each line is checked against the row’s model. A line that does not parse, or does not meet the model, fails the whole output, and its errors name the line: line 3: line: Input should be greater than or equal to 1 (got: 0). The correction round sends those errors back, so the agent corrects the lines named and keeps the rest. A file that is one JSON array is named as such. A blank line is no row, and a file with no row is an empty list.

As a value, rows are a Rows, a list of the models in the file’s order. A validator is given that list, the flow receives it, and a task it is passed to gets it: a Python task as a list of models (list[Finding]), an agent task as a list of JSON objects in its call file. The example a stub run writes is one row, the example of the row’s model. A Python task may return rows too (-> Rows[Finding], returning Rows([...])). preserve and unwrap take a JSON document, not rows.

A launch stopped for its time

The runner stops a launch past its timeout_s (timeout) or silent for its stall_s (stalled), and the task fails, whatever the agent had written. A long task often has written something useful by then. partial_on names the stops after which it is kept: agent_task(partial_on=["timeout", "stalled"]), or partial_on in [tasks.<kind>].

After such a stop, the output is checked as after any round: it must exist, be written by this attempt (a file from an earlier run does not count), and pass its schema and every validator. When it passes, the task’s result is partial, as when a Python task returns Partial (Partial results): ok, stop partial, its reason the launch’s stop (timeout, stalled), and the flow receives the output. The hop keeps the launch’s own stop. A resume runs the task again, with the tasks that wait for it. When the output fails, the task fails on the stop as before; a stopped launch is never sent a correction round. A test plays such a launch on the stub agent with writes(value, stop=Stop.TIMEOUT) (Stop from contraagent.backends; Checking and testing).

With rows, the rows written so far are the output. A last line with no newline after it that is not valid JSON was being written when the launch was stopped: the runner removes that line before the check.

review = agent_task(
    "review",
    instructions="Review every document under docs/ for unit $unit, and write one finding per problem you find.",
    output=Rows[Finding],
    validators=[distinct],
    partial_on=["timeout", "stalled"],
    max_turns=200,
    timeout_s=3600,
    stall_s=600,
)


@python_task
def tally(unit: str, findings: list[Finding]) -> Tally:
    """How many findings, on how many documents."""
    return Tally(findings=len(findings), documents=len({f.document for f in findings}))

Only these two stops are kept. Both are the runner’s own: it ended the launch for time, and the agent said nothing about its work. A turn cap (max_turns), a cost cap (budget) or a context cap (context_budget) is a limit a pipeline sets to bound a launch’s spend, and a refusal, an API error or a refused credential says the work went wrong; those fail the task as before. An output in the final message does not exist after a stopped launch, so partial_on takes an output the agent writes to a file; contraagent check reports a kind whose config sets it for one that does not.

Checked twice, by the same rules

Inside the session, on Claude Code, Codex and omp. The plugin’s stop hook runs contraagent call validate, which reads the call file ($CONTRAAGENT_CALL), imports the output model and runs the command validators. A failure keeps the agent working in the same turn. opencode and pi have no stop hook that continues a session (Coding agents): for them the runner’s check below is the first one, and its errors come back in a correction round.
By the runner, after the turn. The schema, freshness (a file left by an earlier run does not pass), the parts to keep (preserve, below) and every validator. A failure resumes the same session with the errors as the next turn, until the file passes or the correction rounds are spent (invalid_output).

Each launch of the session is a round, and each resume a correction round. A kind has 3 correction rounds after its first round by default. max_corrections sets another number, from 0 to 20: agent_task(max_corrections=1), or max_corrections in [tasks.<kind>], which outranks the declaration (routing). With 0 the first failed check ends the task invalid_output. A short output, such as a verdict, rarely needs more than one round to correct its form; an output that is still wrong after many rounds is not getting closer, and each round is a full turn of a session whose context grows. contraagent check reports a value out of range.

The journal records how many rounds a task took (rounds), how many correction rounds its kind allowed (max_corrections) and the errors each resume was sent (round_errors), and the monitor shows each round of its transcript. The run’s progress line for the attempt names the first round’s first error. The attempt’s cost is its rounds’ cost, each round counted once: Claude Code reports a resumed session’s cost so far, omp each launch’s own (Backend.cost_is_session_total), and launches.jsonl keeps each round’s own (round_costs).

An output in the final message

A task that needs no tool (a verdict, a classification) may give its output as its final message: agent_task(..., output_from="message"). The call file says so (output.from), the agent ends with the output as its whole message and writes no file, and the runner writes that message to the output file and checks it there, by the same schema and validators. A message that fails its checks goes back to the same session as the next round, asking for the corrected output as the reply. The stop hook leaves it to the runner: the message exists only after the turn. It saves the turn a Write takes, and the permission to write.

By default the whole message must be the document. A model that is told so still puts a sentence before it now and then, or a fence around it, and each time costs a round. unwrap=True takes the document out of the message instead: the last fenced block that parses, else the longest JSON object or array in the message. The document is the output file, the whole message is kept beside it (<output>.message.txt), and the run’s progress line and launches.jsonl (unwrapped) say that it was taken out. A message that holds no document is checked as it is, and goes back as before.

# A verdict needs no tool: the reviewer's final message is its output (no turn spent on a Write).
# A verdict is short: one correction round is enough to fix its form (default: 3).
review = agent_task(
    "review",
    instructions="Review params.summary.",
    output=Verdict,
    output_from="message",
    max_corrections=1,
    max_turns=3,
    timeout_s=300,
)

Parts of a file to keep

A task may fill in a file that holds values already: a Python task writes a report’s items, and an agent task writes their summaries into the same file. output_of names the task declaration whose file it writes: for each task, the file of the latest task of that kind created so far for the same unit (or key=), else the file that kind’s output_name names for it. The task waits for the latest task created so far that writes that file, of any kind, so two never write it at once. output is the schema of the whole file after the task. preserve names the parts the agent must keep as they were, as JSON pointers (RFC 6901: /source, /items/0/id), with one extension: a segment * is every member of an array or object, so /items/*/id is the id of every item.

class Item(BaseModel):
    id: str
    title: str
    summary: str = Field(default="", description="two sentences on the item; empty until it is filled in")


class Report(BaseModel):
    source: str = Field(description="the document the items come from, relative to the working directory")
    items: list[Item]


@python_task
def outline(unit: str, titles: list[str]) -> Report:
    """The report's skeleton: an item per title, each with an id of its own."""
    return Report(source=f"inputs/{unit}.md", items=[Item(id=f"{unit}-{i}", title=t) for i, t in enumerate(titles, 1)])


fill = agent_task(
    "fill",
    instructions="Write the summary of each item of the report from its source document.",
    output=Report,
    output_of=outline,  # the file outline wrote for the same unit
    preserve=["/source", "/items/*/id", "/items/*/title"],
    max_turns=20,
    timeout_s=600,
)


def reports(units: dict[str, list[str]]):
    for unit, titles in units.items():
        outline(unit, titles=titles)
        fill(unit)  # writes outline's file for the unit, after outline

Before the first round the runner reads the file and takes the values the pointers select in it. After each round, each of them must still be in the file, equal; one that is not is an error like any other check’s, and goes back to the same session: /items/1/id changed: it must stay "alpha-2" (now "alpha-9"), /source removed: it must stay "inputs/alpha.md". A pointer selects by position, so an agent that reorders a list changes the values at its indexes. An attempt that does not pass puts the file back as it was, so the next attempt starts from the same values. A file that is missing before the launch, or that does not parse, keeps nothing.

The call file lists the pointers (output.preserve), and the prompt tells the agent to read the file first. The stop hook does not judge them: it sees the file after the turn only. preserve takes a JSON output the agent writes, not a text output nor one in the final message, and each pointer is checked against the output’s schema when the task is declared: /sorce is refused there.

A judgement in one turn: a light launch

Such a task still starts with all that a coding agent loads before its first turn: its own system prompt and the definitions of its tools, and it spends a turn reading its call file. A task declared with no tools, tools=(), is launched light instead:

its role’s text without the frontmatter (for a task with no plugin, its instructions) replaces the coding agent’s system prompt;
the agent gets no tool, and no plugin, settings file, skill, hook or MCP server is loaded;
the prompt holds the call itself: the params as JSON and the output’s JSON Schema (the call file is still written, for the record);
its final message is its output, so it needs output_from="message"; a message that fails its checks goes back to the same session, as above.

class Classified(BaseModel):
    category: Literal["security", "product", "other"]
    promo: bool = Field(description="true when the item advertises a product or a service")


classify = agent_task(
    "classify",
    instructions="You classify one news item: its category, and whether it is promotional.",
    output=Classified,
    output_from="message",
    tools=(),
    max_turns=2,
    timeout_s=120,
)

A light launch is one call, and a small one. Measured on Haiku with a one-line judgement:

	light launch	the same judgement, the agent’s defaults	the same task as its role, with tools
Claude Code	1 call, 644 tokens of input (1,180 with a role and a schema)	1 call, 16,863 (25 tools)	2 calls, 2,230 each: `--agent` already puts the role in place of the default system prompt, and `tools: Read` keeps one tool
omp	1 call, 369 (658)	1 call, 11,828	2 calls, 12,311 each: omp keeps its own system prompt and every tool

The journal row holds the first call’s input (usage.resident), and the monitor shows it on the tokens of each attempt. Each coding agent does it with its own flags:

	system prompt	tools	settings, skills, MCP servers
Claude Code	`--system-prompt`	`--tools ""`	`--setting-sources ""` and `--strict-mcp-config`, as on every launch; no `--plugin-dir`, no `--agent`
omp	`--system-prompt`; omp still appends its environment footer (the working directory, the system)	`--no-tools --no-lsp`	the isolation overlay every launch has (no context files, skills, rules or MCP servers); no `--plugin-dir`
pi	`--system-prompt`	`--no-tools`	the flags every launch has (`--no-extensions`, `--no-skills`, `--no-context-files`, …); no `--skill`, no `-e`
opencode	an agent of its own (`--agent`) whose `prompt` it is; opencode’s free models refuse such a session	that agent’s `permission` denies every tool	the user’s opencode config is not read either (an empty `XDG_CONFIG_HOME`: with no tool, nothing else reads it); no plugin

Codex runs no light launch: a coding agent that cannot launch so is refused for a kind with tools=(), on every route of the kind, failover hops included (Backend.light_launch; Coding agents). The agent can read nothing and run nothing: give it all it needs in its params. A list of tool names is not taken, since each coding agent names its tools its own way; a role names the ones it may use (tools:), as plugins says.

Validators

A validator is any callable (path, value) -> errors. value is the checked content (the model instance, or the text of a text output); an empty result passes. Its errors go back to the agent exactly like schema errors, so they name what is wrong and what is expected.

def three_points(path: Path, summary: Summary):
    """A structural check the schema cannot express."""
    if len(summary.points) != 3:
        return [f"points: exactly three expected, got {len(summary.points)}"]

A rule that relates the output to what the task was given takes the task’s params as the keyword argument params (an agent task’s as its call file holds them; a Python task’s with every upstream value resolved). A validator that declares call is given the path of an agent task’s call file.

A Python validator runs in the runner’s process, in its current directory, for an agent task and a Python task alike: a relative path it opens (one from params) resolves there, not against Pipeline(cwd=), which is where the coding agents and CommandValidators run. contraagent run runs both in the directory it is started in; a test that passes Pipeline(cwd=project) runs the Python code in pytest’s directory, so it calls monkeypatch.chdir(project) too.

def its_own_unit(path: Path, summary: Summary, *, params: dict):
    """A check against the task's own params: the summary names the unit it was given."""
    if summary.unit != params["prepared"]["unit"]:
        return [f"unit: {params['prepared']['unit']!r} expected (the call file's params.prepared.unit)"]

A rule about what the task changed takes before: the file as it was before the task, parsed as value is (the model instance, or the text), or None when there was no file or it did not meet its schema then. It is the file ahead of the task’s first round, on every correction round alike, so a round is judged by all that the task changed, not by what it changed since the last round. The runner reads the file ahead of the task only for a contract with such a validator, and an attempt that does not pass puts the file back as it was, so the next attempt has the same before. A Python task’s validators get the same: the file before its function was called.

def adds_an_idea(path: Path, backlog: Backlog, *, before: Backlog | None):
    """At least one idea more than the file held before the task, and none of those removed."""
    if before is None:
        return None  # no backlog before the task: any backlog is an addition
    gone = [idea for idea in before.ideas if idea not in backlog.ideas]
    errors = [f"ideas: {idea!r} was removed; keep every idea the file held" for idea in gone]
    if len(backlog.ideas) <= len(before.ideas):
        errors.append(f"ideas: add at least one idea to the {len(before.ideas)} the file held")
    return errors


grow = agent_task(
    "grow",
    instructions="Add one new idea to the topic's backlog.",
    output=Backlog,
    output_of=start,  # the file start wrote for the same unit
    validators=[adds_an_idea],
    max_turns=10,
    timeout_s=300,
)

CommandValidator turns an external program (textlint, a structure checker) into one: exit 0 passes, anything else fails with the program’s output as the errors. {path} in its arguments is the output file and {call} the call file, which holds the params: a script given {call} checks the output against the input, inside the agent’s session too. It runs in the task’s working directory, in the runner and in the hook alike, and it runs even when the schema check fails, so one round reports every error.

#: A linter for the Japanese note, as an external command (`textlint` in a real setup).
has_heading = CommandValidator(
    [
        sys.executable,
        "-c",
        "import sys; t = open(sys.argv[1], encoding='utf-8').read(); "
        "sys.exit(None if t.startswith('# ') else '1:1 the note must start with a # heading')",
        "{path}",
    ],
    name="heading",
)

Given {call}, a command checks the output against the input, and the agent runs it before it stops:

#: A rule about the output and the input, as a command given the call file: the agent's own stop hook runs it too.
names_its_unit = CommandValidator(
    [sys.executable, "-m", "contraagent.docs.examples.demo.checks", "{path}", "{call}"],
    name="names-its-unit",
)

"""A check run as a command: the output file and the task's call file in, exit 0 to pass.

    python -m contraagent.docs.examples.demo.checks <note.md> <call.json>

What it prints when it fails is what the agent is sent.
"""

from __future__ import annotations

import json
import sys
from pathlib import Path


def main(argv: list[str] | None = None) -> int:
    note, call = argv if argv is not None else sys.argv[1:]
    unit = json.loads(Path(call).read_text(encoding="utf-8"))["params"]["summary"]["unit"]
    if unit not in Path(note).read_text(encoding="utf-8"):
        print(f"the note never names its unit {unit!r} (the call file's params.summary.unit)")
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

	Python function	`CommandValidator`
Runs in the runner	yes	yes
Runs inside the agent’s session	no	yes: it travels in the call file
Takes `before`	yes	no: it reads the file as it is

Inputs

A task that receives another task’s result (a Pending) reads that task’s file as an input contract: checked before launch, and a task whose input fails is refused (invalid_input) rather than run on nothing. An input fails when its file is missing or off its schema, or when the task that wrote it did not succeed: an output its validators rejected (invalid_output) meets its schema and is still refused, its reason input from <key> ended <stop>. A partial result succeeded: its output is read.