Give a model subprocess an output contract

Shipped

I added an optional Codex backend for the scheduled local-fitness brief while retaining Claude as the default. The new transport sends prepared context to codex exec, requests schema-shaped output, and reads the final response from a dedicated file. Both backends feed the existing finalization path rather than introducing separate brief persistence logic.

The subprocess boundary is worth testing on its own. You do not need a model to find out that your wrapper reads the wrong file, ignores a failed exit code, or accepts an empty response.

Build the transport before connecting the provider

Use Python 3.11 or newer and create an empty directory. This walkthrough uses a fake worker, so it needs neither Codex nor an account. It tests transport behavior, not generation quality or a provider’s schema implementation.

Save this first block as transport.py. The worker contract is deliberately small: accept a prompt on stdin and write a JSON object to the output path supplied as the last argument.

import json
import subprocess
import tempfile
from pathlib import Path

def complete(command: list[str], prompt: str, timeout: float = 5.0) -> dict:
    with tempfile.TemporaryDirectory(prefix="model-transport-") as directory:
        output = Path(directory) / "response.json"
        try:
            result = subprocess.run(
                [*command, str(output)], input=prompt, text=True,
                capture_output=True, timeout=timeout, cwd=directory,
                check=False,
            )
        except FileNotFoundError as error:
            raise RuntimeError("worker executable missing") from error
        except subprocess.TimeoutExpired as error:
            raise RuntimeError("worker timed out") from error
        if result.returncode != 0:
            raise RuntimeError(f"worker exited {result.returncode}")
        if not output.is_file():
            raise RuntimeError("worker produced no response file")
        raw = output.read_text(encoding="utf-8").strip()
        if not raw:
            raise RuntimeError("worker produced an empty response")
        try:
            value = json.loads(raw)
        except json.JSONDecodeError as error:
            raise RuntimeError("worker produced invalid JSON") from error
        if (not isinstance(value, dict) or set(value) != {"summary"}
                or not isinstance(value["summary"], str)
                or not value["summary"].strip()):
            raise RuntimeError("worker response violates the application contract")
        return value

Passing a list keeps arguments separate without asking a shell to interpret the prompt. The prompt travels through stdin, not through a command string. Python documents run as the recommended interface for supported subprocess use cases, including captured output and a timeout that kills and waits for the child.

The response path is new for every call. That matters when yesterday’s job left a perfectly valid JSON file behind. A failed generation must not look successful because the wrapper discovered an older answer. TemporaryDirectory gives this example a per-call directory that is cleaned up on leaving the context.

This is workspace hygiene, not a security sandbox. Changing the working directory does not prevent a child process from reading other accessible files. Keep that distinction explicit when adapting the example to a model CLI.

Use a worker that fails on purpose

Save this as fake_worker.py, next to the transport. It has one successful mode and several deliberately broken modes. Its stdout message is noise on purpose: the wrapper must read the response file, not whichever text looks helpful in the logs.

import json
import sys
import time
from pathlib import Path

mode, destination = sys.argv[1:]
prompt = sys.stdin.read()
output = Path(destination)

if mode == "fail":
    sys.exit(7)
if mode == "missing":
    sys.exit(0)
if mode == "slow":
    time.sleep(2)
elif mode == "empty":
    output.write_text("  ", encoding="utf-8")
elif mode == "malformed":
    output.write_text("{", encoding="utf-8")
elif mode == "wrong-shape":
    output.write_text('{"summary": 42}', encoding="utf-8")
else:
    output.write_text(json.dumps({"summary": prompt.upper()}), encoding="utf-8")
print("diagnostic output, not the response")

The fake does not impersonate Codex flags. It exercises the interface owned by the wrapper. That makes a failed test much easier to interpret: authentication, model availability, and remote latency cannot explain it.

Save the test driver as check_transport.py:

import sys
from pathlib import Path
from transport import complete

worker = str(Path(__file__).with_name("fake_worker.py").resolve())
def command(mode: str) -> list[str]:
    return [sys.executable, worker, mode]

assert complete(command("ok"), "release ready") == {"summary": "RELEASE READY"}
print("valid response: RELEASE READY")
for mode in ("fail", "missing", "empty", "malformed", "wrong-shape", "slow"):
    try:
        complete(command(mode), "release ready", timeout=0.5)
    except RuntimeError:
        print(mode + ": rejected")
    else:
        raise AssertionError(mode + " should have failed")

The absolute worker path is intentional. The transport changes the child’s directory, so a relative path to fake_worker.py would point inside the temporary workspace. The same mistake can affect a scheduled CLI job that works from an interactive terminal.

Run the check from the directory containing all three files:

python3 check_transport.py

Expected output:

valid response: RELEASE READY
fail: rejected
missing: rejected
empty: rejected
malformed: rejected
wrong-shape: rejected
slow: rejected

Connect the real CLI at one boundary

In the tagged implementation, the provider-specific command uses --output-schema and --output-last-message, with the prompt read from stdin. Those interfaces are documented in Codex non-interactive mode. The schema and output destination are temporary files created before the process starts.

Your application will need a command builder for its actual provider; the fake’s positional output argument is not a drop-in Codex invocation. Keep that provider translation separate from the response checks above, and add a small authenticated smoke test only after the offline checks pass.

The production branch also calls the blocking transport through asyncio.to_thread and then uses the shared brief finalizer. That placement keeps provider selection from turning into a second implementation of validation and saving. A new backend should produce the same application input, including the same error behavior where practical.

Do not mistake a successful process exit for a valid application response. The illustrative validator requires one nonempty summary; a real report needs its own semantic checks. Schema-constrained generation can help shape the answer, but it does not establish that a fact in the answer is true.

Gotchas

These are failure modes exercised by the local harness, not claims that each occurred in a production run.

A worker can exit successfully without producing the artifact. The missing and empty modes both return normally. Reading a fresh, nonempty response file after checking the exit code catches them. Do not fall back to a previous file or scrape logs for a plausible answer.

An empty working directory does not disable tools. The release sets a read-only sandbox and asks the composer not to use tools. Read-only is not the same claim as no reads, no subprocesses, or no network. If your security requirement is stronger, enforce it with supported runtime controls and verify the effective capabilities separately.

A parent timeout is not a general process-tree supervisor. Python’s documented run timeout handles its child, and the fake proves that case. A CLI that launches long-lived descendants needs additional supervision. Do not describe this test as proving that every descendant is terminated.

Captured logs still need a budget. capture_output buffers output. For a worker with potentially unbounded logging, replace the small-example approach with bounded capture or controlled log files. Also avoid copying raw stderr into user-visible errors when it can contain private prompts or credentials.

Sources

Python subprocess — argument lists, captured output, and timeout behavior.
Python tempfile — per-call temporary workspaces and cleanup.
Codex non-interactive mode — schema output and final-message files.

Changelog

feat: add Codex scheduled brief backend (#254) (#255) (6322064)