Build a Jev-Gated Claude Router

Introduction

30 Second Summary

Automated coding tools can act on a request in seconds. A confident but mistaken guess can send work down the wrong path before anyone notices.

In this project, you will build a local agent router that governs whether software tasks can reach Claude Code. Jev supplies classification evidence that Pydantic validates before deterministic policy selects a restricted specialist or requests human review.

What You'll Build

A terminal demo shows each software task reaching a read-only Claude Code specialist only after the router's policy approves it.

By the end of this project, you'll have:

• A visible routing comparison that exposes the frontend-only baseline on a backend task before showing the governed route select backend.
• A fail-closed test suite that proves malformed evidence, classifier failure, low confidence, or high security risk cannot reach execution.
• A portfolio-ready evidence pack with a labeled dataset, a Mermaid architecture diagram, plus an evaluation report covering routing accuracy, escalation recall, assignment counts, latency, token usage, and cost.
• Secret Mission: Evaluate a disguised authorization bug that looks visual but must be blocked for human review before Claude Code executes.

Are there any prerequisites?

You need Python 3.11+, Claude Code 2.1.281+, an OpenRouter API key with Jev access, plus an existing TypeSafe Python SDK integration. You should also be comfortable reading advanced Python.

Before We Start

This step is your moment to define the security boundary before any hands-on work begins. You are building a local fail-closed agent router where Jev provides probabilistic classification evidence while deterministic policy controls execution authority and restricted Claude Code delegation.

Set Up the Local Evaluation Harness

Security tests only prove something when every run starts from the same dependencies. A stale package or unsupported command can make a boundary appear safer than it is.

This step gives the router an isolated Python virtual environment with pinned packages. You will also verify that Claude Code supports the agent features used later.

In this step, get ready to:

• Create the local project workspace with pinned Python dependencies.
• Verify a supported Claude Code installation with authenticated access.
• Prepare the fixture directories with a shell-only OpenRouter key.

Create the pinned Python workspace

A project-local environment isolates this router from packages installed elsewhere on your Mac. Exact pins make the later tests reproducible.

• Click the Finder icon in the Dock.
• Select Desktop in the Finder sidebar.
• Click File in the menu bar.
• Choose New Folder.
• Enter jev-agent-router as the folder name.
• Press Return.

You now see the jev-agent-router folder on your Desktop. This folder holds every local artifact created by the project.

• Double-click the jev-agent-router folder.
• Click View in the Finder menu bar.
• Choose Show Path Bar.

You should see the folder path along the bottom of the Finder window.

• Control-click jev-agent-router in the path bar.
• Choose Open in Terminal.

A Terminal window opens with jev-agent-router as its current folder.

• Create the project-local .venv environment by running this command:

python3 -m venv .venv

What does this command do?

Python creates an isolated interpreter with its own package directory inside .venv. The environment keeps this router's dependencies separate from other Python projects.

• Activate the new environment by running this command:

source .venv/bin/activate

What does activation change?

Activation points Python commands at the interpreter inside .venv. The change applies only to this Terminal session.

You should see (.venv) at the start of your Terminal prompt. That visible prefix confirms the isolated environment is active.

Missing the environment prefix?

• Check that the Terminal prompt is inside the jev-agent-router folder.
• Confirm that the activation command uses the exact .venv/bin/activate path.

Ask for help with activating the project environment on macOS.

The package list turns dependency versions into part of the test harness. Anyone using the file receives the same SDK and validation behavior.

• Open the jev-agent-router folder in your code editor.
• Create requirements.txt inside the jev-agent-router folder using the editor's new-file control.
• Add the pinned dependencies by pasting this file content:

typesafe-sdk==0.7.4
pydantic==2.14.0
pytest==9.1.1

Why pin these packages?

• The TypeSafe Python SDK at 0.7.4 provides the client used to request Jev classifications.
• Pydantic at 2.14.0 validates classifier output against strict contracts.
• pytest at 9.1.1 runs the fail-closed test suite.

• Save requirements.txt.
• Confirm that requirements.txt appears in the editor's file explorer.

The ignore file keeps disposable environments and generated reports outside future source-control snapshots.

• Create .gitignore inside the jev-agent-router folder using the editor's new-file control.
• Add the project exclusions by pasting this file content:

.venv/
__pycache__/
.pytest_cache/
*.pyc
*-report.md

What stays out of source control?

The entries exclude the virtual environment and Python cache files. They also exclude pytest cache data and generated evaluation reports.

• Save .gitignore.
• Confirm that .gitignore appears in the editor's file explorer.

The environment is active. Install the pinned packages into it next.

• Install the project dependencies by running this command in the Terminal from earlier:

python -m pip install -r requirements.txt

What does this installation use?

The active environment makes python use the interpreter inside .venv. The requirements file supplies all three exact package versions.

The download can take a minute while Python resolves and installs the packages. The Terminal prompt returns when installation finishes.

Dependency installation failed?

• Confirm that (.venv) still appears in the Terminal prompt.
• Check that requirements.txt contains the three pins exactly as shown.
• Confirm that your Python version is 3.11 or newer.

Ask for help with installing the pinned dependencies in this virtual environment.

Use these tabs to compare both project files before moving on.

✔️ Awesome, I've got everything!

Your dependency pins and exclusions are saved. Keep (.venv) visible in the Terminal prompt for the rest of this project.

ⓧ I'd like to double check the full code

Compare your requirements.txt file with this reference.

typesafe-sdk==0.7.4
pydantic==2.14.0
pytest==9.1.1

What should match?

The file contains three lines. Each line uses an exact version pin with two equals signs.

Compare your .gitignore file with this reference.

.venv/
__pycache__/
.pytest_cache/
*.pyc
*-report.md

What should match?

The file contains five exclusions. The final pattern keeps every generated report with a matching filename out of source control.

Verify Claude Code and authentication

The router later passes file-based agent definitions to Claude Code. That feature requires version 2.1.281 or newer.

• Check the installed Claude Code version by running this command:

claude --version

What does this check prove?

The command prints the installed Claude Code version. Compare its numeric version with the minimum required for file-based agent definitions.

✔️ I see version 2.1.281 or higher

Your Claude Code installation supports the agent-definition workflow used later. That prerequisite is ready.

ⓧ I see an older version

The installed release is too old for this project's file-based agent definitions. Update Claude Code before continuing.

• Update Claude Code to the latest release by running this command:

claude update

What does the update do?

Claude Code checks for a newer release and updates the installed command. The required agent-definition support is available from version 2.1.281 onward.

• Check the updated version by running the version command again:

claude --version

What should the second check show?

The printed version should now be 2.1.281 or newer. This confirms the update reached the required feature set.

ⓧ Command not found

Claude Code is missing from the current shell. The native installer adds the command needed by the router.

The installer can stay quiet while it downloads Claude Code. Leave the Terminal window open until the process finishes.

• Install Claude Code using the official macOS installer by running this command:

curl -fsSL https://claude.ai/install.sh | bash

What does this installer do?

The command downloads the official Claude Code installer from Anthropic. The native installation receives automatic updates.

• Open a new Terminal window after the installer finishes.
• Return to the jev-agent-router folder using the Finder path bar.
• Activate .venv again using the earlier activation command.
• Verify the installation by running this command:

claude --version

What should the install check show?

A working installation prints a Claude Code version. Continue once that version is 2.1.281 or newer.

Claude Code still unavailable?

Use the setup note printed by the installer if the command is outside your shell path. You can also follow the official installation troubleshooting guide.

Ask for help with making Claude Code available in macOS Terminal.

Version support covers the feature set. Authentication confirms that the installed command can access your existing Claude Code account.

• Check the current Claude Code authentication status by running this command:

claude auth status

What does the authentication check prove?

Claude Code reports its authentication state as JSON. A successful authenticated status exits cleanly without starting an agent session.

You should see an authenticated status. Claude Code is now ready for the restricted delegation step later in the project.

Authentication is not ready?

Complete the sign-in flow for your existing Claude Code installation. Run the status check again after authentication finishes.

Ask for help with checking Claude Code authentication without exposing credentials.

Protect the API key and verify the harness

An environment variable lets the TypeSafe client read your OpenRouter credential without placing it in a project file. The exported value disappears when this Terminal session ends.

API keys need careful handling because they can authorize metered requests. This command keeps the key in the current shell and outside the repository.

• Replace the plain text placeholder with your OpenRouter API key.
• Export the key only in the current Terminal session by running this command:

export OPENROUTER_API_KEY="your-api-key-here"

Where does the key live?

The shell stores OPENROUTER_API_KEY for commands launched from this Terminal session. No project file receives the secret.

• Create an empty tests directory inside jev-agent-router using your editor's new-folder control.
• Create an empty sandbox_workspace directory inside jev-agent-router using the same control.

You should now see tests and sandbox_workspace beside the two project files in the editor's file explorer.

Before you run the final check, which result would reveal the most basic setup gap: an import failure, an old Claude Code version, or an unauthenticated status?

• Verify the Python imports and Claude Code prerequisites by running these commands:

python -c "from typesafe_sdk import TypeSafeClient; from pydantic import BaseModel; import pytest; print('Python dependencies ready')"
claude --version
claude auth status

What does the final check cover?

• The Python command imports one symbol from each installed dependency before printing its readiness message.
• The version command confirms support for the Claude Code agent-definition workflow.
• The authentication command confirms that Claude Code can use your existing signed-in account.

You should see Python dependencies ready. You should also see Claude Code version 2.1.281 or newer followed by an authenticated status.

One of the final checks failed?

• Reactivate .venv if the Python import command fails.
• Return to the matching version tab if Claude Code reports an older release.
• Complete Claude Code authentication if the status check is unsuccessful.

Ask for help with diagnosing the failed harness check.

Your local harness now has pinned dependencies and verified Claude Code access. Next, you will give it an always-frontend baseline and watch that baseline misroute a backend task.

Expose the Frontend-Only Baseline

Your pinned environment is ready for a controlled routing experiment. You can now compare a weak decision with the governed behavior you build next.

A router that always selects frontend can still produce confident-looking output. You need to see that failure before Jev classification can prove its value.

In this step, get ready to:

• Define four read-only Claude Code agent profiles.
• Build a checkout fixture with an intentionally weak baseline router.
• Run a backend task through the baseline to expose its frontend misroute.

Define read-only Claude Code profiles

Each Claude Code profile represents a specialist that can inspect the fixture workspace. The JSON definitions limit every profile to read-only tools.

• In your editor's file sidebar, create agents.json inside sandbox_workspace.
• Paste these four agent definitions into sandbox_workspace/agents.json:

{
  "frontend":{"description":"Read-only frontend specialist.","prompt":"Inspect this fixture workspace and return frontend evidence with file paths. Never request write, command, network or MCP access.","tools":["Read","Grep","Glob"],"disallowedTools":["Bash","Edit","Write","WebFetch","WebSearch","mcp__*"],"permissionMode":"dontAsk","maxTurns":2},
  "backend":{"description":"Read-only backend specialist.","prompt":"Inspect this fixture workspace and return backend evidence with file paths. Never request write, command, network or MCP access.","tools":["Read","Grep","Glob"],"disallowedTools":["Bash","Edit","Write","WebFetch","WebSearch","mcp__*"],"permissionMode":"dontAsk","maxTurns":2},
  "database":{"description":"Read-only database specialist.","prompt":"Inspect this fixture workspace and return database evidence with file paths. Never request write, command, network or MCP access.","tools":["Read","Grep","Glob"],"disallowedTools":["Bash","Edit","Write","WebFetch","WebSearch","mcp__*"],"permissionMode":"dontAsk","maxTurns":2},
  "security":{"description":"Read-only security specialist.","prompt":"Inspect this fixture workspace and return security evidence and human review questions. Never request write, command, network or MCP access.","tools":["Read","Grep","Glob"],"disallowedTools":["Bash","Edit","Write","WebFetch","WebSearch","mcp__*"],"permissionMode":"dontAsk","maxTurns":2}
}

