four

Four functions compose. The loop is the evaluator.

An agent that runs bash. A generator that produces notebooks. A tool that writes Python. A supervisor that rescues the agent. Same loop. Different functions.

from four import run, litellm_invoke, regex_parse, local_env, save_trajectory

run(
    G=litellm_invoke("anthropic/claude-sonnet-4-5-20250929"),
    V1=regex_parse(),
    V2=local_env(),
    emit=save_trajectory(),
    system="You are a helpful assistant that executes bash commands.",
    prompt="Find all Python files in /tmp and count lines in each",
    max_steps=50,
)

The algebra

invoke   : G   -- messages → Result[raw]
parse    : V1  -- raw → Result[list[action]]
validate : V2  -- action → Result[observation | Exit]
emit     : IO  -- (messages, outcome) → Path

The loop chains them: (G → V1 → [V2, V2, ...])* → emit

Each step: G queries the model, V1 extracts all actions, V2 executes each one. If V1 fails, the error becomes a user message and the loop continues — the model sees its mistake and self-corrects on the next turn. Four functions compose.

The loop

The entire evaluator, verbatim from src/four/core.py:

def run(G, V1, V2, emit, system, prompt, max_steps=100, max_format_errors=3):
    messages = [
        {"role": "system", "content": system},
        {"role": "user", "content": prompt},
    ]

    consecutive_format_errors = 0

    for step in range(max_steps):
        # G: invoke
        raw = G(messages)
        if isinstance(raw, Err):
            return emit(messages, f"model_error: {raw.error}")

        # Append assistant message (for chat variant to see its own output)
        messages.append({"role": "assistant", "content": raw.value})

        # V1: parse
        actions = V1(raw.value)
        if isinstance(actions, Err):
            # exit:* signals are terminal, not format errors
            if actions.error.startswith("exit:"):
                return emit(messages, actions.error)
            consecutive_format_errors += 1
            if 0 < max_format_errors <= consecutive_format_errors:
                return emit(messages, f"repeated_format_error: {actions.error}")
            messages.append({
                "role": "user",
                "content": f"Format error: {actions.error}. Please respond with exactly one bash command in the expected format.",
            })
            continue

        consecutive_format_errors = 0

        # V2: validate / execute each action
        for action in actions.value:
            command = action["command"] if isinstance(action, dict) else action
            tool_call_id = action.get("tool_call_id") if isinstance(action, dict) else None
            result = V2(command)
            if isinstance(result, Err):
                return emit(messages, result.error)
            observation = result.value
            if tool_call_id:
                observation["tool_call_id"] = tool_call_id
            messages.append(observation)

    return emit(messages, "max_steps_reached")

That's the whole thing. 47 lines, no framework. No config files. No YAML. No SDK. No Pydantic models. No Jinja2 templates baked into the code. Four functions that take and return well-typed values, chained in a loop. Read it once and note the exits: the model fails (model_error), the model is done (exit:* from V1 or V2), or the budget runs out (max_steps_reached). Termination is a value, not an exception.

The test suite runs this loop with mock G functions — 41 tests, no LLM required.

Why four?

Four is the minimum. Remove any one and the loop breaks:

• No G → nothing to evaluate
• No V1 → can't extract actions from raw text
• No V2 → can't execute or observe
• No emit → can't persist results

Format recovery is built into the loop — no separate function needed.

What it replaces

mini-swe-agent	four
YAML config with 40+ parameters	Four function arguments
Pydantic model configs	Plain functions
Jinja2 templates in config	Templates passed as strings
FormatError + InterruptAgentFlow hierarchy	Ok | Err
Inner retry loop for format errors	Format error as user message, outer loop continues
1000+ lines of boilerplate	One 47-line function

Same capability. Different shape.

The components

G — invoke. Queries the LLM. Returns Ok(text) or Err(reason).

• litellm_invoke() — plain text with markdown code blocks
• litellm_toolcall_invoke() — structured tool calls
• http_response_invoke() — Responses API
• context_aware_invoke() — fast model for lean turns, escalates to a large-context model past a token limit
• summarizing_invoke() — progressively summarizes history instead of truncating it
• retry_invoke(fn) — wraps any G: exponential backoff on transient errors, immediate abort on auth errors

V1 — parse. Extracts actions from raw output. Returns Ok(list[action]) or Err(reason).

