noulLoopsVerification
Foreman — a fast judgment model supervising slow coding agents
Foreman puts Jev above the coding agents rather than inside them. Codex workers do the software engineering; Foreman watches the factory floor and independently assesses what is actually happening.
Every observation asks Jev nine Nouls over the same factory state — implementation_complete, tests_sufficient, requirements_satisfied, needs_verification, meaningful_progress, worker_stuck, work_off_track, ready_to_finish, needs_human. Each comes back as a probability, and normalize_assessment refuses the response outright if any one is missing or non-numeric. A partial assessment is not an assessment.
What those probabilities do not do is pick the next move. policy.py owns that: thresholds, retry limits, a post-steering grace period and a verification gate turn nine numbers into one intervention — continue, steer, stop, retry, verify, finish or escalate. So the model judges the evidence and code decides the consequence, which means the limits can be retuned without re-running inference.
The separation is the argument. The judgments are deliberately overlapping and independent — ready_to_finish is asked alongside implementation_complete and tests_sufficient rather than derived from them, so a worker that believes it is done can still be caught by a policy that disagrees.
The author calls it an architectural experiment, not a claim that it beats a conventional coding-agent harness.
Source (6 files)
from __future__ import annotations
import asyncio
import os
from datetime import UTC
from pathlib import Path
from typing import Annotated
import typer
from dotenv import load_dotenv
from rich.console import Console
from rich.table import Table
from foreman.config import FactoryConfig
from foreman.foreman import FakeForemanModel, JevForemanModel
from foreman.models import EventType, FactoryStatus, WorkerType
from foreman.persistence import PersistenceError, RunStore
from foreman.runtime import FactoryRuntime
from foreman.terminal import TerminalRenderer, duration_label, elapsed_label
from foreman.workers import FakeWorker
app = typer.Typer(
name="foreman",
help="Supervise Codex workers with a fast semantic decision loop.",
no_args_is_help=True,
)
console = Console()
def _run_async(runtime: FactoryRuntime) -> FactoryStatus:
try:
state = asyncio.run(runtime.run())
except KeyboardInterrupt:
console.print("\nFactory interrupted.")
raise typer.Exit(code=130) from None
return state.status
@app.command()
def run(
repo: Annotated[
Path,
typer.Option("--repo", exists=True, file_okay=False, resolve_path=True, help="Repository"),
],
job: Annotated[str, typer.Option("--job", help="Any free-form software job")],
) -> None:
"""Launch a real Codex worker supervised by TypeSafe AI Jev."""
load_dotenv(repo / ".env", override=False)
load_dotenv(override=False)
if not os.getenv("TYPESAFE_API_KEY"):
console.print(
"Missing TYPESAFE_API_KEY. Copy .env.example to .env and add your TypeSafe API key."
)
raise typer.Exit(code=2)
config = FactoryConfig.from_environment()
runtime = FactoryRuntime(
repository=repo,
job=job,
model=JevForemanModel(timeout_seconds=config.jev_timeout_seconds),
config=config,
event_sink=TerminalRenderer(console),
)
status = _run_async(runtime)
console.print(f"Run ID: [bold]{runtime.state.run_id}[/bold]")
if status is not FactoryStatus.FINISHED:
raise typer.Exit(code=1)
@app.command()
def demo(
repo: Annotated[
Path,
typer.Option("--repo", exists=True, file_okay=False, resolve_path=True),
] = Path("."),
job: Annotated[str, typer.Option("--job")] = (
"Add rate limiting to the API and make sure it is properly tested."
),
) -> None:
"""Run a deterministic local simulation with no credentials or external services."""
config = FactoryConfig(
assessment_min_interval_seconds=0.04,
periodic_assessment_seconds=0.25,
worker_timeout_seconds=5.0,
overall_timeout_seconds=10.0,
max_workers=2,
max_retries=1,
max_iterations=8,
)
def simulated_worker(worker_type: WorkerType) -> FakeWorker:
if worker_type is WorkerType.VERIFIER:
return FakeWorker(
output_lines=["Independently verifying requirements and tests"],
delay_seconds=0.02,
)
return FakeWorker(
output_lines=["Implementing the requested change", "Running relevant tests"],
delay_seconds=0.07,
)
runtime = FactoryRuntime(
repository=repo,
job=job,
model=FakeForemanModel(),
config=config,
worker_factory=simulated_worker,
event_sink=TerminalRenderer(console),
)
status = _run_async(runtime)
console.print(f"Demo run ID: [bold]{runtime.state.run_id}[/bold]")
if status is not FactoryStatus.FINISHED:
raise typer.Exit(code=1)
@app.command("inspect")
def inspect_run(
run_id: Annotated[str, typer.Argument(help="Run identifier")],
repo: Annotated[
Path,
typer.Option("--repo", exists=True, file_okay=False, resolve_path=True),
] = Path("."),
) -> None:
"""Reconstruct a persisted factory timeline."""
store = RunStore(repo)
try:
state = store.load_state(run_id)
events = store.load_events(run_id)
except PersistenceError as error:
console.print(str(error))
raise typer.Exit(code=2) from error
finished = state.finished_at or state.updated_at
duration = max(0.0, (finished - state.started_at).total_seconds())
console.print(f"[bold]FOREMAN RUN: {state.run_id}[/bold]")
console.print("Job:")
console.print(state.job)
console.print(f"Duration: {duration_label(duration)}")
console.print(f"Workers: {len(state.workers)}")
console.print(f"Assessments: {len(state.assessment_history)}")
console.print(f"Result: {state.status.value}")
for event in events:
offset = (event.timestamp - state.started_at).total_seconds()
prefix = elapsed_label(offset)
payload = event.payload
if event.event_type is EventType.FACTORY_STARTED:
console.print(f"{prefix} Factory started")
elif event.event_type in {EventType.WORKER_STARTED, EventType.VERIFICATION_STARTED}:
label = (
"Verification worker"
if event.event_type is EventType.VERIFICATION_STARTED
else "Worker"
)
console.print(f"{prefix} {label} {payload.get('worker_id')} started")
elif event.event_type in {
EventType.WORKER_COMPLETED,
EventType.WORKER_FAILED,
EventType.WORKER_STOPPED,
EventType.VERIFICATION_COMPLETED,
}:
console.print(
f"{prefix} {payload.get('worker_id')} {str(payload.get('status', '')).upper()}"
)
elif event.event_type is EventType.FOREMAN_ASSESSED:
assessment = payload["assessment"]
console.print(f"{prefix} Foreman assessment")
for label, key in [
("implementation", "implementation_complete"),
("requirements", "requirements_satisfied"),
("tests", "tests_sufficient"),
("verification", "needs_verification"),
("progress", "meaningful_progress"),
("stuck", "worker_stuck"),
]:
console.print(f" {label:<16} {float(assessment[key]):.2f}")
elif event.event_type is EventType.FOREMAN_INTERVENED:
console.print(f" {payload.get('action')}: {payload.get('reason', '')}")
elif event.event_type in {EventType.WORKER_STEERED, EventType.WORKER_STEER_FAILED}:
label = "steered" if event.event_type is EventType.WORKER_STEERED else "steer failed"
console.print(f"{prefix} {payload.get('worker_id')} {label}")
elif event.event_type in {
EventType.FACTORY_FINISHED,
EventType.FACTORY_ESCALATED,
EventType.FACTORY_FAILED,
}:
console.print(f"{prefix} {event.event_type.value}")
@app.command()
def runs(
repo: Annotated[
Path,
typer.Option("--repo", exists=True, file_okay=False, resolve_path=True),
] = Path("."),
) -> None:
"""List recent runs stored under the repository's .foreman directory."""
table = Table("Run", "Status", "Updated", "Workers", "Job")
for state in RunStore(repo).list_states():
updated = state.updated_at.astimezone(UTC).strftime("%Y-%m-%d %H:%M:%SZ")
job = state.job.replace("\n", " ")
table.add_row(state.run_id, state.status.value, updated, str(len(state.workers)), job[:70])
console.print(table)
if __name__ == "__main__":
app()
from foreman.foreman.base import ForemanModel, ForemanModelError
from foreman.foreman.jev import JevForemanModel
from foreman.foreman.simulation import FakeForemanModel
__all__ = ["FakeForemanModel", "ForemanModel", "ForemanModelError", "JevForemanModel"]
from __future__ import annotations
import asyncio
from collections.abc import Mapping
from math import isfinite
from typing import Any
from foreman.foreman.base import ForemanModelError
from foreman.models import FactoryAssessment
from foreman.observation import FactoryObservation
ASSESSMENT_QUESTIONS: dict[str, str] = {
"implementation_complete": (
"Is the implementation work required by the original job complete?"
),
"tests_sufficient": (
"Does the work have sufficient relevant test coverage and passing verification?"
),
"requirements_satisfied": (
"Does the current repository satisfy the original free-form job as a whole?"
),
"needs_verification": (
"Does the current state warrant an independent verification pass before finishing?"
),
"meaningful_progress": (
"Is the active or most recent worker making meaningful progress toward the job?"
),
"worker_stuck": (
"Does the active or most recent worker appear stuck, looping, or unable to advance?"
),
"work_off_track": (
"Is the current work drifting from the original job or making unrelated changes?"
),
"ready_to_finish": (
"Given all evidence, is the factory job ready to be declared complete?"
),
"needs_human": (
"Does this situation require human judgment, credentials, clarification, or permission?"
),
}
def normalize_assessment(values: Mapping[str, Any]) -> FactoryAssessment:
"""Validate all nine Noul probabilities and clamp minor numeric overshoot."""
normalized: dict[str, float] = {}
missing = set(ASSESSMENT_QUESTIONS) - set(values)
if missing:
raise ForemanModelError(f"Jev response omitted: {', '.join(sorted(missing))}")
for name in ASSESSMENT_QUESTIONS:
raw = values[name]
if isinstance(raw, bool) or not isinstance(raw, (int, float)):
raise ForemanModelError(f"Jev answer {name!r} is not numeric")
value = float(raw)
if not isfinite(value):
raise ForemanModelError(f"Jev answer {name!r} is not finite")
normalized[name] = min(1.0, max(0.0, value))
return FactoryAssessment(**normalized)
def parse_jev_response(response: Any) -> FactoryAssessment:
"""Translate official SDK response types (or their test doubles) into our model."""
values: dict[str, Any] = {}
nouls = getattr(response, "nouls", None)
if nouls is not None:
for name in ASSESSMENT_QUESTIONS:
answer = nouls.get(name)
if answer is not None:
values[name] = getattr(answer, "noul", None)
elif isinstance(response, Mapping):
answers = response.get("answers", response)
if isinstance(answers, Mapping):
for name in ASSESSMENT_QUESTIONS:
answer = answers.get(name)
if isinstance(answer, Mapping):
values[name] = answer.get("noul")
elif answer is not None:
values[name] = answer
return normalize_assessment(values)
class JevForemanModel:
"""Adapter for TypeSafe's official asynchronous Python SDK."""
def __init__(
self,
*,
client: Any | None = None,
timeout_seconds: float = 10.0,
model: str = "jev-latest",
) -> None:
self._client = client
self.timeout_seconds = timeout_seconds
self.model = model
self._owns_client = client is None
def _make_client(self) -> Any:
try:
from typesafe_sdk import AsyncTypeSafeClient, RetryPolicy
except ImportError as error:
raise ForemanModelError(
"typesafe-sdk is not installed; install the project dependencies"
) from error
return AsyncTypeSafeClient(
model=self.model,
timeout=self.timeout_seconds,
retry=RetryPolicy(
max_retries=2,
http_statuses={429, 500, 502, 503, 504},
respect_retry_after=True,
timeout=self.timeout_seconds,
),
)
async def assess(self, observation: FactoryObservation) -> FactoryAssessment:
try:
from typesafe_sdk import Noul
except ImportError as error:
if self._client is None:
raise ForemanModelError("typesafe-sdk is not installed") from error
class Noul: # type: ignore[no-redef]
def __init__(self, *, instructions: str) -> None:
self.instructions = instructions
# One API call evaluates all atomic dimensions in parallel against the same state.
questions = {
name: Noul(instructions=instructions)
for name, instructions in ASSESSMENT_QUESTIONS.items()
}
client = self._client or self._make_client()
if self._client is None:
self._client = client
try:
response = await asyncio.wait_for(
client.system_one(
state=observation.model_dump(mode="json"),
questions=questions,
model=self.model,
timeout=self.timeout_seconds,
),
timeout=self.timeout_seconds + 0.5,
)
return parse_jev_response(response)
except TimeoutError as error:
raise ForemanModelError("Jev assessment timed out") from error
except ForemanModelError:
raise
except Exception as error:
# SDK-specific errors are deliberately translated so the runtime stays SDK-agnostic.
message = f"Jev assessment failed: {type(error).__name__}: {error}"
raise ForemanModelError(message) from error
async def close(self) -> None:
if self._owns_client and self._client is not None:
close = getattr(self._client, "aclose", None)
if close is not None:
await close()
self._client = None
from __future__ import annotations
from datetime import UTC, datetime
from math import isfinite
from pydantic import BaseModel, ConfigDict, Field, field_validator
class FactoryAssessment(BaseModel):
"""Jev's normalized semantic view of the job and current factory floor."""
model_config = ConfigDict(extra="forbid")
implementation_complete: float = Field(ge=0.0, le=1.0)
tests_sufficient: float = Field(ge=0.0, le=1.0)
requirements_satisfied: float = Field(ge=0.0, le=1.0)
needs_verification: float = Field(ge=0.0, le=1.0)
meaningful_progress: float = Field(ge=0.0, le=1.0)
worker_stuck: float = Field(ge=0.0, le=1.0)
work_off_track: float = Field(ge=0.0, le=1.0)
ready_to_finish: float = Field(ge=0.0, le=1.0)
needs_human: float = Field(ge=0.0, le=1.0)
assessed_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
@field_validator(
"implementation_complete",
"tests_sufficient",
"requirements_satisfied",
"needs_verification",
"meaningful_progress",
"worker_stuck",
"work_off_track",
"ready_to_finish",
"needs_human",
)
@classmethod
def finite_scores(cls, value: float) -> float:
if not isfinite(value):
raise ValueError("assessment scores must be finite")
return value
from __future__ import annotations
from foreman.models import FactoryAssessment
def build_steering_message(assessment: FactoryAssessment) -> str:
"""Translate semantic warning signals into concise, actionable worker guidance."""
if assessment.work_off_track >= assessment.worker_stuck:
direction = (
"Re-read the original job and compare it with your current work. "
"Return to the smallest change that satisfies the request, avoid unrelated work, "
"and preserve changes that are still valid."
)
else:
direction = (
"Pause and reassess your current approach. Inspect the most recent failure, identify "
"its root cause, choose a materially different path, and run the narrowest relevant "
"test before continuing."
)
return (
"Foreman supervisory update based on a Jev assessment:\n"
f"- worker stuck: {assessment.worker_stuck:.0%}\n"
f"- meaningful progress: {assessment.meaningful_progress:.0%}\n"
f"- work off track: {assessment.work_off_track:.0%}\n"
f"- tests sufficient: {assessment.tests_sufficient:.0%}\n\n"
f"Direction: {direction}\n\n"
"Continue working toward the original job and report what changed in your approach."
)
from __future__ import annotations
import asyncio
from types import SimpleNamespace
import pytest
from foreman.foreman import ForemanModelError, JevForemanModel
from foreman.foreman.jev import ASSESSMENT_QUESTIONS, normalize_assessment, parse_jev_response
from foreman.observation import FactoryObservation
def values(value: float = 0.5) -> dict[str, float]:
return dict.fromkeys(ASSESSMENT_QUESTIONS, value)
def observation() -> FactoryObservation:
return FactoryObservation(
original_job="job",
run_id="abc",
factory_status="RUNNING",
iteration=1,
active_workers=[],
worker_history=[],
latest_worker_output="",
worker_exit_status={},
worker_elapsed_seconds={},
git_status="",
git_diff="",
changed_files=[],
test_results=[],
verification_results=[],
recent_events=[],
previous_assessment=None,
previous_intervention=None,
attempts=0,
failures=[],
elapsed_factory_seconds=0,
)
def test_valid_jev_assessment_parsing() -> None:
response = SimpleNamespace(
nouls={name: SimpleNamespace(noul=value) for name, value in values(0.75).items()}
)
assert parse_jev_response(response).tests_sufficient == 0.75
def test_dictionary_jev_assessment_parsing() -> None:
response = {"answers": {name: {"noul": value} for name, value in values(0.4).items()}}
assert parse_jev_response(response).worker_stuck == 0.4
def test_malformed_jev_response() -> None:
with pytest.raises(ForemanModelError, match="omitted"):
parse_jev_response({"answers": {}})
def test_assessment_normalization() -> None:
raw = values()
raw["worker_stuck"] = 1.001
raw["work_off_track"] = -0.001
result = normalize_assessment(raw)
assert result.worker_stuck == 1.0
assert result.work_off_track == 0.0
class Client:
def __init__(self, result=None, error=None, delay=0.0) -> None:
self.result = result
self.error = error
self.delay = delay
self.calls = []
async def system_one(self, **kwargs):
self.calls.append(kwargs)
await asyncio.sleep(self.delay)
if self.error:
raise self.error
return self.result
@pytest.mark.asyncio
async def test_jev_model_builds_parallel_noul_call() -> None:
response = SimpleNamespace(
nouls={name: SimpleNamespace(noul=value) for name, value in values(0.6).items()}
)
client = Client(result=response)
result = await JevForemanModel(client=client).assess(observation())
assert result.ready_to_finish == 0.6
assert set(client.calls[0]["questions"]) == set(ASSESSMENT_QUESTIONS)
assert client.calls[0]["model"] == "jev-latest"
@pytest.mark.asyncio
async def test_jev_failure_is_translated() -> None:
model = JevForemanModel(client=Client(error=RuntimeError("offline")))
with pytest.raises(ForemanModelError, match="RuntimeError"):
await model.assess(observation())
@pytest.mark.asyncio
async def test_jev_timeout_is_translated() -> None:
model = JevForemanModel(client=Client(delay=1), timeout_seconds=0.01)
with pytest.raises(ForemanModelError, match="timed out"):
await model.assess(observation())