What do these definitions control?

• The four top-level keys identify the frontend, backend, database, and security specialists.
• The tools lists allow only Read, Grep, and Glob.
• The disallowedTools lists deny write, shell, web, and MCP access.
• The dontAsk permission mode prevents an unavailable tool from being rescued through an approval prompt.

• Save sandbox_workspace/agents.json.
• Confirm that agents.json appears under sandbox_workspace in your editor's file sidebar.

Agent definitions not saving?

Check that the file is named agents.json inside sandbox_workspace. Make sure every profile remains inside the outer braces.

Ask for help checking the structure of the agent definitions.

✔️ Awesome, I've got everything!

Your four read-only agent profiles are saved in the fixture workspace.

ⓧ I'd like to double check the full code

Compare your file with this complete version.

{
  "frontend":{"description":"Read-only frontend specialist.","prompt":"Inspect this fixture workspace and return frontend evidence with file paths. Never request write, command, network or MCP access.","tools":["Read","Grep","Glob"],"disallowedTools":["Bash","Edit","Write","WebFetch","WebSearch","mcp__*"],"permissionMode":"dontAsk","maxTurns":2},
  "backend":{"description":"Read-only backend specialist.","prompt":"Inspect this fixture workspace and return backend evidence with file paths. Never request write, command, network or MCP access.","tools":["Read","Grep","Glob"],"disallowedTools":["Bash","Edit","Write","WebFetch","WebSearch","mcp__*"],"permissionMode":"dontAsk","maxTurns":2},
  "database":{"description":"Read-only database specialist.","prompt":"Inspect this fixture workspace and return database evidence with file paths. Never request write, command, network or MCP access.","tools":["Read","Grep","Glob"],"disallowedTools":["Bash","Edit","Write","WebFetch","WebSearch","mcp__*"],"permissionMode":"dontAsk","maxTurns":2},
  "security":{"description":"Read-only security specialist.","prompt":"Inspect this fixture workspace and return security evidence and human review questions. Never request write, command, network or MCP access.","tools":["Read","Grep","Glob"],"disallowedTools":["Bash","Edit","Write","WebFetch","WebSearch","mcp__*"],"permissionMode":"dontAsk","maxTurns":2}
}

Build the fixture and baseline router

The fixture gives a future backend agent a small file to inspect. Its two functions model a checkout status lookup plus the message rendered from that status.

• In your editor's file sidebar, create app.py inside sandbox_workspace.
• Paste this checkout fixture into sandbox_workspace/app.py:

def fetch_checkout_status(user_id: str) -> dict[str, str]:
    return {"user_id": user_id, "status": "ready"}


def render_checkout_message(status: dict[str, str]) -> str:
    return f"Checkout status: {status['status']}"

What does this fixture provide?

• The fetch_checkout_status() function returns a predictable checkout result for a supplied user.
• The render_checkout_message() function converts that result into a short display string.
• The file gives the restricted backend profile real Python code to inspect later.

• Save sandbox_workspace/app.py.
• Confirm that app.py appears beside agents.json in the fixture workspace.

Fixture showing a syntax problem?

Check the indentation beneath both function definitions. Confirm that the dictionary key inside the formatted string uses single quotes.

Ask for help comparing the checkout fixture.

✔️ Awesome, I've got everything!

Your checkout fixture is ready for restricted inspection.

ⓧ I'd like to double check the full code

Compare your fixture with this complete version.

def fetch_checkout_status(user_id: str) -> dict[str, str]:
    return {"user_id": user_id, "status": "ready"}


def render_checkout_message(status: dict[str, str]) -> str:
    return f"Checkout status: {status['status']}"

The baseline needs a strict output shape even though its routing logic is deliberately poor. Pydantic models make the resulting decision easy to serialize as JSON.

• In your editor's file sidebar, create router.py inside jev-agent-router.
• Paste these strict models and classification labels into router.py:

from enum import StrEnum

from pydantic import BaseModel, ConfigDict, Field


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid")


class Domain(StrEnum):
    FRONTEND = "frontend"
    BACKEND = "backend"
    DATABASE = "database"
    SECURITY = "security"


class Complexity(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class Risk(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

What do these contracts establish?

• The StrictModel base rejects fields outside a model's declared schema.
• The Domain enum limits ownership to four engineering specialties.
• The Complexity enum provides three implementation levels.
• The Risk enum provides three security levels.

• Save router.py.
• Confirm that your editor shows router.py at the top level of jev-agent-router.

Imports or enums underlined?

Confirm that your editor is using the activated .venv environment. Check that each enum member remains indented inside its class.

Ask for help checking the baseline contracts.

A route decision records whether work may execute. The naive function supplies the designed weakness by selecting frontend for every task.

• In router.py, place your cursor below the Risk enum.
• Append the baseline decision model and routing function:

class Action(StrEnum):
    EXECUTE = "execute"
    HUMAN_REVIEW = "human_review"


class RouteDecision(StrictModel):
    action: Action
    selected_agent: Domain | None = None
    review_agents: list[Domain] = Field(default_factory=list)
    reason: str


def naive_route(task: str) -> RouteDecision:
    return RouteDecision(action=Action.EXECUTE, selected_agent=Domain.FRONTEND, reason="naive_always_frontend_baseline")

How does the baseline decide?

• The Action enum separates execution from human review.
• The RouteDecision model records the action, selected agent, review agents, and reason.
• The naive_route() function ignores the task content.
• Every call returns the frontend agent with permission to execute.

• Save router.py.
• Confirm that your editor shows no unclosed brackets in RouteDecision or naive_route().

Baseline model showing an error?

Check that Action appears below Risk. Confirm that RouteDecision inherits from StrictModel.

Ask for help checking the route model and naive function.

✔️ Awesome, I've got everything!

Your strict decision model and intentionally naive route are complete.

ⓧ I'd like to double check the full code

Compare your complete router.py file with this version.

from enum import StrEnum

from pydantic import BaseModel, ConfigDict, Field


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid")


class Domain(StrEnum):
    FRONTEND = "frontend"
    BACKEND = "backend"
    DATABASE = "database"
    SECURITY = "security"


class Complexity(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class Risk(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class Action(StrEnum):
    EXECUTE = "execute"
    HUMAN_REVIEW = "human_review"


class RouteDecision(StrictModel):
    action: Action
    selected_agent: Domain | None = None
    review_agents: list[Domain] = Field(default_factory=list)
    reason: str


def naive_route(task: str) -> RouteDecision:
    return RouteDecision(action=Action.EXECUTE, selected_agent=Domain.FRONTEND, reason="naive_always_frontend_baseline")

The command-line entry point accepts a task as plain text. Its baseline flag prints the route decision as formatted JSON.

• In your editor's file sidebar, create demo.py inside jev-agent-router.
• Paste this baseline command-line interface into demo.py:

import argparse

from router import naive_route


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("task")
    parser.add_argument("--baseline", action="store_true")
    args = parser.parse_args()
    if args.baseline:
        print(naive_route(args.task).model_dump_json(indent=2))
        return


if __name__ == "__main__":
    main()

What does the baseline CLI do?

• The positional task argument captures the software request.
• The --baseline flag selects the naive routing path.
• The model_dump_json() call serializes the validated decision with indentation.
• The main guard runs the command-line interface when Python executes demo.py.

• Save demo.py.
• Confirm that demo.py appears beside router.py in jev-agent-router.

Command-line file showing an import problem?

Check that demo.py and router.py are in the same directory. Confirm that the import uses the exact function name naive_route.

Ask for help checking the baseline command-line interface.

✔️ Awesome, I've got everything!

Your baseline command-line interface is saved beside the router.

ⓧ I'd like to double check the full code

Compare your complete demo.py file with this version.

import argparse

from router import naive_route


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("task")
    parser.add_argument("--baseline", action="store_true")
    args = parser.parse_args()
    if args.baseline:
        print(naive_route(args.task).model_dump_json(indent=2))
        return


if __name__ == "__main__":
    main()

Run the designed misroute

The task clearly concerns checkout API failures. Before you run it, which specialist do you think the frontend-only baseline selects?

• Expose the baseline decision by running this command from jev-agent-router with .venv still activated:

python demo.py "Add retry handling around checkout API failures" --baseline

What does the result prove?

You will see a JSON decision with execute as the action. The selected agent is frontend.

The reason is naive_always_frontend_baseline. The valid output format cannot compensate for routing logic that ignores the task.

Baseline command not producing JSON?

Confirm that your terminal is inside jev-agent-router. Check that .venv remains activated before running the command again.

Ask for help diagnosing the baseline run.

You have caught the designed failure: a backend task receives a frontend assignment because the baseline ignores its meaning. Next, you will replace that guess with validated Jev evidence and a deterministic authorization gate.

Add Jev Classification and Fail-Closed Policy

Your baseline sent a backend retry task to the frontend agent. That visible failure proved a fixed route cannot govern mixed engineering work.

This step adds Jev classification as evidence. Pydantic validation checks that evidence before deterministic Python policy grants execution or requires human review.

In this step, get ready to:

• Classify tasks by domain, complexity, and security risk.
• Apply deterministic authorization rules that fail closed.
• Prove malformed data, classifier failures, and risky tasks cannot execute.

Build the classification evidence layer

A probabilistic classification includes a selected label plus evidence about uncertainty. The router records the lowest confidence across three questions so the weakest answer controls the safety decision.

• Return to router.py in your code editor.
• Replace its import section with the following imports:

from __future__ import annotations

import json
import os
import subprocess
from enum import StrEnum
from pathlib import Path
from typing import Any, Protocol

from pydantic import BaseModel, ConfigDict, Field
from typesafe_sdk import Choice, TypeSafeClient

What do these imports support?

• The TypeSafe Python SDK provides Choice questions plus the TypeSafeClient used to call Jev through OpenRouter.
• The typing imports define a classifier interface that accepts live or deterministic implementations.
• The remaining imports support structured results plus the later process boundary.

• Find the existing RouteDecision model.
• Replace that model with the following group:

class Classification(StrictModel):
    domain: Domain
    complexity: Complexity
    security_risk: Risk
    confidence: float = Field(ge=0.0, le=1.0)
    probabilities: dict[str, dict[str, float]]
    model: str
    jev_cost_usd: float | None = Field(default=None, ge=0.0)
    input_tokens: int | None = Field(default=None, ge=0)


class RouteDecision(StrictModel):
    action: Action
    selected_agent: Domain | None = None
    review_agents: list[Domain] = Field(default_factory=list)
    reason: str


class ExecutionResult(StrictModel):
    agent: Domain
    result: str
    claude_cost_usd: float | None = Field(default=None, ge=0.0)
    command: list[str]


class OrchestrationOutcome(StrictModel):
    task: str
    classification: Classification | None = None
    decision: RouteDecision
    execution: ExecutionResult | None = None

What do these contracts protect?

• Classification requires every label, confidence value, probability map, model name, cost value, and token count expected by policy.
• StrictModel rejects unexpected fields through its existing forbidden-extra configuration.
• OrchestrationOutcome keeps classification evidence separate from the authorization decision and any execution result.

• Save router.py.

Before you rerun the baseline, do you expect these new contracts to change its fixed frontend decision?

• Confirm the baseline still runs by executing:

python demo.py "Add retry handling around checkout API failures" --baseline

What does this check prove?

You still see an execution decision selecting frontend. The new contracts load successfully while the original baseline remains available for comparison.

Seeing an import or model error?

• Confirm the import section appears once at the top of router.py.
• Confirm Classification appears before OrchestrationOutcome.

Ask for help with the contract update.

• Add the classifier interface below OrchestrationOutcome by pasting:

class Classifier(Protocol):
    def classify(self, task: str) -> object: ...


class FixedClassifier:
    def __init__(self, value: object = None, error: Exception | None = None) -> None:
        self.value = value
        self.error = error

    def classify(self, task: str) -> object:
        if self.error is not None:
            raise self.error
        return self.value

Why use two classifier implementations?

Classifier describes the single method orchestration needs. FixedClassifier supplies repeatable data or a deliberate exception for tests.

This interface keeps policy tests independent from network availability and metered calls.

• Save router.py.
• Confirm the unchanged baseline still loads by running:

python demo.py "Add retry handling around checkout API failures" --baseline

What should you see?

You see the same frontend assignment. The deterministic test double now exists without changing baseline behavior.

• Paste the first half of JevClassifier below FixedClassifier using this block:

class JevClassifier:
    def classify(self, task: str) -> dict[str, Any]:
        with TypeSafeClient(
            api_key=os.environ["OPENROUTER_API_KEY"],
            base_url="https://openrouter.ai/api",
        ) as client:
            response = client.system_one(
                model="jev-1.13",
                state={"task": task},
                questions={
                    "domain": Choice(
                        instructions="Which engineering domain should own this task?",
                        criteria={
                            "frontend": "Browser UI, styling, accessibility or client interaction",
                            "backend": "Server logic, APIs, queues or integrations",
                            "database": "Schemas, migrations, indexes or query behavior",
                            "security": "Authentication, authorization, secrets or abuse prevention",
                        },
                    ),
                    "complexity": Choice(
                        instructions="What is the implementation complexity?",
                        criteria={
                            "low": "Localized and easy to verify",
                            "medium": "Touches several functions or needs integration checks",
                            "high": "Cross-cutting, architectural or difficult to reverse",
                        },
                    ),

How does Jev receive the task?

• TypeSafeClient reads OPENROUTER_API_KEY from the invoking shell.
• system_one() sends the task to model jev-1.13 with separate domain and complexity questions.
• Choice limits each answer to policy labels defined by the router.

• Complete the JevClassifier method by pasting this continuation immediately below the previous block:

                    "security_risk": Choice(
                        instructions="What security risk would executing this task create?",
                        criteria={
                            "low": "No meaningful change to trust boundaries or protected data",
                            "medium": "Touches sensitive flows but does not directly grant authority",
                            "high": "Changes authentication, authorization, secrets or privilege boundaries",
                        },
                    ),
                },
            )

        answers = {name: response.choices[name] for name in ("domain", "complexity", "security_risk")}
        usage = response.raw_http_response.json().get("usage", {})
        cost = usage.get("cost")
        return {
            "domain": answers["domain"].choice,
            "complexity": answers["complexity"].choice,
            "security_risk": answers["security_risk"].choice,
            "confidence": min(answer.confidence for answer in answers.values()),
            "probabilities": {name: dict(answer.probabilities) for name, answer in answers.items()},
            "model": response.model,
            "jev_cost_usd": float(cost) if cost is not None else None,
            "input_tokens": response.usage.input_tokens,
        }

How is classifier evidence recorded?

• The third Choice question estimates security risk independently from task ownership.
• The minimum confidence makes the least-certain answer control the policy threshold.
• The returned dictionary preserves full probabilities plus API-reported model, cost, and input-token data.

• Save router.py.
• Confirm the completed classifier did not break the baseline by running:

python demo.py "Add retry handling around checkout API failures" --baseline

What does the result confirm?

The baseline JSON still selects frontend. Python can now load the complete Jev classifier alongside the comparison route.

Baseline no longer loading?

• Check that the security_risk question remains inside the questions dictionary.
• Check that answers begins after the TypeSafeClient context closes.
• Check the indentation around the returned evidence dictionary.

Help me repair the Jev classifier structure.

Add deterministic authorization

Classification supplies evidence to the router. The authorization policy remains ordinary Python with explicit review conditions that can be tested line by line.

• Find the existing naive_route() function in router.py.
• Replace that function with the following policy block:

def review_agents(classification: Classification) -> list[Domain]:
    agents = [Domain.SECURITY, Domain.BACKEND]
    if classification.domain not in agents:
        agents.append(classification.domain)
    return agents


def authorize(classification: Classification) -> RouteDecision:
    if classification.confidence < 0.80:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="confidence_below_0.80")
    if classification.domain == Domain.SECURITY:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="security_domain_requires_human")
    if classification.security_risk == Risk.HIGH:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="high_security_risk")
    if classification.complexity == Complexity.HIGH and classification.security_risk == Risk.MEDIUM:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="high_complexity_medium_risk")
    return RouteDecision(action=Action.EXECUTE, selected_agent=classification.domain, reason="deterministic_policy_approved")