• regex_parse() — fenced code blocks (```mswea_bash_command, ```bash, ```sh) plus <code> tags; returns all matches. Plain text with no block means the model is finished → exit:task_complete
• toolcall_parse() — JSON tool-call payloads → list of {command, tool_call_id}

V2 — validate. Executes each action. Returns Ok(observation) or Err(exit).

• local_env() — subprocess execution with output truncation and exit-signal detection
• super_env() — same, with larger limits for long-running work

emit — IO. Saves the trajectory. Returns Path.

• save_trajectory() — JSON files with outcome and full message history

Why "algebra"?

A fair question, and the right one to ask first. Here is the claim, precisely:

• Ok/Err is the error monad. G, V1, V2 are Kleisli arrows over it: each takes a value and returns a computation that may fail. Binding is: feed the value forward, or short-circuit on Err.
• run is a fixed-point combinator. The star in (G → V1 → V2*)* is the Kleene star — the loop computes the least fixed point of the step function on the message state. Each pass is an approximation; max_steps is where the approximation sequence is truncated; Err is where the step maps into a sink. That's why termination is a value: the absorbing state, not the exception.
• The evaluator is universal: it doesn't know what it's evaluating. It sees the shapes of the values flowing through it and nothing else. Swap V1/V2/emit and it's a different machine; the loop is unchanged.
• And it is closed under self-application. The package ships generators (src/four/generators/) that emit new spoke programs — and each generated program is the same 47-line loop with new functions. The system can evaluate itself.

That last line is not a metaphor. It's why the next section exists.

What this built

The loop is not a demo framework. It is a production builder, and the toolchain around it is its output. Every repository below was produced by a pipeline run of the loop — not hand-written:

Repo	What the loop built
mission-compiler	free-text mission → complete validated launch
spoke-lint	static validator: prompt's spoke invocations vs real argparse
loop-doctor	pre-launch readiness auditor, GO/NO-GO
launch-gate	launch-moment gate: redirect safety, endpoint contention, wall sizing
fourseer	loop intelligence: per-cycle metrics, failure taxonomy, plan drift
sentry	deterministic rescue supervisor: detects driver death, wall-kills, stalls; relaunches or kills
fleet	portfolio scanner: one-page status over every project

The first four gate a launch. The last three observe and rescue a run. The same algebra, applied to the loop itself.

The load-bearing line: the model is the only non-deterministic component in the system. Everything that verifies a run, observes it, and rescues it is deterministic code reading the artifacts the loop writes (trajectories, logs, git history). The loop built its own observer and its own supervisor.

The same algebra, different domains:

Domain	V2 validates	emit produces
Notebooks	AST + chart execution	.ipynb with embedded PNGs
Agents	bash execution	JSON trajectory
Python code	type checking	.py files
CLI tools	compilation	binary + man page
Observers	artifact parsing	metrics + taxonomy (fourseer)
Supervisors	process state	rescue actions (sentry)

The loop doesn't know what it's evaluating. It only chains Result types.

Evidence you can check right now:

• examples/agents/*.report.json — real trajectories from live runs, checked into this repo.
• This repo's own git history — the loop's experiments on itself: a text-transform pipeline, a defrag of the delivery path, a dual stream/PR pipeline. One commit is just the word "sorry."
• The per-project build records (how each of the seven repos above was produced) are held privately by the operator. The artifacts are the repos.

Extending

Every component is swappable. The loop doesn't care:

# Tool-calling instead of regex
G=litellm_toolcall_invoke("openai/gpt-4o"),
V1=toolcall_parse(),

# Two-model escalation: fast for lean turns, deep for large context
G=context_aware_invoke(
    fast_model="openai/fast-qwen",
    large_model="openai/qwen",
    context_limit=50_000,
),

# Summarize history instead of truncating
G=summarizing_invoke(litellm_invoke("openai/gpt-4o"),
                     summarize_model="openai/fast-qwen"),

# Retry on transient errors
G=retry_invoke(litellm_invoke("openai/gpt-4o")),

# Larger limits for long-running work
V2=super_env(),

# Abort after 2 format errors instead of 3
max_format_errors=2,

Philosophy

The framework doesn't call itself category theory. It calls itself algebra. Four functions compose; the loop is the evaluator — a fixed point, and a system that can evaluate itself.

This repository is the algebra, complete and self-contained. The production drivers that run it against real projects are private; the artifacts they produce are the repos above.

#agenticcoding #functional-programming #python #llm #agents #monads #fixed-point-combinator