Cjbuilds · bottlerex · Aug 30, 2026 · Sep 7, 2026
@@ -1,5 +1,12 @@
 # Changelog
 
+## 0.9.4 — Unreleased
+
+- Accept the current Claude CLI's documented runtime metadata string fields while
+  retaining numeric validation for usage counters, and permit the observed exact
+  Haiku helper alongside the sealed Claude Opus 5 primary. Unknown metadata
+  fields and helper identities remain fail-closed.
+
 ## 0.9.3 — Unreleased
 
 - Raise the bounded Advisor approval loop from five to eight reviews while

@@ -1,6 +1,6 @@
 {
   "name": "codex-orchestration",
-  "version": "0.9.3",
+  "version": "0.9.4",
   "description": "Give Codex and audited external models safe, provider-pinned roles.",
   "author": {
     "name": "CJ Zafir",

@@ -241,7 +241,7 @@ suggestions, and requires JSON runtime metadata to contain an allowed primary.
 For the Fable route, the reviewed primary identities are `claude-fable-5` and
 its resolved runtime identity `claude-opus-4-8`; only the exact internal helper
 `claude-haiku-4-5-20251001` is additionally permitted. The separate Opus route
-still requires `claude-opus-5`, with no helper. Advisor decisions use
+requires `claude-opus-5` and permits that same exact observed helper. Advisor decisions use
 `--json-schema` and are locally revalidated; raw prose is not approval. Any
 missing primary or unknown additional model fails closed. Identity rotation
 therefore requires a reviewed plugin update rather than a wildcard. Setup and

@@ -45,12 +45,14 @@
 )
 ALLOWED_RUNTIME_MODELS_BY_PRIMARY = {
     FABLE_MODEL: ALLOWED_RUNTIME_MODELS,
-    # No Opus helper identity has been independently verified. Fail closed if
-    # Claude Code reports anything beyond the sealed primary.
-    OPUS_MODEL: frozenset({OPUS_MODEL}),
+    # Claude Code 2.1.251 reports this exact helper alongside the sealed Opus
+    # primary. Keep unknown helpers fail-closed.
+    OPUS_MODEL: frozenset({OPUS_MODEL, FABLE_HELPER_MODEL}),
 }
+MODEL_USAGE_STRING_FIELDS = frozenset({"canonicalModel", "provider", "costBasis"})
 CLAUDE_TIMEOUT_SECONDS = 600
 AUTH_TIMEOUT_SECONDS = 20
+SUBPROCESS_DIAGNOSTIC_MAX_CHARS = 4096
 # Applies to the combined user-controlled text sent by one model operation.
 MAX_INPUT_CHARS = 200_000
 PLAN_REVIEW_SCHEMA = {
@@ -157,6 +159,56 @@ class AdvisorError(RuntimeError):
     """Fail-closed error for any bundled Claude bridge operation."""
 
 
+def _classify_subprocess_failure(*outputs: str | None) -> str:
+    """Return a non-sensitive, bounded category for a failed Claude subprocess.
+
+    Raw subprocess output can contain prompt fragments, account information, or
+    provider diagnostics. The MCP response must never return it. Restricting
+    classification to a small, fixed vocabulary preserves useful remediation
+    without turning error responses into an output-exfiltration channel.
+    """
+
+    diagnostic = "\n".join(
+        value[:SUBPROCESS_DIAGNOSTIC_MAX_CHARS]
+        for value in outputs
+        if isinstance(value, str)
+    ).lower()
+    if any(
+        token in diagnostic
+        for token in (
+            "not logged",
+            "authentication",
+            "auth login",
+            "oauth",
+            "unauthorized",
+            "forbidden",
+        )
+    ):
+        return "auth"
+    if any(
+        token in diagnostic
+        for token in ("rate limit", "usage limit", "quota", "limit reached")
+    ):
+        return "usage_limit"
+    if any(
+        token in diagnostic
+        for token in ("network", "connection", "connect", "dns", "enotfound", "econn")
+    ):
+        return "transport"
+    if any(
+        token in diagnostic
+        for token in (
+            "json schema",
+            "invalid schema",
+            "unknown option",
+            "unrecognized option",
+            "unsupported option",
+        )
+    ):
+        return "cli_contract"
+    return "unknown"
+
+
 def codex_home() -> Path:
     value = os.environ.get("CODEX_HOME")
     return Path(value).expanduser() if value else Path.home() / ".codex"
@@ -240,8 +292,10 @@ def _run_json(command: list[str], *, timeout: int) -> dict[str, Any]:
     except OSError as exc:
         raise AdvisorError("Could not run Claude Code authentication check.") from exc
     if result.returncode != 0:
+        category = _classify_subprocess_failure(result.stderr, result.stdout)
         raise AdvisorError(
-            f"Claude Code authentication check exited with {result.returncode}; output withheld."
+            "Claude Code authentication check failed "
+            f"(exit={result.returncode}; category={category}; output withheld)."
         )
     try:
         payload = json.loads(result.stdout)
@@ -384,10 +438,15 @@ def _validate_runtime_models(
                 and math.isfinite(value)
                 and value >= 0
             )
+            is_known_metadata_string = (
+                field in MODEL_USAGE_STRING_FIELDS
+                and isinstance(value, str)
+                and bool(value.strip())
+            )
             if (
                 not isinstance(field, str)
                 or not field.strip()
-                or not is_nonnegative_finite_number
+                or not (is_nonnegative_finite_number or is_known_metadata_string)
             ):
                 raise AdvisorError(
                     "Runtime metadata has a malformed modelUsage value."
@@ -567,8 +626,10 @@ def _invoke_fable(
     except OSError as exc:
         raise AdvisorError(f"Could not start {display_name} {operation}.") from exc
     if result.returncode != 0:
+        category = _classify_subprocess_failure(result.stderr, result.stdout)
         raise AdvisorError(
-            f"{display_name} {operation} exited with {result.returncode}; output withheld."
+            f"{display_name} {operation} failed "
+            f"(exit={result.returncode}; category={category}; output withheld)."
         )
     try:
         decoded = json.loads(result.stdout)

@@ -660,6 +660,14 @@ def test_runtime_model_usage_values_fail_closed(self) -> None:
             {FABLE.FABLE_MODEL: {"outputTokens": 0}},
             {FABLE.FABLE_MODEL: {"outputTokens": 10**309}},
             {FABLE.FABLE_MODEL: {"costUSD": 0.25, "outputTokens": 12}},
+            {
+                FABLE.FABLE_MODEL: {
+                    "outputTokens": 12,
+                    "canonicalModel": "claude-fable-5",
+                    "provider": "firstParty",
+                    "costBasis": "list",
+                }
+            },
             {
                 FABLE.FABLE_MODEL: {"outputTokens": 12},
                 FABLE.FABLE_HELPER_MODEL: {"outputTokens": 1},
@@ -674,6 +682,23 @@ def test_runtime_model_usage_values_fail_closed(self) -> None:
                 )
                 self.assertEqual(result["decision"], "PLAN_APPROVED")
 
+        opus_usage = {
+            FABLE.OPUS_MODEL: {"outputTokens": 12},
+            FABLE.FABLE_HELPER_MODEL: {"outputTokens": 1},
+        }
+        self.assertEqual(
+            FABLE._validate_runtime_models(opus_usage, FABLE.OPUS_MODEL),
+            sorted(opus_usage),
+        )
+        with self.assertRaisesRegex(FABLE.AdvisorError, "outside the allowed"):
+            FABLE._validate_runtime_models(
+                {
+                    FABLE.OPUS_MODEL: {"outputTokens": 12},
+                    "unknown-helper": {"outputTokens": 1},
+                },
+                FABLE.OPUS_MODEL,
+            )
+
     def test_each_operation_pins_its_authorized_seat_effort(self) -> None:
         self.write_state(planner=self.route("low"))
         created, create_calls = self.invoke_with_results(
@@ -744,7 +769,7 @@ def test_route_validation_is_constrained_and_backward_compatible(self) -> None:
         self.write_state(schema=5, advisor=self.route())
         self.assertEqual(FABLE.load_fable_route(self.home)["model"], FABLE.FABLE_MODEL)
 
-    def test_opus_route_pins_primary_and_rejects_every_unverified_helper(self) -> None:
+    def test_opus_route_pins_primary_allows_observed_helper_and_rejects_unknown(self) -> None:
         self.write_state(schema=5, advisor=self.opus_route("xhigh"))
         result, calls = self.invoke_with_results(
             FABLE.review_plan,
@@ -771,7 +796,7 @@ def test_opus_route_pins_primary_and_rejects_every_unverified_helper(self) -> No
                 model_response="PLAN_APPROVED\nNo material gap.",
                 model_usage={
                     FABLE.OPUS_MODEL: {"outputTokens": 12},
-                    FABLE.FABLE_HELPER_MODEL: {"outputTokens": 1},
+                    "unknown-helper": {"outputTokens": 1},
                 },
             )
         with self.assertRaisesRegex(
@@ -1200,6 +1225,7 @@ def test_subprocess_failures_and_timeouts_do_not_leak_prompt_output(self) -> Non
             with self.assertRaises(FABLE.AdvisorError) as failure:
                 FABLE.review_plan(secret)
         self.assertIn("17", str(failure.exception))
+        self.assertIn("category=unknown", str(failure.exception))
         self.assertNotIn(secret, str(failure.exception))
 
         timeout = subprocess.TimeoutExpired(["claude"], 600, output=secret, stderr=secret)
@@ -1217,6 +1243,21 @@ def test_subprocess_failures_and_timeouts_do_not_leak_prompt_output(self) -> Non
         self.assertIn("timed out", str(timed_out.exception))
         self.assertNotIn(secret, str(timed_out.exception))
 
+    def test_subprocess_failure_categories_are_bounded_and_non_sensitive(self) -> None:
+        secret = "TOP-SECRET-SUBPROCESS-OUTPUT"
+        cases = (
+            ("OAuth token expired", "auth"),
+            ("usage limit reached", "usage_limit"),
+            ("network connection reset", "transport"),
+            ("unknown option --json-schema", "cli_contract"),
+            (f"unexpected failure {secret}", "unknown"),
+        )
+        for output, expected in cases:
+            with self.subTest(output=output):
+                category = FABLE._classify_subprocess_failure(output)
+                self.assertEqual(category, expected)
+                self.assertNotIn(secret, category)
+
     def test_input_bound_is_checked_before_subprocess(self) -> None:
         with mock.patch.object(FABLE.subprocess, "run") as run:
             with self.assertRaisesRegex(FABLE.AdvisorError, "character combined limit"):