def naive_route(task: str) -> RouteDecision:
    return RouteDecision(action=Action.EXECUTE, selected_agent=Domain.FRONTEND, reason="naive_always_frontend_baseline")


def fail_closed(reason: str) -> RouteDecision:
    return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=[Domain.SECURITY, Domain.BACKEND], reason=reason)

How does the policy fail closed?

• Confidence below 0.80 requires human review.
• Security ownership or high security risk requires human review.
• High-complexity work with medium risk requires human review.
• Every remaining classification receives an explicit execution decision for its classified domain.

• Save router.py.
• Confirm the comparison route remains available by running:

python demo.py "Add retry handling around checkout API failures" --baseline

Why keep the weak baseline?

The output still selects frontend because naive_route() remains intentionally unchanged. Later evaluation can compare that fixed decision with governed routing.

The orchestrator needs a defined execution boundary even when policy blocks execution. This keeps classifier errors, review decisions, and executor failures inside one fail-closed outcome.

• Add the first half of ClaudeExecutor below fail_closed() using this block:

class ClaudeExecutor:
    SAFE_ENV_KEYS = ("PATH", "HOME", "LANG", "LC_ALL", "TERM", "TMPDIR")

    def __init__(self, workspace: Path) -> None:
        self.workspace = workspace.resolve()

    def run(self, agent: Domain, task: str) -> ExecutionResult:
        if not self.workspace.is_dir() or not (self.workspace / "agents.json").is_file():
            raise FileNotFoundError("sandbox workspace or agents.json is missing")
        command = [
            "claude", "-p", task,
            "--agents", "./agents.json",
            "--agent", agent.value,
            "--tools", "Read,Grep,Glob",
            "--disallowedTools", "mcp__*",
            "--restricted",
            "--permission-mode", "dontAsk",
            "--max-budget-usd", "0.25",
            "--max-turns", "2",
            "--output-format", "json",
        ]

What boundary does this define?

ClaudeExecutor resolves the fixture workspace before constructing a Claude Code argument list. The argument list keeps the selected agent plus runtime restrictions visible to tests.

• Complete ClaudeExecutor.run() by pasting this continuation below the command list:

        safe_env = {key: os.environ[key] for key in self.SAFE_ENV_KEYS if key in os.environ}
        completed = subprocess.run(
            command, cwd=self.workspace, env=safe_env, capture_output=True,
            text=True, timeout=120, check=True,
        )
        payload = json.loads(completed.stdout)
        cost = payload.get("total_cost_usd")
        return ExecutionResult(
            agent=agent,
            result=str(payload.get("result", "")),
            claude_cost_usd=float(cost) if cost is not None else None,
            command=command,
        )

How is the process isolated?

• The subprocess receives only allowlisted environment keys.
• OPENROUTER_API_KEY stays out of the delegated process environment.
• The JSON response becomes an ExecutionResult with the exact command recorded for auditing.

• Save router.py.
• Confirm the completed class still allows the baseline to load by running:

python demo.py "Add retry handling around checkout API failures" --baseline

What should remain unchanged?

The output remains the naive frontend decision. Defining the process boundary does not invoke Claude Code during this check.

• Add Orchestrator below ClaudeExecutor by pasting:

class Orchestrator:
    def __init__(self, classifier: Classifier, executor: ClaudeExecutor | None = None) -> None:
        self.classifier = classifier
        self.executor = executor

    def handle(self, task: str, execute: bool = False) -> OrchestrationOutcome:
        try:
            classification = Classification.model_validate(self.classifier.classify(task))
        except Exception as error:
            return OrchestrationOutcome(task=task, decision=fail_closed(f"classification_failed:{type(error).__name__}"))
        decision = authorize(classification)
        if not execute or decision.action == Action.HUMAN_REVIEW:
            return OrchestrationOutcome(task=task, classification=classification, decision=decision)
        try:
            if self.executor is None or decision.selected_agent is None:
                raise RuntimeError("executor unavailable")
            execution = self.executor.run(decision.selected_agent, task)
        except Exception as error:
            return OrchestrationOutcome(task=task, classification=classification, decision=fail_closed(f"execution_failed:{type(error).__name__}"))
        return OrchestrationOutcome(task=task, classification=classification, decision=decision, execution=execution)

How does orchestration preserve authority?

• Classification.model_validate() validates the classifier response before policy receives it.
• Any classification exception becomes a human-review decision through fail_closed().
• A review decision returns before the executor can run.
• An executor failure also becomes a fail-closed outcome.

• Save router.py.

✔️ Awesome, I've got everything!

Your governed router now separates evidence, validation, policy, orchestration, and execution results.

ⓧ I'd like to double check the full code

from __future__ import annotations

import json
import os
import subprocess
from enum import StrEnum
from pathlib import Path
from typing import Any, Protocol

from pydantic import BaseModel, ConfigDict, Field
from typesafe_sdk import Choice, TypeSafeClient


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid")


class Domain(StrEnum):
    FRONTEND = "frontend"
    BACKEND = "backend"
    DATABASE = "database"
    SECURITY = "security"


