An agent is an HTTP call in a loop. Everything else is the harness.

def wrap_email(email: str) -> dict:
    """Job 4.1: the email enters as the first user message, fenced so it reads as data."""
    return {"role": "user", "content": [{"type": "text", "text": f"New inbound email.\n<email>\n{email}\n</email>"}]}


def assemble_context(caller: Caller, state: dict, model: str) -> dict:
    """Job 4.1: build the full request body. The API renders tools, then system, then messages."""
    cache = {"type": "ephemeral"}  # a prompt-cache breakpoint: everything up to this block is the cached prefix
    system = [
        {"type": "text", "text": SYSTEM_PROMPT},                                         # 1. same for every run
        {"type": "text", "text": ROLE_INSTRUCTIONS[caller.role], "cache_control": cache},  # 2. same per role
        {"type": "text", "text": "Underwriting rules:\n" + retrieve_rules(caller, state["email"]),
         "cache_control": cache},                                                        # 3. same within a run
    ]
    return {"model": model, "max_tokens": MAX_TOKENS, "system": system,
            "tools": tool_schemas_for(caller),  # stable order, so the prefix stays byte-identical
            "messages": state["messages"]}      # 4. the history: it changes every step, so it comes last

def run_tool_guarded(name: str, args: dict, timeout_s: float = 2.0) -> tuple[str, bool]:
    """Job 4.2: run one tool behind two guards. Returns (result text, is_error).

    The guards are an allowlist and a wait limit. This is NOT a sandbox: the tool runs in our process,
    and a call that times out is abandoned, not killed. Real systems run tools in a container or microVM.
    """
    if name not in TOOL_ALLOWLIST:
        return f"blocked: {name} is not on the tool allowlist", True
    pool = ThreadPoolExecutor(max_workers=1)
    try:
        return json.dumps(pool.submit(TOOL_ALLOWLIST[name], **args).result(timeout=timeout_s)), False
    except TimeoutError:
        return f"{name} timed out after {timeout_s}s", True
    except Exception as error:  # e.g. the model sent wrong arguments; tell it, so it can correct itself
        return f"{name} failed: {error}", True
    finally:
        pool.shutdown(wait=False)

def run(db: sqlite3.Connection, model, state: dict, *, approve: bool = False,
        crash_after: int | None = None, verbose: bool = False) -> dict:
    """Job 4.3: the loop. One step is one model call or one batch of tool calls."""
    caller = Caller(**state["caller"])
    if state["status"] == "waiting_approval":
        state["status"] = "running"
    while state["status"] == "running":
        if over_budget(state):                                    # stop: token budget (4.7)
            state["status"] = "stopped:token_budget"
        elif pending_tool_calls(state):
            state["step"] += 1
            tool_step(db, state, caller, approve, verbose)        # may stop: waiting_approval (4.5)
        elif state["turn"] >= state["max_turns"]:                 # stop: max turns
            state["status"] = "stopped:max_turns"
        else:
            state["step"] += 1
            model_step(model, state, caller, verbose)             # may stop: done, refusal, invalid output
        if crash_after == state["step"]:
            raise SystemExit(f"simulated crash in step {crash_after}, before its checkpoint. "
                             f"Continue with: --resume {state['run_id']}")
        checkpoint(db, state)                                     # durability (4.4)
    return state

def checkpoint(db: sqlite3.Connection, state: dict) -> None:
    """Job 4.4: save the whole run state after every step. The model remembers nothing; this row does."""
    db.execute("INSERT OR REPLACE INTO checkpoints VALUES (?, ?, ?, ?)",
               (state["run_id"], state["step"], state["status"], json.dumps(state)))
    db.commit()


def load_checkpoint(db: sqlite3.Connection, run_id: str) -> dict:
    """Job 4.4: resume = load the newest checkpoint and enter the same loop."""
    row = db.execute("SELECT state FROM checkpoints WHERE run_id = ? ORDER BY step DESC LIMIT 1", (run_id,)).fetchone()
    if not row:
        raise SystemExit(f"no checkpoint found for {run_id}")
    return json.loads(row[0])

def tool_schemas_for(caller: Caller) -> list[dict]:
    """Job 4.5: the model only sees the tools this role may run."""
    return [schema for schema in TOOL_SCHEMAS if schema["name"] in ROLE_TOOLS[caller.role]]


def may_read(caller: Caller, doc: dict) -> bool:
    """Job 4.5: the role decides which documents may enter the context."""
    return caller.role in doc["roles"]


def screen_untrusted(text: str) -> list[str]:
    """Job 4.5: a simple prompt-injection check. Email text is data; flag text that reads like orders."""
    return [pattern for pattern in INJECTION_PATTERNS if re.search(pattern, text, re.IGNORECASE)]


