Stop Sending Every Agent Turn to the Frontier Model

#!/usr/bin/env python3
"""route_turn: pick a tier for one agent turn from mechanical signals only.

Default is the Workhorse tier. The router escalates to the Frontier tier when
  (a) the turn is a plan or decompose step,
  (b) the pre-commit gate returned exit 1 twice for the same file,
  (c) the diff changes a public signature (AST check, not a regex),
  (d) the Workhorse output failed schema validation.
Escalations are capped per trajectory, a file is pinned to the Frontier tier
after two repairs, and every decision is logged with its reason.
Standard library only.
"""
import ast
import json
from dataclasses import dataclass, field

WORKHORSE = "workhorse"
FRONTIER = "frontier"
ESCALATION_CAP = 8        # per trajectory; 20 percent of a 40-turn run
REPAIRS_BEFORE_PIN = 2    # after this many gate failures a file stays on Frontier


@dataclass
class State:
    escalations: int = 0
    gate_failures: dict = field(default_factory=dict)   # path -> consecutive exit-1 count
    pinned: set = field(default_factory=set)             # paths pinned to the Frontier tier
    log: list = field(default_factory=list)              # one dict per routed turn


def public_signatures(source: str) -> dict:
    """Name -> signature for every top-level def or class not starting with '_'."""
    out = {}
    for node in ast.parse(source).body:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and not node.name.startswith("_"):
            out[node.name] = ast.dump(node.args)
        elif isinstance(node, ast.ClassDef) and not node.name.startswith("_"):
            out[node.name] = "class"
    return out


def public_exports_changed(before: str, after: str) -> bool:
    return public_signatures(before) != public_signatures(after)


def route_turn(turn: dict, state: State) -> str:
    """turn = {"n", "kind", "path"?, "before"?, "after"?, "schema_failed"?}."""
    kind, path = turn["kind"], turn.get("path")
    tier, reason = WORKHORSE, "default"

    if kind == "plan":
        tier, reason = FRONTIER, "plan step"
    elif path in state.pinned:
        tier, reason = FRONTIER, "file pinned after repeated repairs"
    elif kind == "repair" and state.gate_failures.get(path, 0) >= 2:
        tier, reason = FRONTIER, "gate exit 1 twice on same file"
    elif kind == "edit" and turn.get("before") is not None \
            and public_exports_changed(turn["before"], turn["after"]):
        tier, reason = FRONTIER, "diff touches public export"
    elif turn.get("schema_failed"):
        tier, reason = FRONTIER, "workhorse output failed schema"

    if tier == FRONTIER and state.escalations >= ESCALATION_CAP:
        tier, reason = WORKHORSE, "cap %d reached, stayed on workhorse" % ESCALATION_CAP
    if tier == FRONTIER:
        state.escalations += 1

    state.log.append({"turn": turn["n"], "kind": kind, "tier": tier, "reason": reason})
    return tier


def observe_gate(turn: dict, state: State, exit_code: int) -> None:
    """Feed a gate result back so the next repair on this file can escalate."""
    path = turn.get("path")
    if exit_code == 0:
        state.gate_failures.pop(path, None)
        return
    state.gate_failures[path] = state.gate_failures.get(path, 0) + 1
    if state.gate_failures[path] >= REPAIRS_BEFORE_PIN:
        state.pinned.add(path)


def dump_log(state: State) -> str:
    return "\n".join(json.dumps(entry, separators=(",", ":")) for entry in state.log)

{"turn":1,"kind":"plan","tier":"frontier","reason":"plan step"}
{"turn":2,"kind":"read","tier":"workhorse","reason":"default"}
{"turn":18,"kind":"edit","tier":"frontier","reason":"diff touches public export"}
{"turn":32,"kind":"repair","tier":"frontier","reason":"file pinned after repeated repairs"}
{"turn":38,"kind":"edit","tier":"workhorse","reason":"cap 8 reached, stayed on workhorse"}