class Complexity(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class Risk(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class Action(StrEnum):
    EXECUTE = "execute"
    HUMAN_REVIEW = "human_review"


class Classification(StrictModel):
    domain: Domain
    complexity: Complexity
    security_risk: Risk
    confidence: float = Field(ge=0.0, le=1.0)
    probabilities: dict[str, dict[str, float]]
    model: str
    jev_cost_usd: float | None = Field(default=None, ge=0.0)
    input_tokens: int | None = Field(default=None, ge=0)


class RouteDecision(StrictModel):
    action: Action
    selected_agent: Domain | None = None
    review_agents: list[Domain] = Field(default_factory=list)
    reason: str


class ExecutionResult(StrictModel):
    agent: Domain
    result: str
    claude_cost_usd: float | None = Field(default=None, ge=0.0)
    command: list[str]


class OrchestrationOutcome(StrictModel):
    task: str
    classification: Classification | None = None
    decision: RouteDecision
    execution: ExecutionResult | None = None


class Classifier(Protocol):
    def classify(self, task: str) -> object: ...


class FixedClassifier:
    def __init__(self, value: object = None, error: Exception | None = None) -> None:
        self.value = value
        self.error = error

    def classify(self, task: str) -> object:
        if self.error is not None:
            raise self.error
        return self.value


class JevClassifier:
    def classify(self, task: str) -> dict[str, Any]:
        with TypeSafeClient(
            api_key=os.environ["OPENROUTER_API_KEY"],
            base_url="https://openrouter.ai/api",
        ) as client:
            response = client.system_one(
                model="jev-1.13",
                state={"task": task},
                questions={
                    "domain": Choice(
                        instructions="Which engineering domain should own this task?",
                        criteria={
                            "frontend": "Browser UI, styling, accessibility or client interaction",
                            "backend": "Server logic, APIs, queues or integrations",
                            "database": "Schemas, migrations, indexes or query behavior",
                            "security": "Authentication, authorization, secrets or abuse prevention",
                        },
                    ),
                    "complexity": Choice(
                        instructions="What is the implementation complexity?",
                        criteria={
                            "low": "Localized and easy to verify",
                            "medium": "Touches several functions or needs integration checks",
                            "high": "Cross-cutting, architectural or difficult to reverse",
                        },
                    ),
                    "security_risk": Choice(
                        instructions="What security risk would executing this task create?",
                        criteria={
                            "low": "No meaningful change to trust boundaries or protected data",
                            "medium": "Touches sensitive flows but does not directly grant authority",
                            "high": "Changes authentication, authorization, secrets or privilege boundaries",
                        },
                    ),
                },
            )

        answers = {name: response.choices[name] for name in ("domain", "complexity", "security_risk")}
        usage = response.raw_http_response.json().get("usage", {})
        cost = usage.get("cost")
        return {
            "domain": answers["domain"].choice,
            "complexity": answers["complexity"].choice,
            "security_risk": answers["security_risk"].choice,
            "confidence": min(answer.confidence for answer in answers.values()),
            "probabilities": {name: dict(answer.probabilities) for name, answer in answers.items()},
            "model": response.model,
            "jev_cost_usd": float(cost) if cost is not None else None,
            "input_tokens": response.usage.input_tokens,
        }


def review_agents(classification: Classification) -> list[Domain]:
    agents = [Domain.SECURITY, Domain.BACKEND]
    if classification.domain not in agents:
        agents.append(classification.domain)
    return agents


def authorize(classification: Classification) -> RouteDecision:
    if classification.confidence < 0.80:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="confidence_below_0.80")
    if classification.domain == Domain.SECURITY:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="security_domain_requires_human")
    if classification.security_risk == Risk.HIGH:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="high_security_risk")
    if classification.complexity == Complexity.HIGH and classification.security_risk == Risk.MEDIUM:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="high_complexity_medium_risk")
    return RouteDecision(action=Action.EXECUTE, selected_agent=classification.domain, reason="deterministic_policy_approved")


def naive_route(task: str) -> RouteDecision:
    return RouteDecision(action=Action.EXECUTE, selected_agent=Domain.FRONTEND, reason="naive_always_frontend_baseline")


def fail_closed(reason: str) -> RouteDecision:
    return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=[Domain.SECURITY, Domain.BACKEND], reason=reason)


class ClaudeExecutor:
    SAFE_ENV_KEYS = ("PATH", "HOME", "LANG", "LC_ALL", "TERM", "TMPDIR")

    def __init__(self, workspace: Path) -> None:
        self.workspace = workspace.resolve()

    def run(self, agent: Domain, task: str) -> ExecutionResult:
        if not self.workspace.is_dir() or not (self.workspace / "agents.json").is_file():
            raise FileNotFoundError("sandbox workspace or agents.json is missing")
        command = [
            "claude", "-p", task,
            "--agents", "./agents.json",
            "--agent", agent.value,
            "--tools", "Read,Grep,Glob",
            "--disallowedTools", "mcp__*",
            "--restricted",
            "--permission-mode", "dontAsk",
            "--max-budget-usd", "0.25",
            "--max-turns", "2",
            "--output-format", "json",
        ]
        safe_env = {key: os.environ[key] for key in self.SAFE_ENV_KEYS if key in os.environ}
        completed = subprocess.run(
            command, cwd=self.workspace, env=safe_env, capture_output=True,
            text=True, timeout=120, check=True,
        )
        payload = json.loads(completed.stdout)
        cost = payload.get("total_cost_usd")
        return ExecutionResult(
            agent=agent,
            result=str(payload.get("result", "")),
            claude_cost_usd=float(cost) if cost is not None else None,
            command=command,
        )


class Orchestrator:
    def __init__(self, classifier: Classifier, executor: ClaudeExecutor | None = None) -> None:
        self.classifier = classifier
        self.executor = executor

    def handle(self, task: str, execute: bool = False) -> OrchestrationOutcome:
        try:
            classification = Classification.model_validate(self.classifier.classify(task))
        except Exception as error:
            return OrchestrationOutcome(task=task, decision=fail_closed(f"classification_failed:{type(error).__name__}"))
        decision = authorize(classification)
        if not execute or decision.action == Action.HUMAN_REVIEW:
            return OrchestrationOutcome(task=task, classification=classification, decision=decision)
        try:
            if self.executor is None or decision.selected_agent is None:
                raise RuntimeError("executor unavailable")
            execution = self.executor.run(decision.selected_agent, task)
        except Exception as error:
            return OrchestrationOutcome(task=task, classification=classification, decision=fail_closed(f"execution_failed:{type(error).__name__}"))
        return OrchestrationOutcome(task=task, classification=classification, decision=decision, execution=execution)

What should match?

Use this full-file reference to compare class order, policy conditions, and indentation. Keep every identifier exactly as shown so the fixture and tests use the same contracts.

Prove fail-closed behavior

A deterministic fixture lets you exercise the governed route without spending tokens or depending on a live response. The test suite then attacks the validation and policy boundaries directly.

• Replace the contents of demo.py with the following code:

import argparse
from pathlib import Path

from router import ClaudeExecutor, FixedClassifier, JevClassifier, Orchestrator, naive_route