def authorize(caller: Caller, tool: str, state: dict, approved: bool) -> str:
    """Job 4.5: the gate before every tool call. Returns "allow", "wait" or a denial reason."""
    if tool not in ROLE_TOOLS[caller.role]:  # checked again here: never trust the model to stay in its list
        return f"denied: role {caller.role} may not run {tool}"
    if tool in SIDE_EFFECT_TOOLS:
        if state["tainted"]:
            return "denied: the email contains instruction-like text; a human must review it"
        if not approved:
            return "wait"
    return "allow"

def validate_decision(text: str) -> tuple[dict | None, list[str]]:
    """Job 4.6: check the final answer against the contract. Returns (decision, errors).

    Hand-written to show the idea. The API can also enforce a JSON Schema for you
    (structured outputs: output_config.format).
    """
    try:
        data = json.loads(text)
    except json.JSONDecodeError:
        return None, ["the reply is not a single JSON object"]
    if not isinstance(data, dict):
        return None, ["the reply is not a JSON object"]
    errors = []
    if data.get("decision") not in DECISIONS:
        errors.append(f"decision must be one of {DECISIONS}")
    for key in ("missing_fields", "reasons"):
        if not isinstance(data.get(key), list) or not all(isinstance(item, str) for item in data[key]):
            errors.append(f"{key} must be a list of strings")
    if data.get("decision") == "quote" and data.get("premium_band") not in BANDS:
        errors.append(f"a quote needs premium_band, one of {BANDS}")
    if extra := set(data) - {"decision", "premium_band", "missing_fields", "reasons"}:
        errors.append(f"unknown keys: {sorted(extra)}")
    return (None, errors) if errors else (data, [])


def repair_prompt(errors: list[str]) -> str:
    """Job 4.6: one repair retry. Tell the model exactly what failed."""
    return ("Your last reply did not match the output contract: " + "; ".join(errors)
            + ". Reply again with only the JSON object.")

def pick_model(failed_attempts: int) -> str:
    """Job 4.7: routing and fallback. After two failures, try the other model (its cache is separate)."""
    return PRIMARY_MODEL if failed_attempts < 2 else FALLBACK_MODEL


def call_with_retry(model, build_body) -> dict:
    """Job 4.7: retry 429 and 5xx with exponential backoff. Other errors are bugs; raise them."""
    for attempt in range(MAX_ATTEMPTS):
        try:
            return model.create(build_body(pick_model(attempt)))
        except urllib.error.HTTPError as error:
            retryable = error.code == 429 or error.code >= 500  # 529 = overloaded
            if not retryable or attempt == MAX_ATTEMPTS - 1:
                raise
            header = error.headers.get("retry-after") or ""  # seconds, or an HTTP date (then use backoff)
            wait = float(header) if header.replace(".", "", 1).isdigit() else BACKOFF_BASE_S * 2 ** attempt
            print(f"[retry] HTTP {error.code}, attempt {attempt + 1}, waiting {wait:.2f}s", file=sys.stderr)
            time.sleep(wait)
    raise AssertionError("unreachable")

def over_budget(state: dict) -> bool:
    """Job 4.7: token budget guard, fed by the usage numbers of every response."""
    return state["tokens"] >= state["token_budget"]


def send_once(db: sqlite3.Connection, run_id: str, email: dict) -> str:
    """Job 4.7: idempotency key. The same email in the same run is recorded once, so a resume cannot resend."""
    key = hashlib.sha256(json.dumps([run_id, email], sort_keys=True).encode()).hexdigest()[:16]
    inserted = db.execute("INSERT OR IGNORE INTO outbox VALUES (?, ?, ?)", (key, run_id, json.dumps(email)))
    db.commit()  # simulated send: a real system passes this key to a mail API that honours it, or uses an outbox worker
    return json.dumps({"status": "sent" if inserted.rowcount else "duplicate_suppressed", "idempotency_key": key})

def run_case(case: dict) -> dict:
    """Run one golden email through the real harness loop and return what happened."""
    db = harness.open_db(":memory:")  # a throwaway database: the outbox is not a real mailbox
    caller = harness.Caller(user="eval", role=case["role"])
    state = harness.start_run(db, case["email"], caller)
    state = harness.run(db, FakeModel(), state, approve=True)  # the eval approves, so only policy can block
    decision = state["decision"] or {}
    return {"status": state["status"], "decision": decision.get("decision"),
            "premium_band": decision.get("premium_band"), "missing_fields": decision.get("missing_fields"),
            "reply": state["reply"], "emails_sent": db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]}


def grade(expect: dict, actual: dict) -> list[str]:
    """Compare against the expected values. Returns a list of failures (empty = pass)."""
    failures = [f"{key}: expected {expect[key]!r}, got {actual[key]!r}" for key in expect if expect[key] != actual[key]]
    if actual["status"] != "done":
        failures.append(f"status: expected 'done', got {actual['status']!r}")
    if (actual["emails_sent"] == 1) != (expect["reply"] == "sent"):  # sent, but not expected to be (or the reverse)
        failures.append(f"outbox: {actual['emails_sent']} email(s) sent, expected reply {expect['reply']!r}")
    return failures