feat: add minimal task phase routing

README.md

@@ -22,6 +22,8 @@ readiness.
 - one injected helper transport, one attempt, no retry or fallback;
 - raw artifact hashes and locators outside the compact result;
 - one compact result on stdout and a named operator indicator on stderr;
+- one small deterministic route selector for `PLAN`, `BUILD`, `TEST`,
+  `DEPLOY`, and `IDEAS` pipeline phases;
 - an `opsle.value-receipt.v1` sidecar with bounded observational claims.
 
 The bundled CLI executes deterministic gears only. Cognitive execution requires
@@ -32,7 +34,9 @@ request when no transport is present; it never silently falls back.
 
 Gearbox is not a durable supervisor, queue, scheduler, discovery engine,
 retry/recovery controller, persistent agent hierarchy, exact-session resume
-mechanism, general autonomous-task platform, or provider router.
+mechanism, general autonomous-task platform, dynamic discovery engine, or
+probabilistic provider router. Its small route selector uses only explicit
+caller-supplied availability and a fixed rule.
 
 Context Firewall is an external integration. Gearbox decides where bounded work
 executes; Context Firewall decides what decision-relevant evidence returns. The
@@ -95,6 +99,18 @@ Canonical result JSON is written to stdout. The concise `[Gearbox] ...`
 indicator is written to stderr. Raw stdout/stderr and helper evidence stay under
 the mode-0700 state root.
 
+For a small task pipeline, route one phase by sending a request on stdin:
+
+```text
+PYTHONPATH=src python3 -m opsle_gearbox.route <<'JSON'
+{"schema":"opsle.gearbox.route-request.v1","phase":"BUILD","choice":"auto","available":{"codex":true,"claude":true,"deterministic":false}}
+JSON
+```
+
+The selector returns one route: `codex`, `claude`, `deterministic`, or `noop`.
+`TEST` and `DEPLOY` are deterministic when a command is configured and otherwise
+become explicit no-ops. It performs no execution, retries, discovery, or state.
+
 ## Verification
 
 ```text

pyproject.toml

@@ -17,6 +17,7 @@ classifiers = [
 
 [project.scripts]
 gearbox = "opsle_gearbox.cli:main"
+gearbox-route = "opsle_gearbox.route:main"
 
 [tool.setuptools.packages.find]
 where = ["src"]

src/opsle_gearbox/route.py

@@ -0,0 +1,78 @@
+"""Small deterministic route selector for task-pipeline phases."""
+
+from __future__ import annotations
+
+import json
+import sys
+from typing import Any
+
+REQUEST_SCHEMA = "opsle.gearbox.route-request.v1"
+RESULT_SCHEMA = "opsle.gearbox.route-result.v1"
+PHASES = {"PLAN", "BUILD", "TEST", "DEPLOY", "IDEAS"}
+CHOICES = {"auto", "codex", "claude"}
+
+
+class RouteError(ValueError):
+    """The caller supplied an invalid or impossible routing request."""
+
+
+def select_route(value: object) -> dict[str, Any]:
+    if not isinstance(value, dict) or set(value) != {"schema", "phase", "choice", "available"}:
+        raise RouteError("route request must contain exactly schema, phase, choice, and available")
+    if value["schema"] != REQUEST_SCHEMA:
+        raise RouteError("unsupported route request schema")
+
+    phase = value["phase"]
+    choice = value["choice"]
+    available = value["available"]
+    if not isinstance(phase, str) or phase not in PHASES:
+        raise RouteError("unknown phase")
+    if not isinstance(choice, str) or choice not in CHOICES:
+        raise RouteError("unknown provider choice")
+    if not isinstance(available, dict) or set(available) != {"codex", "claude", "deterministic"}:
+        raise RouteError("available must contain exactly codex, claude, and deterministic")
+    if any(not isinstance(item, bool) for item in available.values()):
+        raise RouteError("availability values must be booleans")
+
+    if phase in {"TEST", "DEPLOY"}:
+        route = "deterministic" if available["deterministic"] else "noop"
+        reason = f"{phase.lower()} command configured" if available["deterministic"] else f"no {phase.lower()} command configured"
+    elif choice != "auto":
+        if not available[choice]:
+            raise RouteError(f"{choice} is not available")
+        route = choice
+        reason = "chosen by user"
+    elif available["codex"]:
+        route = "codex"
+        reason = "auto selected first available provider"
+    elif available["claude"]:
+        route = "claude"
+        reason = "auto selected first available provider"
+    else:
+        raise RouteError("no provider is available")
+
+    return {
+        "schema": RESULT_SCHEMA,
+        "phase": phase,
+        "route": route,
+        "reason": reason,
+    }
+
+
+def main() -> int:
+    try:
+        request = json.load(sys.stdin)
+        result = select_route(request)
+        sys.stdout.write(json.dumps(result, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + "\n")
+        return 0
+    except (RouteError, json.JSONDecodeError) as error:
+        sys.stdout.write(json.dumps({
+            "schema": "opsle.gearbox.route-error.v1",
+            "status": "failed",
+            "error": str(error),
+        }, separators=(",", ":"), sort_keys=True) + "\n")
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_route.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import unittest
+
+from opsle_gearbox.route import REQUEST_SCHEMA, RouteError, select_route
+
+
+def request(phase="PLAN", choice="auto", *, codex=True, claude=True, deterministic=False):
+    return {
+        "schema": REQUEST_SCHEMA,
+        "phase": phase,
+        "choice": choice,
+        "available": {"codex": codex, "claude": claude, "deterministic": deterministic},
+    }
+
+
+class RouteTests(unittest.TestCase):
+    def test_explicit_provider_wins(self):
+        self.assertEqual(select_route(request(choice="claude"))["route"], "claude")
+
+    def test_auto_is_deterministic(self):
+        self.assertEqual(select_route(request())["route"], "codex")
+        self.assertEqual(select_route(request(codex=False))["route"], "claude")
+
+    def test_test_and_deploy_are_deterministic_or_noop(self):
+        self.assertEqual(select_route(request("TEST", deterministic=True))["route"], "deterministic")
+        self.assertEqual(select_route(request("DEPLOY"))["route"], "noop")
+
+    def test_unavailable_explicit_provider_fails(self):
+        with self.assertRaises(RouteError):
+            select_route(request(choice="claude", claude=False))
+
+    def test_unhashable_phase_and_choice_fail_as_route_errors(self):
+        invalid_phase = request()
+        invalid_phase["phase"] = []
+        invalid_choice = request()
+        invalid_choice["choice"] = {}
+        with self.assertRaisesRegex(RouteError, "unknown phase"):
+            select_route(invalid_phase)
+        with self.assertRaisesRegex(RouteError, "unknown provider choice"):
+            select_route(invalid_choice)
+
+    def test_cli_reads_one_json_request_from_stdin(self):
+        environment = {**os.environ, "PYTHONPATH": "src"}
+        result = subprocess.run(
+            [sys.executable, "-m", "opsle_gearbox.route"],
+            input=json.dumps(request("BUILD")), text=True, capture_output=True, env=environment, check=True,
+        )
+        self.assertEqual(json.loads(result.stdout)["route"], "codex")
+
+
+if __name__ == "__main__":
+    unittest.main()