def fixture_classification() -> dict[str, object]:
    return {
        "domain": "backend", "complexity": "medium", "security_risk": "low", "confidence": 0.96,
        "probabilities": {"domain": {"backend": 0.96}, "complexity": {"medium": 0.87}, "security_risk": {"low": 0.97}},
        "model": "fixture", "jev_cost_usd": 0.0, "input_tokens": 0,
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("task")
    parser.add_argument("--baseline", action="store_true")
    parser.add_argument("--fixture", action="store_true")
    parser.add_argument("--execute", action="store_true")
    args = parser.parse_args()
    if args.baseline:
        print(naive_route(args.task).model_dump_json(indent=2))
        return
    classifier = FixedClassifier(fixture_classification()) if args.fixture else JevClassifier()
    outcome = Orchestrator(classifier, ClaudeExecutor(Path("sandbox_workspace"))).handle(args.task, execute=args.execute)
    print(outcome.model_dump_json(indent=2))


if __name__ == "__main__":
    main()

How does the governed demo work?

• fixture_classification() supplies a backend label with medium complexity, low risk, and 0.96 confidence.
• The --fixture path selects FixedClassifier for a repeatable local check.
• The default path selects JevClassifier for a live classification call.
• Orchestrator handles both paths through the same validation and authorization logic.

• Save demo.py.

Before you run the fixture, do you expect the retry task to execute as backend or stop for human review?

• Check the governed fixture route by running:

python demo.py "Add retry handling around checkout API failures" --fixture

What should the fixture return?

You see a backend classification with an execute action. The output also shows deterministic_policy_approved as the decision reason.

That is the baseline gap closed. The same backend task now receives a validated domain assignment before authorization.

Fixture route stopping unexpectedly?

• Confirm demo.py passes fixture_classification() into FixedClassifier.
• Confirm the fixture confidence is 0.96.
• Confirm the security risk is low.

Help me debug the fixture decision.

✔️ Awesome, I've got everything!

Your demo now supports baseline, fixture, and live classifier paths through one orchestrator.

ⓧ I'd like to double check the full code

import argparse
from pathlib import Path

from router import ClaudeExecutor, FixedClassifier, JevClassifier, Orchestrator, naive_route


def fixture_classification() -> dict[str, object]:
    return {
        "domain": "backend", "complexity": "medium", "security_risk": "low", "confidence": 0.96,
        "probabilities": {"domain": {"backend": 0.96}, "complexity": {"medium": 0.87}, "security_risk": {"low": 0.97}},
        "model": "fixture", "jev_cost_usd": 0.0, "input_tokens": 0,
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("task")
    parser.add_argument("--baseline", action="store_true")
    parser.add_argument("--fixture", action="store_true")
    parser.add_argument("--execute", action="store_true")
    args = parser.parse_args()
    if args.baseline:
        print(naive_route(args.task).model_dump_json(indent=2))
        return
    classifier = FixedClassifier(fixture_classification()) if args.fixture else JevClassifier()
    outcome = Orchestrator(classifier, ClaudeExecutor(Path("sandbox_workspace"))).handle(args.task, execute=args.execute)
    print(outcome.model_dump_json(indent=2))


if __name__ == "__main__":
    main()

What should match?

Compare the fixture values, parser flags, and orchestrator call with your file. The fixture path must stay deterministic so policy tests remain reproducible.

• Create tests/test_router.py inside the existing tests directory using your code editor's new-file control.
• Paste the test helpers into the new file:

import json
import subprocess

import router
from router import Action, ClaudeExecutor, Domain, ExecutionResult, FixedClassifier, Orchestrator, naive_route


def valid_classification(**updates):
    value = {
        "domain": "backend", "complexity": "medium", "security_risk": "low", "confidence": 0.95,
        "probabilities": {"domain": {"backend": 0.95}, "complexity": {"medium": 0.95}, "security_risk": {"low": 0.95}},
        "model": "fixture", "jev_cost_usd": 0.0, "input_tokens": 0,
    }
    value.update(updates)
    return value


class RecordingExecutor:
    def __init__(self): self.calls = []
    def run(self, agent, task):
        self.calls.append((agent, task))
        return ExecutionResult(agent=agent, result="ok", claude_cost_usd=0.0, command=["recorded"])

What do the test helpers control?

• valid_classification() starts with an approved backend classification.
• Each test can replace one field to isolate a policy condition.
• RecordingExecutor records calls without starting a real Claude Code process.

• Add the six policy tests below RecordingExecutor by pasting:

def test_naive_baseline_always_routes_frontend():
    assert naive_route("backend task").selected_agent == Domain.FRONTEND


def test_malformed_response_fails_closed():
    outcome = Orchestrator(FixedClassifier({"domain": "backend"})).handle("task")
    assert outcome.decision.action == Action.HUMAN_REVIEW and outcome.execution is None


def test_api_failure_fails_closed():
    outcome = Orchestrator(FixedClassifier(error=RuntimeError("api unavailable"))).handle("task")
    assert outcome.decision.action == Action.HUMAN_REVIEW and outcome.execution is None


def test_low_confidence_fails_closed():
    outcome = Orchestrator(FixedClassifier(valid_classification(confidence=0.79))).handle("task")
    assert outcome.decision.reason == "confidence_below_0.80"


def test_high_risk_fails_closed():
    outcome = Orchestrator(FixedClassifier(valid_classification(security_risk="high"))).handle("task")
    assert outcome.decision.action == Action.HUMAN_REVIEW
    assert Domain.SECURITY in outcome.decision.review_agents and Domain.BACKEND in outcome.decision.review_agents


def test_approved_route_reaches_executor():
    executor = RecordingExecutor()
    outcome = Orchestrator(FixedClassifier(valid_classification()), executor).handle("task", execute=True)
    assert outcome.decision.action == Action.EXECUTE and executor.calls == [(Domain.BACKEND, "task")]

Which boundaries do these tests prove?

• The first test preserves the deliberately weak frontend baseline.
• Malformed output and classifier exceptions both produce human review without execution.
• Confidence of 0.79 falls below the explicit 0.80 threshold.
• High risk escalates to security and backend reviewers while an approved backend task reaches the injected executor.

• Save tests/test_router.py.

Before you run the suite, do you expect any malformed, failed, uncertain, or high-risk classification to reach the recording executor?

• Run the fail-closed test suite by executing:

pytest -q

What should the tests prove?

You see all six tests pass. Only the approved backend classification reaches RecordingExecutor.

That is the security boundary expressed as repeatable evidence. Classifier output can inform execution without granting authority by itself.

Seeing a failing policy test?

• Read the first failed assertion to identify the policy branch that differs.
• Confirm Classification.model_validate() runs before authorize().
• Confirm review decisions return before the executor branch.

Help me trace the failing test.

✔️ Awesome, I've got everything!

Your six tests now cover the baseline, invalid evidence, classifier failure, uncertainty, high risk, and an approved backend route.

ⓧ I'd like to double check the full code

import json
import subprocess

import router
from router import Action, ClaudeExecutor, Domain, ExecutionResult, FixedClassifier, Orchestrator, naive_route


def valid_classification(**updates):
    value = {
        "domain": "backend", "complexity": "medium", "security_risk": "low", "confidence": 0.95,
        "probabilities": {"domain": {"backend": 0.95}, "complexity": {"medium": 0.95}, "security_risk": {"low": 0.95}},
        "model": "fixture", "jev_cost_usd": 0.0, "input_tokens": 0,
    }
    value.update(updates)
    return value


class RecordingExecutor:
    def __init__(self): self.calls = []
    def run(self, agent, task):
        self.calls.append((agent, task))
        return ExecutionResult(agent=agent, result="ok", claude_cost_usd=0.0, command=["recorded"])


def test_naive_baseline_always_routes_frontend():
    assert naive_route("backend task").selected_agent == Domain.FRONTEND


def test_malformed_response_fails_closed():
    outcome = Orchestrator(FixedClassifier({"domain": "backend"})).handle("task")
    assert outcome.decision.action == Action.HUMAN_REVIEW and outcome.execution is None


def test_api_failure_fails_closed():
    outcome = Orchestrator(FixedClassifier(error=RuntimeError("api unavailable"))).handle("task")
    assert outcome.decision.action == Action.HUMAN_REVIEW and outcome.execution is None


def test_low_confidence_fails_closed():
    outcome = Orchestrator(FixedClassifier(valid_classification(confidence=0.79))).handle("task")
    assert outcome.decision.reason == "confidence_below_0.80"


def test_high_risk_fails_closed():
    outcome = Orchestrator(FixedClassifier(valid_classification(security_risk="high"))).handle("task")
    assert outcome.decision.action == Action.HUMAN_REVIEW
    assert Domain.SECURITY in outcome.decision.review_agents and Domain.BACKEND in outcome.decision.review_agents


def test_approved_route_reaches_executor():
    executor = RecordingExecutor()
    outcome = Orchestrator(FixedClassifier(valid_classification()), executor).handle("task", execute=True)
    assert outcome.decision.action == Action.EXECUTE and executor.calls == [(Domain.BACKEND, "task")]

What should match?

Compare the helper values plus all six test functions with your file. The final test must be the only case that records an executor call.

You have replaced the fixed route with a governed path that validates evidence and fails closed under uncertainty. Next, you will prove those decisions remain enforced when a real Claude Code process starts.

Delegate to Restricted Claude Code Agents

Your router can now separate model evidence from execution authority. Approved tasks still need a runtime boundary before they can safely reach Claude Code.

A routing decision alone cannot restrict the process that performs the work. In this step, you will enforce read-only tools at launch. You will also isolate the workspace before parsing the process result as JSON.

In this step, get ready to:

• Launch the selected Claude Code profile through an isolated subprocess.
• Test the runtime restrictions without starting a real Claude Code process.
• Run one restricted backend delegation against the fixture workspace.

Build the runtime boundary

The executor turns an approved route into a process launch. Its argument list applies least privilege at runtime. Its environment allowlist keeps the OpenRouter credential in the parent shell.

• In router.py, replace the existing import os line with imports for json, os, and subprocess.
• Find the existing ClaudeExecutor declaration in router.py.
• Replace that declaration with the workspace validation and command construction below:

class ClaudeExecutor:
    SAFE_ENV_KEYS = ("PATH", "HOME", "LANG", "LC_ALL", "TERM", "TMPDIR")

    def __init__(self, workspace: Path) -> None:
        self.workspace = workspace.resolve()

    def run(self, agent: Domain, task: str) -> ExecutionResult:
        if not self.workspace.is_dir() or not (self.workspace / "agents.json").is_file():
            raise FileNotFoundError("sandbox workspace or agents.json is missing")
        command = [
            "claude", "-p", task,
            "--agents", "./agents.json",
            "--agent", agent.value,
            "--tools", "Read,Grep,Glob",
            "--disallowedTools", "mcp__*",
            "--restricted",
            "--permission-mode", "dontAsk",
            "--max-budget-usd", "0.25",
            "--max-turns", "2",
            "--output-format", "json",
        ]

How does the command enforce access?

• The resolved workspace gives the executor one concrete directory to validate.
• The agent definition file supplies the four specialist profiles from the previous step.
• The selected domain determines which profile Claude Code receives.
• The tool allowlist exposes only file inspection tools.
• The remaining flags deny MCP access. They also restrict permissions, turns, and estimated spend.

• Append the subprocess launch logic directly after the command list:

        safe_env = {key: os.environ[key] for key in self.SAFE_ENV_KEYS if key in os.environ}
        completed = subprocess.run(
            command, cwd=self.workspace, env=safe_env, capture_output=True,
            text=True, timeout=120, check=True,
        )
        payload = json.loads(completed.stdout)
        cost = payload.get("total_cost_usd")
        return ExecutionResult(
            agent=agent,
            result=str(payload.get("result", "")),
            claude_cost_usd=float(cost) if cost is not None else None,
            command=command,
        )

What does the launch logic protect?

• The environment comprehension copies only the six approved process variables.
• The argument list launches Claude Code without invoking a shell.
• The resolved workspace becomes the subprocess working directory.
• The timeout stops a process that exceeds 120 seconds.
• The JSON parser extracts the response text and client-side cost estimate.

• Save router.py.

Before you run the tests, do you expect the existing fail-closed policy checks to stay green after adding the real executor?

• Check the existing policy suite by running:

pytest -q

What did this check prove?

Your existing tests should still pass. The new executor has not weakened classification validation or authorization behavior.

Tests failing after the executor edit?

• Check that json and subprocess are imported beside os.
• Check that safe_env remains indented inside run().
• Check that the closing parenthesis for ExecutionResult aligns with its return statement.

Help me inspect my executor edit.

✔️ Awesome, I've got everything!

Great. Your router now has the runtime boundary that approved tasks need.

ⓧ I'd like to double check the full code

from __future__ import annotations

import json
import os
import subprocess
from enum import StrEnum
from pathlib import Path
from typing import Any, Protocol

from pydantic import BaseModel, ConfigDict, Field
from typesafe_sdk import Choice, TypeSafeClient


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid")


class Domain(StrEnum):
    FRONTEND = "frontend"
    BACKEND = "backend"
    DATABASE = "database"
    SECURITY = "security"


class Complexity(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class Risk(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class Action(StrEnum):
    EXECUTE = "execute"
    HUMAN_REVIEW = "human_review"


class Classification(StrictModel):
    domain: Domain
    complexity: Complexity
    security_risk: Risk
    confidence: float = Field(ge=0.0, le=1.0)
    probabilities: dict[str, dict[str, float]]
    model: str
    jev_cost_usd: float | None = Field(default=None, ge=0.0)
    input_tokens: int | None = Field(default=None, ge=0)


class RouteDecision(StrictModel):
    action: Action
    selected_agent: Domain | None = None
    review_agents: list[Domain] = Field(default_factory=list)
    reason: str


class ExecutionResult(StrictModel):
    agent: Domain
    result: str
    claude_cost_usd: float | None = Field(default=None, ge=0.0)
    command: list[str]


class OrchestrationOutcome(StrictModel):
    task: str
    classification: Classification | None = None
    decision: RouteDecision
    execution: ExecutionResult | None = None


class Classifier(Protocol):
    def classify(self, task: str) -> object: ...


class FixedClassifier:
    def __init__(self, value: object = None, error: Exception | None = None) -> None:
        self.value = value
        self.error = error

    def classify(self, task: str) -> object:
        if self.error is not None:
            raise self.error
        return self.value


class JevClassifier:
    def classify(self, task: str) -> dict[str, Any]:
        with TypeSafeClient(
            api_key=os.environ["OPENROUTER_API_KEY"],
            base_url="https://openrouter.ai/api",
        ) as client:
            response = client.system_one(
                model="jev-1.13",
                state={"task": task},
                questions={
                    "domain": Choice(
                        instructions="Which engineering domain should own this task?",
                        criteria={
                            "frontend": "Browser UI, styling, accessibility or client interaction",
                            "backend": "Server logic, APIs, queues or integrations",
                            "database": "Schemas, migrations, indexes or query behavior",
                            "security": "Authentication, authorization, secrets or abuse prevention",
                        },
                    ),
                    "complexity": Choice(
                        instructions="What is the implementation complexity?",
                        criteria={
                            "low": "Localized and easy to verify",
                            "medium": "Touches several functions or needs integration checks",
                            "high": "Cross-cutting, architectural or difficult to reverse",
                        },
                    ),
                    "security_risk": Choice(
                        instructions="What security risk would executing this task create?",
                        criteria={
                            "low": "No meaningful change to trust boundaries or protected data",
                            "medium": "Touches sensitive flows but does not directly grant authority",
                            "high": "Changes authentication, authorization, secrets or privilege boundaries",
                        },
                    ),
                },
            )

        answers = {name: response.choices[name] for name in ("domain", "complexity", "security_risk")}
        usage = response.raw_http_response.json().get("usage", {})
        cost = usage.get("cost")
        return {
            "domain": answers["domain"].choice,
            "complexity": answers["complexity"].choice,
            "security_risk": answers["security_risk"].choice,
            "confidence": min(answer.confidence for answer in answers.values()),
            "probabilities": {name: dict(answer.probabilities) for name, answer in answers.items()},
            "model": response.model,
            "jev_cost_usd": float(cost) if cost is not None else None,
            "input_tokens": response.usage.input_tokens,
        }


def review_agents(classification: Classification) -> list[Domain]:
    agents = [Domain.SECURITY, Domain.BACKEND]
    if classification.domain not in agents:
        agents.append(classification.domain)
    return agents


def authorize(classification: Classification) -> RouteDecision:
    if classification.confidence < 0.80:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="confidence_below_0.80")
    if classification.domain == Domain.SECURITY:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="security_domain_requires_human")
    if classification.security_risk == Risk.HIGH:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="high_security_risk")
    if classification.complexity == Complexity.HIGH and classification.security_risk == Risk.MEDIUM:
        return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=review_agents(classification), reason="high_complexity_medium_risk")
    return RouteDecision(action=Action.EXECUTE, selected_agent=classification.domain, reason="deterministic_policy_approved")


def naive_route(task: str) -> RouteDecision:
    return RouteDecision(action=Action.EXECUTE, selected_agent=Domain.FRONTEND, reason="naive_always_frontend_baseline")


def fail_closed(reason: str) -> RouteDecision:
    return RouteDecision(action=Action.HUMAN_REVIEW, review_agents=[Domain.SECURITY, Domain.BACKEND], reason=reason)


class ClaudeExecutor:
    SAFE_ENV_KEYS = ("PATH", "HOME", "LANG", "LC_ALL", "TERM", "TMPDIR")

    def __init__(self, workspace: Path) -> None:
        self.workspace = workspace.resolve()

    def run(self, agent: Domain, task: str) -> ExecutionResult:
        if not self.workspace.is_dir() or not (self.workspace / "agents.json").is_file():
            raise FileNotFoundError("sandbox workspace or agents.json is missing")
        command = [
            "claude", "-p", task,
            "--agents", "./agents.json",
            "--agent", agent.value,
            "--tools", "Read,Grep,Glob",
            "--disallowedTools", "mcp__*",
            "--restricted",
            "--permission-mode", "dontAsk",
            "--max-budget-usd", "0.25",
            "--max-turns", "2",
            "--output-format", "json",
        ]
        safe_env = {key: os.environ[key] for key in self.SAFE_ENV_KEYS if key in os.environ}
        completed = subprocess.run(
            command, cwd=self.workspace, env=safe_env, capture_output=True,
            text=True, timeout=120, check=True,
        )
        payload = json.loads(completed.stdout)
        cost = payload.get("total_cost_usd")
        return ExecutionResult(
            agent=agent,
            result=str(payload.get("result", "")),
            claude_cost_usd=float(cost) if cost is not None else None,
            command=command,
        )


class Orchestrator:
    def __init__(self, classifier: Classifier, executor: ClaudeExecutor | None = None) -> None:
        self.classifier = classifier
        self.executor = executor

    def handle(self, task: str, execute: bool = False) -> OrchestrationOutcome:
        try:
            classification = Classification.model_validate(self.classifier.classify(task))
        except Exception as error:
            return OrchestrationOutcome(task=task, decision=fail_closed(f"classification_failed:{type(error).__name__}"))
        decision = authorize(classification)
        if not execute or decision.action == Action.HUMAN_REVIEW:
            return OrchestrationOutcome(task=task, classification=classification, decision=decision)
        try:
            if self.executor is None or decision.selected_agent is None:
                raise RuntimeError("executor unavailable")
            execution = self.executor.run(decision.selected_agent, task)
        except Exception as error:
            return OrchestrationOutcome(task=task, classification=classification, decision=fail_closed(f"execution_failed:{type(error).__name__}"))
        return OrchestrationOutcome(task=task, classification=classification, decision=decision, execution=execution)

Prove the restrictions with a test

A command-construction test checks the security boundary without spending money or contacting Claude Code. The test replaces the real process launch with a recorder.

• In tests/test_router.py, replace the import section with this complete import group:

import json
import subprocess

import router
from router import Action, ClaudeExecutor, Domain, ExecutionResult, FixedClassifier, Orchestrator, naive_route

Why add these imports?

• The JSON module creates a realistic Claude Code response for the fake process.
• The subprocess module supplies the completed-process result returned by the fake launch.
• The router module gives the test a direct patch point for subprocess.run().
• The executor import lets the test invoke the real command-building logic.

• Add the runtime restriction test at the end of tests/test_router.py by copying this function:

def test_runtime_command_enforces_restrictions(monkeypatch, tmp_path):
    workspace = tmp_path / "sandbox_workspace"
    workspace.mkdir()
    (workspace / "agents.json").write_text("{}")
    captured = {}
    def fake_run(command, **kwargs):
        captured.update(command=command, kwargs=kwargs)
        return subprocess.CompletedProcess(command, 0, stdout=json.dumps({"result":"ok","total_cost_usd":0.01}), stderr="")
    monkeypatch.setenv("OPENROUTER_API_KEY", "must-not-reach-claude")
    monkeypatch.setattr(router.subprocess, "run", fake_run)
    result = ClaudeExecutor(workspace).run(Domain.BACKEND, "inspect app.py")
    command = captured["command"]
    assert command[command.index("--agent") + 1] == "backend"
    assert command[command.index("--tools") + 1] == "Read,Grep,Glob"
    assert command[command.index("--disallowedTools") + 1] == "mcp__*"
    assert "--restricted" in command
    assert command[command.index("--permission-mode") + 1] == "dontAsk"
    assert command[command.index("--max-budget-usd") + 1] == "0.25"
    assert captured["kwargs"]["cwd"] == workspace.resolve()
    assert "OPENROUTER_API_KEY" not in captured["kwargs"]["env"]
    assert result.result == "ok"

What does this test inspect?

• The temporary workspace proves that the executor accepts a resolved fixture directory containing agents.json.
• The patched process records every command argument without launching Claude Code.
• The assertions prove that tool restrictions and the budget cap reach the command.
• The environment assertion proves that OPENROUTER_API_KEY does not reach the child process.
• The final assertion proves that JSON output becomes an execution result.

• Save tests/test_router.py.

Before you run the suite, do you expect the fake subprocess to receive the OpenRouter key from the parent test environment?

• Verify the policy and runtime controls by running:

pytest -q

What should pass now?

All tests should pass. The new runtime test proves that approved execution receives the restricted command and the allowlisted environment.

Strong work. Your security claims are now backed by an exact command-construction test instead of prompt wording alone.

Runtime restriction test failing?

• Check that the fake function patches router.subprocess.run.
• Check that the temporary workspace contains agents.json before the executor runs.
• Check that OPENROUTER_API_KEY is absent from the captured env mapping.

Help me debug the runtime restriction test.

✔️ Awesome, I've got everything!

Your policy tests and runtime enforcement test are all passing.

ⓧ I'd like to double check the full code

import json
import subprocess

import router
from router import Action, ClaudeExecutor, Domain, ExecutionResult, FixedClassifier, Orchestrator, naive_route


def valid_classification(**updates):
    value = {
        "domain": "backend", "complexity": "medium", "security_risk": "low", "confidence": 0.95,
        "probabilities": {"domain": {"backend": 0.95}, "complexity": {"medium": 0.95}, "security_risk": {"low": 0.95}},
        "model": "fixture", "jev_cost_usd": 0.0, "input_tokens": 0,
    }
    value.update(updates)
    return value


class RecordingExecutor:
    def __init__(self): self.calls = []
    def run(self, agent, task):
        self.calls.append((agent, task))
        return ExecutionResult(agent=agent, result="ok", claude_cost_usd=0.0, command=["recorded"])


def test_naive_baseline_always_routes_frontend():
    assert naive_route("backend task").selected_agent == Domain.FRONTEND


def test_malformed_response_fails_closed():
    outcome = Orchestrator(FixedClassifier({"domain": "backend"})).handle("task")
    assert outcome.decision.action == Action.HUMAN_REVIEW and outcome.execution is None


def test_api_failure_fails_closed():
    outcome = Orchestrator(FixedClassifier(error=RuntimeError("api unavailable"))).handle("task")
    assert outcome.decision.action == Action.HUMAN_REVIEW and outcome.execution is None


def test_low_confidence_fails_closed():
    outcome = Orchestrator(FixedClassifier(valid_classification(confidence=0.79))).handle("task")
    assert outcome.decision.reason == "confidence_below_0.80"


def test_high_risk_fails_closed():
    outcome = Orchestrator(FixedClassifier(valid_classification(security_risk="high"))).handle("task")
    assert outcome.decision.action == Action.HUMAN_REVIEW
    assert Domain.SECURITY in outcome.decision.review_agents and Domain.BACKEND in outcome.decision.review_agents


def test_approved_route_reaches_executor():
    executor = RecordingExecutor()
    outcome = Orchestrator(FixedClassifier(valid_classification()), executor).handle("task", execute=True)
    assert outcome.decision.action == Action.EXECUTE and executor.calls == [(Domain.BACKEND, "task")]


def test_runtime_command_enforces_restrictions(monkeypatch, tmp_path):
    workspace = tmp_path / "sandbox_workspace"
    workspace.mkdir()
    (workspace / "agents.json").write_text("{}")
    captured = {}
    def fake_run(command, **kwargs):
        captured.update(command=command, kwargs=kwargs)
        return subprocess.CompletedProcess(command, 0, stdout=json.dumps({"result":"ok","total_cost_usd":0.01}), stderr="")
    monkeypatch.setenv("OPENROUTER_API_KEY", "must-not-reach-claude")
    monkeypatch.setattr(router.subprocess, "run", fake_run)
    result = ClaudeExecutor(workspace).run(Domain.BACKEND, "inspect app.py")
    command = captured["command"]
    assert command[command.index("--agent") + 1] == "backend"
    assert command[command.index("--tools") + 1] == "Read,Grep,Glob"
    assert command[command.index("--disallowedTools") + 1] == "mcp__*"
    assert "--restricted" in command
    assert command[command.index("--permission-mode") + 1] == "dontAsk"
    assert command[command.index("--max-budget-usd") + 1] == "0.25"
    assert captured["kwargs"]["cwd"] == workspace.resolve()
    assert "OPENROUTER_API_KEY" not in captured["kwargs"]["env"]
    assert result.result == "ok"

Run the restricted delegation

The deterministic fixture approves the backend profile. The executor can now send that profile a read-only inspection task inside sandbox_workspace.

The budget cap is a guardrail

This command starts one real Claude Code run. The client-side $0.25 cap can be exceeded slightly before enforcement.

The process has a 120-second timeout. Allow up to two minutes for the response.

Before you run this, what evidence should prove that the backend profile stayed inside its read-only boundary?

• Run the deterministic fixture delegation with the executor enabled:

python demo.py "Inspect app.py and explain where retry handling belongs. Do not modify files." --fixture --execute

What does the result prove?

The output should contain a backend execution result based on app.py. Its command record should expose every launch-time restriction.

• The execution agent is backend.
• The command selects the definitions in ./agents.json.
• The --tools value is Read,Grep,Glob.
• The --disallowedTools value is mcp__*.
• The command includes --restricted.
• The permission mode is dontAsk.
• The budget value is 0.25.

That is the runtime boundary working. Claude Code inspected the fixture through the backend profile without receiving modification tools.

Delegation did not return a result?

• Confirm that your authenticated Claude Code installation is still available in the activated terminal.
• Confirm that sandbox_workspace/agents.json still contains all four agent definitions.
• Confirm that sandbox_workspace/app.py still exists for the read-only inspection.

Help me debug the restricted delegation.

• Optionally remove --fixture from the command you just ran to place live Jev classification before the same runtime gate.

Your router can now turn an approved classification into a restricted process without exposing the OpenRouter key. Next, you will measure how well the complete control plane routes and escalates a labeled dataset.

Evaluate and Document the Router

Your restricted Claude Code delegation now proves that approved work can reach a read-only agent. One successful run cannot show how consistently the router handles a broader set of tasks.

A portfolio-ready router needs repeatable evidence. You will compare the governed route against an always-frontend baseline across labeled tasks.

You will also document how Jev evidence stays separate from execution authority. The finished repository will show its measurements plus its security boundaries.

In this step, get ready to:

• Create a labeled dataset covering frontend, backend, database, and human-review outcomes.
• Measure baseline accuracy plus governed routing behavior.
• Document the architecture plus its reproducibility limits.

Create the labeled evaluation dataset

A labeled dataset gives each task an expected destination. Its fixture classifications make the default benchmark deterministic without calling an external model.

The JSON objects also preserve the confidence plus probability evidence that reaches your policy layer.

• In your code editor, create dataset.json inside the jev-agent-router folder.
• Paste the following labeled cases into dataset.json:

[
  {"task":"Center the checkout button and improve keyboard focus.","expected_target":"frontend","classification":{"domain":"frontend","complexity":"low","security_risk":"low","confidence":0.97,"probabilities":{"domain":{"frontend":0.97},"complexity":{"low":0.96},"security_risk":{"low":0.98}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Add retry handling around checkout API failures.","expected_target":"backend","classification":{"domain":"backend","complexity":"medium","security_risk":"low","confidence":0.96,"probabilities":{"domain":{"backend":0.96},"complexity":{"medium":0.87},"security_risk":{"low":0.97}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Add an index for orders.customer_id and explain query impact.","expected_target":"database","classification":{"domain":"database","complexity":"medium","security_risk":"medium","confidence":0.92,"probabilities":{"domain":{"database":0.93},"complexity":{"medium":0.85},"security_risk":{"medium":0.89}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Review password reset logs for token leakage.","expected_target":"human","classification":{"domain":"security","complexity":"high","security_risk":"high","confidence":0.95,"probabilities":{"domain":{"security":0.95},"complexity":{"high":0.90},"security_risk":{"high":0.96}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Fix CORS handling for the internal reports API.","expected_target":"backend","classification":{"domain":"backend","complexity":"medium","security_risk":"medium","confidence":0.90,"probabilities":{"domain":{"backend":0.91},"complexity":{"medium":0.88},"security_risk":{"medium":0.90}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Update dashboard chart colors to meet contrast guidance.","expected_target":"frontend","classification":{"domain":"frontend","complexity":"low","security_risk":"low","confidence":0.98,"probabilities":{"domain":{"frontend":0.98},"complexity":{"low":0.97},"security_risk":{"low":0.99}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Write a query that finds duplicate invoice references.","expected_target":"database","classification":{"domain":"database","complexity":"low","security_risk":"low","confidence":0.94,"probabilities":{"domain":{"database":0.95},"complexity":{"low":0.94},"security_risk":{"low":0.96}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Audit admin authorization checks before a role migration.","expected_target":"human","classification":{"domain":"security","complexity":"high","security_risk":"high","confidence":0.97,"probabilities":{"domain":{"security":0.97},"complexity":{"high":0.94},"security_risk":{"high":0.98}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}}
]

How is the dataset balanced?

• Two tasks have frontend as their expected destination.
• Two tasks have backend as their expected destination.
• Two tasks have database as their expected destination.
• Two security-sensitive tasks use human as their expected destination.

• Save dataset.json.
• Count the top-level objects inside the outer array.

You should find eight objects. Each object contains a task, its expected target, and deterministic classification evidence.

Dataset structure looks broken?

Check that every object ends with a comma except the final object. Confirm that one outer pair of square brackets surrounds all eight objects.

Use Help me find the JSON structure problem in my dataset..

✔️ Awesome, I've got everything!

Your eight labeled cases are saved in dataset.json.

ⓧ I'd like to double check the full code

[
  {"task":"Center the checkout button and improve keyboard focus.","expected_target":"frontend","classification":{"domain":"frontend","complexity":"low","security_risk":"low","confidence":0.97,"probabilities":{"domain":{"frontend":0.97},"complexity":{"low":0.96},"security_risk":{"low":0.98}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Add retry handling around checkout API failures.","expected_target":"backend","classification":{"domain":"backend","complexity":"medium","security_risk":"low","confidence":0.96,"probabilities":{"domain":{"backend":0.96},"complexity":{"medium":0.87},"security_risk":{"low":0.97}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Add an index for orders.customer_id and explain query impact.","expected_target":"database","classification":{"domain":"database","complexity":"medium","security_risk":"medium","confidence":0.92,"probabilities":{"domain":{"database":0.93},"complexity":{"medium":0.85},"security_risk":{"medium":0.89}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Review password reset logs for token leakage.","expected_target":"human","classification":{"domain":"security","complexity":"high","security_risk":"high","confidence":0.95,"probabilities":{"domain":{"security":0.95},"complexity":{"high":0.90},"security_risk":{"high":0.96}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Fix CORS handling for the internal reports API.","expected_target":"backend","classification":{"domain":"backend","complexity":"medium","security_risk":"medium","confidence":0.90,"probabilities":{"domain":{"backend":0.91},"complexity":{"medium":0.88},"security_risk":{"medium":0.90}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Update dashboard chart colors to meet contrast guidance.","expected_target":"frontend","classification":{"domain":"frontend","complexity":"low","security_risk":"low","confidence":0.98,"probabilities":{"domain":{"frontend":0.98},"complexity":{"low":0.97},"security_risk":{"low":0.99}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Write a query that finds duplicate invoice references.","expected_target":"database","classification":{"domain":"database","complexity":"low","security_risk":"low","confidence":0.94,"probabilities":{"domain":{"database":0.95},"complexity":{"low":0.94},"security_risk":{"low":0.96}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}},
  {"task":"Audit admin authorization checks before a role migration.","expected_target":"human","classification":{"domain":"security","complexity":"high","security_risk":"high","confidence":0.97,"probabilities":{"domain":{"security":0.97},"complexity":{"high":0.94},"security_risk":{"high":0.98}},"model":"fixture","jev_cost_usd":0.0,"input_tokens":0}}
]

Build the batch evaluator

The evaluator measures routing accuracy without supplying a Claude executor. Every batch item stops after classification plus deterministic policy.

Fixture mode produces a reproducible benchmark. Live mode swaps each fixture classifier for a current Jev call while preserving the same policy path.

• Create evaluate.py inside the jev-agent-router folder.
• Add the imports, prediction helper, and evaluation state by pasting this first section:

import argparse
import json
import time
from collections import Counter
from pathlib import Path
from statistics import mean

from router import FixedClassifier, JevClassifier, Orchestrator


def prediction(outcome) -> str:
    return outcome.decision.selected_agent.value if outcome.decision.selected_agent is not None else "human"


def evaluate(live: bool, dataset: Path, output: Path) -> None:
    items = json.loads(dataset.read_text())
    baseline_correct = governed_correct = security_expected = security_escalated = 0
    assignments: Counter[str] = Counter()
    latencies_ms: list[float] = []
    jev_cost_usd = 0.0
    input_tokens = failures = 0
    rows: list[tuple[str, str, str, str]] = []
    live_classifier = JevClassifier() if live else None

What does this section set up?

• The prediction() helper returns a selected agent or the label human.
• The counters hold accuracy, escalation, assignment, latency, token, cost, and failure measurements.
• The live_classifier remains empty in deterministic fixture mode.

• Continue inside evaluate() by pasting the comparison loop below live_classifier:

    for item in items:
        task, expected = item["task"], item["expected_target"]
        baseline = "frontend"
        baseline_correct += int(baseline == expected)
        classifier = live_classifier if live else FixedClassifier(item["classification"])
        started = time.perf_counter()
        outcome = Orchestrator(classifier).handle(task)
        latencies_ms.append((time.perf_counter() - started) * 1000)
        governed = prediction(outcome)
        governed_correct += int(governed == expected)
        assignments[governed] += 1
        if expected == "human":
            security_expected += 1
            security_escalated += int(governed == "human")
        if outcome.classification is None:
            failures += 1
        else:
            jev_cost_usd += outcome.classification.jev_cost_usd or 0.0
            input_tokens += outcome.classification.input_tokens or 0
        rows.append((task, expected, baseline, governed))

How does the comparison loop work?

• The baseline assigns every task to frontend.
• Fixture mode injects the classification stored with each dataset item.
• Live mode reuses one JevClassifier across the dataset.
• The orchestrator receives no executor. Batch evaluation therefore cannot delegate work to Claude Code.

• Finish evaluate() by adding the report builder below the comparison loop:

    count = len(items)
    recall = security_escalated / security_expected if security_expected else 1.0
    lines = [
        "# Evaluation Report", "",
        f"Mode: {'live Jev' if live else 'deterministic fixture'}",
        f"Dataset: {dataset}", f"Dataset size: {count}",
        f"Baseline routing accuracy: {baseline_correct / count:.1%}",
        f"Governed routing accuracy: {governed_correct / count:.1%}",
        f"Security escalation recall: {recall:.1%}",
        f"Mean classification and policy latency: {mean(latencies_ms):.2f} ms",
        f"Jev input tokens: {input_tokens}", f"Reported Jev cost: ${jev_cost_usd:.8f}",
        f"Fail-closed classifications: {failures}",
        f"Agent assignment counts: `{json.dumps(dict(sorted(assignments.items())))}`", "",
        "| Task | Expected | Baseline | Governed |", "| --- | --- | --- | --- |",
    ]
    lines.extend(f"| {task} | {expected} | {baseline} | {governed} |" for task, expected, baseline, governed in rows)
    lines.extend(["", "## Interpretation", "", "Fixture mode costs $0 and is reproducible. Live mode records API-reported Jev cost and may vary. Batch evaluation does not execute Claude Code agents."])
    output.write_text("\n".join(lines) + "\n")
    print(output)

What goes into the report?

• Accuracy compares each predicted destination with expected_target.
• Escalation recall measures how many human-review cases the governed router blocks.
• The report records assignment counts plus mean policy-path latency.
• Live classifications contribute API-reported token usage plus Jev cost.

• Add the command-line entry point at the bottom of evaluate.py:

def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--live", action="store_true")
    parser.add_argument("--dataset", default="dataset.json")
    parser.add_argument("--output", default="evaluation-report.md")
    args = parser.parse_args()
    evaluate(args.live, Path(args.dataset), Path(args.output))


if __name__ == "__main__":
    main()

How does the evaluator stay reusable?

• The --live flag selects current Jev classification.
• The --dataset option accepts another labeled dataset.
• The --output option selects a report path.

• Save evaluate.py.

✔️ Awesome, I've got everything!

Your evaluator now supports deterministic fixtures plus optional live classification.

ⓧ I'd like to double check the full code

import argparse
import json
import time
from collections import Counter
from pathlib import Path
from statistics import mean

from router import FixedClassifier, JevClassifier, Orchestrator


def prediction(outcome) -> str:
    return outcome.decision.selected_agent.value if outcome.decision.selected_agent is not None else "human"


def evaluate(live: bool, dataset: Path, output: Path) -> None:
    items = json.loads(dataset.read_text())
    baseline_correct = governed_correct = security_expected = security_escalated = 0
    assignments: Counter[str] = Counter()
    latencies_ms: list[float] = []
    jev_cost_usd = 0.0
    input_tokens = failures = 0
    rows: list[tuple[str, str, str, str]] = []
    live_classifier = JevClassifier() if live else None

    for item in items:
        task, expected = item["task"], item["expected_target"]
        baseline = "frontend"
        baseline_correct += int(baseline == expected)
        classifier = live_classifier if live else FixedClassifier(item["classification"])
        started = time.perf_counter()
        outcome = Orchestrator(classifier).handle(task)
        latencies_ms.append((time.perf_counter() - started) * 1000)
        governed = prediction(outcome)
        governed_correct += int(governed == expected)
        assignments[governed] += 1
        if expected == "human":
            security_expected += 1
            security_escalated += int(governed == "human")
        if outcome.classification is None:
            failures += 1
        else:
            jev_cost_usd += outcome.classification.jev_cost_usd or 0.0
            input_tokens += outcome.classification.input_tokens or 0
        rows.append((task, expected, baseline, governed))

    count = len(items)
    recall = security_escalated / security_expected if security_expected else 1.0
    lines = [
        "# Evaluation Report", "",
        f"Mode: {'live Jev' if live else 'deterministic fixture'}",
        f"Dataset: {dataset}", f"Dataset size: {count}",
        f"Baseline routing accuracy: {baseline_correct / count:.1%}",
        f"Governed routing accuracy: {governed_correct / count:.1%}",
        f"Security escalation recall: {recall:.1%}",
        f"Mean classification and policy latency: {mean(latencies_ms):.2f} ms",
        f"Jev input tokens: {input_tokens}", f"Reported Jev cost: ${jev_cost_usd:.8f}",
        f"Fail-closed classifications: {failures}",
        f"Agent assignment counts: `{json.dumps(dict(sorted(assignments.items())))}`", "",
        "| Task | Expected | Baseline | Governed |", "| --- | --- | --- | --- |",
    ]
    lines.extend(f"| {task} | {expected} | {baseline} | {governed} |" for task, expected, baseline, governed in rows)
    lines.extend(["", "## Interpretation", "", "Fixture mode costs $0 and is reproducible. Live mode records API-reported Jev cost and may vary. Batch evaluation does not execute Claude Code agents."])
    output.write_text("\n".join(lines) + "\n")
    print(output)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--live", action="store_true")
    parser.add_argument("--dataset", default="dataset.json")
    parser.add_argument("--output", default="evaluation-report.md")
    args = parser.parse_args()
    evaluate(args.live, Path(args.dataset), Path(args.output))


if __name__ == "__main__":
    main()

Before you run the fixture benchmark, what accuracy do you expect from an always-frontend route across this balanced dataset?

• Generate the deterministic evaluation report by running this command:

python evaluate.py

What does this command measure?

The command loads the fixture classifications from dataset.json. It applies the same Pydantic validation plus deterministic authorization used by the demo.

No executor enters the evaluation path. The benchmark measures routing decisions without launching Claude Code.

You should see evaluation-report.md printed in the terminal. The new report should show a dataset size of 8.

The baseline accuracy should be 25.0%. Governed accuracy plus security escalation recall should both be 100.0%.

Report was not generated?

Confirm that your activated terminal is inside jev-agent-router. Check that dataset.json sits beside evaluate.py.

Use Help me debug my missing evaluation report..

That benchmark now proves the governed router handles every fixture case correctly. The optional live run makes metered Jev calls from your parent shell.

• Repeat the evaluation with current Jev classifications by running this optional command:

python evaluate.py --live

What changes in live mode?

Live mode replaces each fixture classifier with JevClassifier. The generated report records current latency, input tokens, and API-reported cost.

The policy stays deterministic. Live classifications can vary because they come from current model responses.

Live evaluation did not complete?

Confirm that OPENROUTER_API_KEY remains available in the activated parent shell. The evaluator needs it only when --live is present.

Use Help me debug my live Jev evaluation..

Document the router for demonstration

The report proves the result. Your README explains how the architecture produced it.

A Mermaid diagram makes the execution split visible. Explicit threat boundaries show which controls enforce the split.

• Create README.md inside the jev-agent-router folder.
• Add the project summary, architecture, and security boundary by pasting this first section:

# Jev-Gated Claude Code Agent Router

A local reference implementation that separates probabilistic classification, schema validation, deterministic authorization and agent execution.

## Architecture

```mermaid
flowchart LR
    T[Task] --> J[Jev 1.13]
    J --> P[Pydantic validation]
    P --> A[Deterministic authorization]
    A -->|approved| C[Claude Code profile]
    A -->|risk or uncertainty| H[Human review]
    C --> S[Restricted fixture workspace]
```

## Security boundary

- Jev supplies evidence. It never authorizes execution.
- Pydantic rejects malformed data.
- Python policy routes low confidence and risky work to a human.
- Claude Code receives only `Read`, `Grep` and `Glob`; MCP is denied; restricted mode confines file tools; the subprocess receives no `OPENROUTER_API_KEY`.
- Batch evaluation never executes Claude Code.

What does the architecture communicate?

The diagram separates model evidence from deterministic authorization. It gives approved work one path plus uncertain or risky work another path.

The security list identifies runtime controls. Prompt text provides agent guidance without becoming the enforcement boundary.

• Save README.md.
• Open a rendered preview of README.md using your editor's preview control.

You should see a flowchart that splits approved work toward a Claude Code profile. The other branch should end at human review.

Diagram does not render?

Confirm that both Mermaid fence markers use three backticks. Check that every flowchart line stays inside those markers.

Use Help me inspect my Mermaid flowchart..

• Return to README.md in your editor.
• Add the setup plus demonstration instructions below the security boundary section:

## Setup on macOS

```bash
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
claude --version
claude auth status
export OPENROUTER_API_KEY="your-api-key-here"
```

Use Claude Code 2.1.281 or newer. Run `claude update` if needed. Keep the key in the shell, not repository files.

## Demo and tests

```bash
python demo.py "Add retry handling around checkout API failures" --baseline
python demo.py "Add retry handling around checkout API failures" --fixture
pytest -q
python demo.py "Inspect app.py and explain where retry handling belongs. Do not modify files." --fixture --execute
```

The last command performs real restricted delegation. Remove `--fixture` to put live Jev classification before execution. A live route can stop for review by design.

Why include both demo paths?

The baseline command reproduces the routing weakness. The fixture command shows the deterministic governed result.

The test command verifies the fail-closed controls. The final demo performs the one explicit restricted delegation.

• Save README.md.
• Refresh the rendered preview.

You should now see separate setup plus demonstration sections. The demo section should contain four commands.

README code blocks run together?

Check that each shell section has an opening bash fence plus a closing fence. Keep the explanatory sentences outside those fences.

Use Help me fix the fenced blocks in my README..

• Return to the bottom of README.md.
• Add the evaluation plus reproducibility notes by pasting this final section:

## Evaluation

```bash
python evaluate.py
python evaluate.py --live
```

Fixture mode is deterministic and costs $0. Live mode uses `jev-1.13`, records latency and reads OpenRouter `usage.cost`. Claude JSON includes the client-side `total_cost_usd` estimate for the single execution demo.

## Reproducibility

Dependencies and request model are pinned. Fixture metrics are stable. Live Jev results can vary, and Claude billing depends on authentication and selected model.

What do these final notes establish?

The evaluation section separates the reproducible fixture benchmark from metered live classification. It also identifies which cost values come from each service response.

The reproducibility section states what remains fixed. It also identifies the live factors that can vary between demonstrations.

• Save README.md.

Documentation missing a section?

Compare the rendered headings with Architecture, Security boundary, Setup on macOS, Demo and tests, Evaluation, and Reproducibility. Add any missing section from the full-file reference below.

Use Help me compare my README with the required project sections..

✔️ Awesome, I've got everything!

Your README now documents the router's architecture, controls, commands, costs, and reproducibility limits.

ⓧ I'd like to double check the full code

# Jev-Gated Claude Code Agent Router

A local reference implementation that separates probabilistic classification, schema validation, deterministic authorization and agent execution.

## Architecture

```mermaid
flowchart LR
    T[Task] --> J[Jev 1.13]
    J --> P[Pydantic validation]
    P --> A[Deterministic authorization]
    A -->|approved| C[Claude Code profile]
    A -->|risk or uncertainty| H[Human review]
    C --> S[Restricted fixture workspace]
```

## Security boundary

- Jev supplies evidence. It never authorizes execution.
- Pydantic rejects malformed data.
- Python policy routes low confidence and risky work to a human.
- Claude Code receives only `Read`, `Grep` and `Glob`; MCP is denied; restricted mode confines file tools; the subprocess receives no `OPENROUTER_API_KEY`.
- Batch evaluation never executes Claude Code.

## Setup on macOS

```bash
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
claude --version
claude auth status
export OPENROUTER_API_KEY="your-api-key-here"
```

Use Claude Code 2.1.281 or newer. Run `claude update` if needed. Keep the key in the shell, not repository files.

## Demo and tests

```bash
python demo.py "Add retry handling around checkout API failures" --baseline
python demo.py "Add retry handling around checkout API failures" --fixture
pytest -q
python demo.py "Inspect app.py and explain where retry handling belongs. Do not modify files." --fixture --execute
```

The last command performs real restricted delegation. Remove `--fixture` to put live Jev classification before execution. A live route can stop for review by design.

## Evaluation

```bash
python evaluate.py
python evaluate.py --live
```

Fixture mode is deterministic and costs $0. Live mode uses `jev-1.13`, records latency and reads OpenRouter `usage.cost`. Claude JSON includes the client-side `total_cost_usd` estimate for the single execution demo.

## Reproducibility

Dependencies and request model are pinned. Fixture metrics are stable. Live Jev results can vary, and Claude billing depends on authentication and selected model.

Before the final check, do you expect the documentation to show model classification as evidence or as execution authority?

• Open the rendered preview of README.md.
• Return to evaluation-report.md in your editor.

The README should place deterministic authorization between Jev plus Claude Code. The report should contain both accuracy measurements, escalation recall, assignment counts, latency, tokens, cost, failure count, and the per-task table.

Strong finish. Your repository now pairs enforced runtime boundaries with repeatable evidence that the governed router improves on its frontend-only baseline.

Secret mission

Catch a Disguised Authorization Bug

A checkout button bug conceals a request to change a user's role before checkout. Build an adversarial dataset to test whether your deterministic gate catches the hidden authority change before delegation.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep the local router, pause its active shell state, or delete the project entirely. Your local files have no ongoing cost.

Cost warning

Live OpenRouter requests are metered when you run them. Real Claude Code delegations are also metered.

The --max-budget-usd 0.25 limit is a client-side estimate. A real Claude Code run can pass it slightly before stopping.

Batch evaluation does not execute Claude Code agents. Pause and Delete both clear the active shell state when you finish.

Resources you used:

• The .venv virtual environment with the pinned Python dependencies.
• The router source files router.py, demo.py, evaluate.py, requirements.txt, and .gitignore.
• The tests/ folder with the fail-closed and runtime enforcement test suite.
• The sandbox_workspace/ folder with the read-only agent definitions and checkout fixture.
• The labeled datasets dataset.json and secret-dataset.json.
• The generated reports evaluation-report.md and secret-report.md.
• The portfolio documentation in README.md.
• The shell-scoped OPENROUTER_API_KEY environment variable.

Keep everything running

No action needed. Choose this if you are still evaluating the router or keeping the repository as a portfolio artifact.

• Retain evaluation-report.md as the reproducible main benchmark.
• Retain secret-report.md as evidence that high security risk overrides a frontend classification.
• Keep .venv if you plan to rerun the pinned tests or fixture evaluations.
• Avoid live evaluation or execution until you are ready for another metered call.

Pause - I'll come back to this later

Clear the active shell state while keeping every project file. This prevents later commands in the same shell from using the OpenRouter key.

• Keep the jev-agent-router folder unchanged.
• Deactivate the virtual environment and remove the shell-scoped API key by running these commands:

deactivate
unset OPENROUTER_API_KEY

What does pausing change?

• The deactivate command exits the active .venv environment in the current shell.
• The unset OPENROUTER_API_KEY command removes the API key variable from the current shell.
• Your source files, datasets, tests, reports, and documentation remain available.

Delete - I don't want to use this again

Deleting the project folder removes the router, tests, fixture workspace, datasets, and generated reports. Your existing OpenRouter account and Claude Code installation remain untouched.

• Clear the active virtual environment and shell-scoped API key by running these commands:

deactivate
unset OPENROUTER_API_KEY

What gets cleared first?

• The first command exits the active .venv environment.
• The second command removes OPENROUTER_API_KEY from the current shell.
• Neither command deletes your local project files.

Delete the local artifacts:

• Use Finder to locate the jev-agent-router folder.
• Delete the jev-agent-router folder through Finder.
• Confirm the jev-agent-router folder no longer appears in its previous location.

Nice Work!

Nice Work!

Mission complete! You built a governed local agent router that keeps probabilistic evidence separate from execution authority.

You've learned how to:

• Use Jev as probabilistic classification evidence. Validate its output with Pydantic. Apply deterministic authorization before an approved task reaches Claude Code.
• Prove fail-closed control flow with automated tests for malformed output. Confirm the same protection for API failures. Confirm it again for low confidence. Confirm it again for high security risk. Enforce least privilege through read-only profiles. Isolate approved execution inside the fixture workspace.
• Measure routing accuracy. Measure escalation recall. Track assignment counts. Report latency. Report token usage. Report cost. Present the results in a generated evaluation report. Document the architecture in a portfolio-ready README.
• Complete an optional Secret Mission that exposes a disguised authorization bug. Prove that high security risk forces human review even when the primary domain remains frontend.

Ready to quiz yourself?