diff --git a/docs/api-reference.md b/docs/api-reference.md index 6a3fb27b..fe336a31 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -578,8 +578,9 @@ cancellation-completion signal. **`intent_verdict`** -- delivered asynchronously when the LLM judge completes its evaluation of a pending tool call. Only sent when intent validation is -enabled (`--judge` or `[judge] enabled = true`). The `call_id` correlates with -the item in the preceding `approve_request` event. +enabled (`judge.enabled` through Admin → Judge or the admin settings API). The +interactive CLI instead uses `--judge` or `[judge] enabled = true`. The +`call_id` correlates with the item in the preceding `approve_request` event. ```json { diff --git a/docs/architecture.md b/docs/architecture.md index 36be0ea4..2a33817f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -2079,12 +2079,15 @@ implemented in `turnstone/core/judge.py`: (approve). First match wins. The heuristic verdict is attached to the `approve_request` SSE event immediately. -2. **LLM judge tier** (asynchronous, daemon thread) -- A multi-turn evaluation - where the judge LLM receives conversation context and tool call details, - optionally uses `read_file`/`list_directory` to gather evidence (with - security-hardened path blocking), and produces a structured JSON verdict. - If the LLM verdict has higher confidence than the heuristic, it replaces - it via an `intent_verdict` SSE event. +2. **LLM judge tier** (asynchronous daemon coordinator) -- A bounded worker + set evaluates independent calls from the batch. Each evaluation receives + conversation context and tool-call details, may use `read_file` / + `list_directory` to gather evidence (with security-hardened path blocking), + and produces a structured JSON verdict. `judge.parallel_evaluations` + controls the per-batch width from 1 through 16; the judge alias's model + admission gate remains the process-wide ceiling. If an LLM verdict has + higher confidence than the heuristic, it replaces it via an + `intent_verdict` SSE event. The main judge is session-scoped (`IntentJudge`) and lazy-initialized on first approval; each evaluation carries its own cancellation/generation identity. @@ -2093,9 +2096,11 @@ Task-agent tool calls use the same intent pipeline in independent judge work. Each human-gated batch is joined to its own `ApprovalCycle`, and a late verdict must match that cycle's call ID and judge identity before it can reach Smart Approvals. Superseded verdicts remain durable audit facts but are -withheld from live decision caches. Configuration comes from `[judge]` or CLI -flags; self-consistency, cross-model, and cross-provider bindings all use the -same `ModelLane`/backend-auth seam. Verdicts persist in `intent_verdicts` with -the exact user or automatic decision, enabling calibration. +withheld from live decision caches. Server and console behavior comes from the +database-backed `judge.*` settings; the interactive CLI reads its flags and +`config.toml` `[judge]` values. Self-consistency, cross-model, and +cross-provider bindings all use the same `ModelLane` / backend-auth seam. +Verdicts persist in `intent_verdicts` with the exact user or automatic +decision, enabling calibration. The console exposes `GET /v1/api/admin/verdicts` for audit queries (requires `admin.judge` permission). diff --git a/docs/diagrams/22-judge-architecture.puml b/docs/diagrams/22-judge-architecture.puml index 9d9639e2..23637089 100644 --- a/docs/diagrams/22-judge-architecture.puml +++ b/docs/diagrams/22-judge-architecture.puml @@ -28,10 +28,13 @@ note over Judge, Model are freshness watermarks: an effective lane change replaces the judge for the next batch, while in-flight work keeps the lane it started with. Dynamic backend auth is resolved for this batch's initiating principal. + parallel_evaluations (1-16) sets per-batch worker width; the model alias's + admission gate remains the process-wide generation ceiling. end note -par LLM judge daemon - loop bounded turns / deadline +par LLM judge daemon coordinator + Judge -> Judge : start min(batch size, parallel_evaluations,\npositive alias capacity) workers + loop each worker claims one independent call Judge -> Model : model_turn(judge lane, canonical Turns,\nread-only evidence tools, cancel_ref) Model --> Judge : ModelTurnResult alt evidence tool requested @@ -39,9 +42,9 @@ par LLM judge daemon else verdict text Judge -> Judge : parse + arbitrate against heuristic end + Judge --> UI : on_intent_verdict(verdict, judge generation) + UI -> Storage : persist LLM verdict / audit update end - Judge --> UI : on_intent_verdict(verdict, judge generation) - UI -> Storage : persist LLM verdict / audit update else approval path continues Session -> UI : approve_tools(items) with one\nSmart Approval config snapshot end diff --git a/docs/diagrams/png/22-judge-architecture.png b/docs/diagrams/png/22-judge-architecture.png index a4b9efc4..5f56805f 100644 --- a/docs/diagrams/png/22-judge-architecture.png +++ b/docs/diagrams/png/22-judge-architecture.png @@ -1,3 +1,3 @@ version https://git-lfs.github.com/spec/v1 -oid sha256:dde4f417956534a70b0e37b24cfe9de383897780669c9da81833c8084d572dfe -size 269928 +oid sha256:636a6b2fc1075e4863421e68b99efe7f6f6f62cedbcff36ef0934c055f39fd46 +size 281161 diff --git a/docs/judge.md b/docs/judge.md index 5f07d300..c87b9304 100644 --- a/docs/judge.md +++ b/docs/judge.md @@ -30,23 +30,30 @@ persisted to the `intent_verdicts` table for audit and future calibration. ## Configuration -### config.toml +### Server and console -```toml -[judge] -enabled = true -model = "" # empty = same as session model -provider = "" # empty = same as session provider -base_url = "" -api_key = "" -smart_approvals = false # auto-approve high-confidence "approve" LLM verdicts (opt-in) -confidence_threshold = 0.95 # Smart Approvals auto-approve bar (LLM recommendation=approve) -max_context_ratio = 0.5 # max % of judge context window for history -timeout = 120.0 # per judge turn; also caps the Smart Approvals wait -read_only_tools = true # judge can use read_file/list_directory -cancel_on_approval = false # stop judging remaining tool calls once user decides +Server and console workstreams read database-backed `judge.*` settings from +the settings registry. Edit them at **Admin → Judge** or through the admin +settings API; changes take effect for the next judge batch without a restart. +The principal settings are: + +```text +judge.enabled = true +judge.model = "" # empty = same alias as the session +judge.smart_approvals = false # opt-in automatic approval +judge.confidence_threshold = 0.95 # Smart Approvals confidence bar +judge.max_context_ratio = 0.5 # fraction of judge context used for history +judge.timeout = 120.0 # per judge turn and Smart Approvals wait +judge.parallel_evaluations = 1 # concurrent calls within one batch, 1-16 +judge.read_only_tools = true # permit read_file/list_directory evidence +judge.cancel_on_approval = false # stop unfinished calls when the gate resolves ``` +`parallel_evaluations = 1` preserves serial evaluation. Raising it reduces the +latency of wide tool-call batches. The selected judge model alias's +`max_concurrency` remains the process-wide generation ceiling, so it can reduce +the actual overlap across judge batches and other roles using that alias. + ### Smart Approvals With `smart_approvals = true` (off by default), a pending batch is approved @@ -78,21 +85,22 @@ Requires the judge to be enabled. Auto-approved calls are tagged `smart_approval` in the dashboard and audit trail. Smart Approvals applies to the web and coordinator surfaces, not the interactive CLI. -All fields are optional. The judge is enabled by default; use `enabled = false` -(or `--no-judge` on the command line) to disable it. +The judge is enabled by default. Disable `judge.enabled` in the admin Judge +settings, or use `--no-judge` in the interactive CLI. ### CLI flags ``` --judge / --no-judge Enable/disable (default: enabled) ---judge-model MODEL Model for judge ---judge-provider PROVIDER Provider for judge +--judge-model ALIAS Registered model alias for judge --judge-timeout SECONDS LLM judge timeout (default: 120) +--judge-parallel-evaluations N Concurrent evaluations per batch, 1-16 (default: 1) --judge-confidence FLOAT Confidence threshold, 0-1 (default: 0.95) ``` -(Smart Approvals is configured via `[judge] smart_approvals` / the admin Judge -settings, not a CLI flag — the interactive CLI prompts for approval directly.) +The same five values can be placed in the CLI's `config.toml` `[judge]` +section. Smart Approvals is configured through the server/console admin Judge +settings, not a CLI flag—the interactive CLI prompts for approval directly. CLI flags override `config.toml` values. @@ -103,20 +111,19 @@ CLI flags override `config.toml` values. - **Default (self-consistency)**: When `model` is empty, the session model evaluates its own tool calls. Research shows self-consistency achieves comparable accuracy to multi-agent debate at a fraction of the cost. -- **Cross-model**: Use a different model for the judge (e.g. local model for - the session, commercial model for the judge). Set `model` and `provider` - in the `[judge]` config section, or use `--judge-model` / `--judge-provider` - CLI flags. -- **Cross-provider**: When both `model` and `provider` are set, the judge - creates its own LLM client. You can optionally specify `base_url` and - `api_key` for non-default endpoints. -- **Google models**: The judge supports `google` as a provider. Note that - read-only tools are disabled for Google models (the Gemini API requires - `thought_signature` in tool call round-trips which the judge's normalized - format does not preserve). +- **Cross-model**: Register the desired model in the Models tab, then set + `judge.model` to that alias (or pass `--judge-model ALIAS` to the CLI). +- **Cross-provider**: A model alias carries its provider, endpoint, and + credential configuration together, so a judge alias may use a different + provider from the session without separate judge connection settings. +- **Google models**: The judge supports `google` aliases, including read-only + evidence tools. Provider-native reasoning state such as Gemini + `thought_signature` stays attached to the pinned model lane across evidence + turns. -The judge creates a fresh HTTP client for each evaluation run and closes it -when done, avoiding stale connection issues across runs. +The judge creates one fresh HTTP client per active batch worker and closes each +when that worker finishes, avoiding cross-thread client sharing and stale +connections across runs. If the LLM judge fails or returns no verdict, a fallback verdict with tier `llm_fallback` is delivered via the callback, ensuring the UI always receives @@ -248,16 +255,23 @@ calls for approval, it calls `_evaluate_intent()` which: 4. Attaches each heuristic verdict to its item as `_heuristic_verdict` 5. The daemon thread runs the LLM judge and delivers results via `ui.on_intent_verdict()` -The daemon evaluates items sequentially, so a large parallel batch can outlive -its approval gate. With `cancel_on_approval = false` (the default) the daemon -runs every item to completion: verdicts that land after the operator decided -still stream to the UI and persist, stamped with the decision. A newer main-loop -batch, session close, or explicit Stop retires the old generation; unfinished -items degrade to `llm_fallback` verdicts. A judge/model binding edit prevents +The daemon coordinates up to `parallel_evaluations` independent workers for +one batch. Completed verdicts stream to the UI as workers finish, and every call +still receives exactly one LLM or `llm_fallback` verdict. The default of 1 keeps +the historical serial behavior; a higher value collapses a wide batch toward +`ceil(batch size / workers)` judge-call intervals. A smaller positive model +alias capacity also bounds the worker count, avoiding surplus threads queued at +the same admission gate. + +With `cancel_on_approval = false` (the default) the daemon runs every item to +completion: verdicts that land after the operator decided still stream to the +UI and persist, stamped with the decision. A newer main-loop batch, session +close, or explicit Stop retires the old generation; unfinished items degrade +to `llm_fallback` verdicts. A judge/model binding or parallelism edit prevents reuse on the next batch, while already-started calls stay pinned to the binding -they began with. With `cancel_on_approval = true`, an ordinary gate decision -additionally aborts the remainder immediately, trading verdict completeness -for inference savings — recommended when the judge shares a single local +and worker count they began with. With `cancel_on_approval = true`, an ordinary +gate decision additionally aborts unfinished work, trading verdict completeness +for inference savings—recommended when the judge shares a single local inference backend with the session model. Explicit Stop always cancels every live judge generation, regardless of this preference. @@ -441,13 +455,12 @@ Redaction types: `api_key`, `private_key`, `password`, `secret`. ### Configuration -```toml -[judge] -output_guard = true # enable output evaluation (default) -redact_secrets = true # auto-redact detected credentials (default) +```text +judge.output_guard = true # enable output evaluation (default) +judge.redact_secrets = true # auto-redact detected credentials (default) ``` -Configurable at runtime via the admin Settings tab. +Configure both at runtime through the admin Judge settings. ### Merge semantics (heuristic + LLM judge) diff --git a/docs/settings.md b/docs/settings.md index 5efd49f8..6eac4e25 100644 --- a/docs/settings.md +++ b/docs/settings.md @@ -71,6 +71,21 @@ attempt. The cap is local to each process, not cluster-wide; account for the number of nodes targeting the same inference server. Direct STT/TTS protocol calls and Cohere/Jina reranking do not currently consume this generation cap. +### Judge batch parallelism + +`judge.parallel_evaluations` controls how many independent tool calls from one +approval batch the intent judge evaluates concurrently. It is an integer from +1 through 16 and defaults to 1, preserving serial evaluation until an operator +opts into wider fan-out. Changes are hot-read at the next batch; work already +in flight keeps its captured worker count. + +This is a per-batch fan-out setting, not another backend capacity limit. The +judge model alias's `max_concurrency` gate still caps total generations across +all judge batches and every other role using that alias. Actual overlap is +therefore bounded by the batch size, `judge.parallel_evaluations`, and available +alias admission slots. A smaller positive alias cap also narrows the batch's +worker pool so excess judge threads do not queue ahead of later alias traffic. + ### Model backend authentication Model definitions support four backend credential modes: @@ -243,7 +258,7 @@ initialization: | `mcp` | config_path, registry_url | | `ratelimit` | enabled, requests_per_second, burst, trusted_proxies | | `health` | backend_probe_interval, backend_probe_timeout, circuit_breaker_threshold, circuit_breaker_cooldown | -| `judge` | enabled, model, provider, base_url, api_key, smart_approvals, confidence_threshold, max_context_ratio, timeout, read_only_tools, output_guard, redact_secrets, cancel_on_approval | +| `judge` | enabled, model, smart_approvals, confidence_threshold, max_context_ratio, timeout, parallel_evaluations, read_only_tools, output_guard, output_guard_budget_seconds, output_guard_llm, output_guard_model, output_guard_llm_timeout, redact_secrets, cancel_on_approval | | `interface` | close_tab_action, theme | | `skills` | discovery_url | | `memory` | relevance_k, fetch_limit, max_content, nudge_cooldown, nudges | @@ -420,13 +435,11 @@ Reset a setting to its registry default by removing it from storage. ## Secret Settings -Settings with `is_secret=True` (currently only `judge.api_key`) are blocked -from the write API with a `403` response. This prevents accidental exposure -through the admin UI or audit logs. Secret settings must be configured via -`config.toml` or environment variables. - -The list endpoint masks secret values: stored secrets appear as `"***"` -rather than their actual value. +The registry currently defines no production secret system setting. The generic +machinery nevertheless treats any future `is_secret=True` entry as write-only: +list and write responses return `"***"`, and submitting that sentinel preserves +the stored value. Model API keys are fields on model definitions—not +`judge.*` system settings—and use the Models tab's separate write-only flow. --- diff --git a/tests/test_app_js.py b/tests/test_app_js.py index d5212c2d..1cdfb00b 100644 --- a/tests/test_app_js.py +++ b/tests/test_app_js.py @@ -708,6 +708,22 @@ def test_audio_roles_gated_to_openai_sdk_providers() -> None: assert '_providerCarriesAudio((md && md.provider) || "openai")' in body +def test_judge_integer_settings_render_as_bounded_number_inputs() -> None: + """The Judge tab has a custom schema renderer separate from Settings. + + Integer settings must not fall through to its text-input branch: doing so + drops the registry's step/min/max affordances for parallel_evaluations. + """ + governance = _CONSOLE_GOVERNANCE_JS.read_text(encoding="utf-8") + start = governance.index("function renderJudgeSettings()") + end = governance.index("\nfunction saveJudgeSetting(", start) + body = governance[start:end] + assert 's.type === "float" || s.type === "int"' in body + assert '(s.type === "int" ? "1" : "0.01")' in body + assert "s.min_value" in body + assert "s.max_value" in body + + # Tile keys that are deliberately NOT ``ModelCapabilities`` fields. # ``supports_rerank`` is a registry-level flag read off the model row. _NON_DATACLASS_TILES = {"supports_rerank"} diff --git a/tests/test_config.py b/tests/test_config.py index ec32a84e..4bc6f8bd 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -2,7 +2,10 @@ import argparse +import pytest + import turnstone.core.config as config_mod +from turnstone.cli import _judge_parallel_evaluations_arg apply_config = config_mod.apply_config load_config = config_mod.load_config @@ -324,6 +327,7 @@ def test_apply_config_judge_section(tmp_path): 'model = "gpt-5"\n' "confidence_threshold = 0.85\n" "timeout = 30.0\n" + "parallel_evaluations = 8\n" "read_only_tools = false\n" ) set_config_path(str(cfg)) @@ -333,6 +337,12 @@ def test_apply_config_judge_section(tmp_path): parser.add_argument("--judge-model", dest="judge_model", default="") parser.add_argument("--judge-confidence", dest="judge_confidence", type=float, default=0.7) parser.add_argument("--judge-timeout", dest="judge_timeout", type=float, default=60.0) + parser.add_argument( + "--judge-parallel-evaluations", + dest="judge_parallel_evaluations", + type=int, + default=1, + ) parser.add_argument("--judge-read-only-tools", dest="judge_read_only_tools", default=True) apply_config(parser, ["judge"]) @@ -342,6 +352,7 @@ def test_apply_config_judge_section(tmp_path): assert args.judge_model == "gpt-5" assert args.judge_confidence == 0.85 assert args.judge_timeout == 30.0 + assert args.judge_parallel_evaluations == 8 assert args.judge_read_only_tools is False @@ -349,19 +360,36 @@ def test_apply_config_judge_cli_overrides(tmp_path): """CLI flags override config.toml [judge] values.""" _reset_cache() cfg = tmp_path / "config.toml" - cfg.write_text("[judge]\nenabled = true\nconfidence_threshold = 0.85\n") + cfg.write_text( + "[judge]\nenabled = true\nconfidence_threshold = 0.85\nparallel_evaluations = 8\n" + ) set_config_path(str(cfg)) parser = argparse.ArgumentParser() parser.add_argument("--judge", dest="judge_enabled", action="store_true", default=False) parser.add_argument("--no-judge", dest="judge_enabled", action="store_false") parser.add_argument("--judge-confidence", dest="judge_confidence", type=float, default=0.7) + parser.add_argument( + "--judge-parallel-evaluations", + dest="judge_parallel_evaluations", + type=int, + default=1, + ) apply_config(parser, ["judge"]) - args = parser.parse_args(["--no-judge"]) + args = parser.parse_args(["--no-judge", "--judge-parallel-evaluations", "3"]) assert args.judge_enabled is False # CLI wins assert args.judge_confidence == 0.85 # config wins (no CLI override) + assert args.judge_parallel_evaluations == 3 + + +def test_judge_parallel_evaluations_cli_type_is_strict_and_bounded(): + assert _judge_parallel_evaluations_arg("1") == 1 + assert _judge_parallel_evaluations_arg("16") == 16 + for invalid in ("0", "17", "01", "+1", "1.0"): + with pytest.raises(argparse.ArgumentTypeError): + _judge_parallel_evaluations_arg(invalid) def test_set_config_path_overrides_default(tmp_path): diff --git a/tests/test_console_session_factory.py b/tests/test_console_session_factory.py index 0b520c78..7aed1919 100644 --- a/tests/test_console_session_factory.py +++ b/tests/test_console_session_factory.py @@ -340,6 +340,14 @@ def test_mcp_no_getter_is_backward_compatible() -> None: assert got["mcp_client"] is None +def test_judge_parallel_evaluations_reaches_coordinator_session() -> None: + got = _capture_chatsession_kwargs( + settings={"judge.parallel_evaluations": 7}, + getter_passed=False, + ) + assert got["judge_config"].parallel_evaluations == 7 + + def test_mcp_getter_resolved_per_construction() -> None: """The getter is consulted at EVERY construction — a manager (re)constructed by the console ensure-helper after factory build must diff --git a/tests/test_judge.py b/tests/test_judge.py index bb53c4a5..ba362e8d 100644 --- a/tests/test_judge.py +++ b/tests/test_judge.py @@ -7,9 +7,11 @@ import threading import time from dataclasses import replace from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import Any from unittest.mock import MagicMock +import pytest + from tests._session_helpers import as_stream from tests._session_helpers import mock_completion_result as _mock_result from turnstone.core.admission import ModelAdmission @@ -20,9 +22,6 @@ from turnstone.core.model_turn import ModelLane, ResolvedModelBinding from turnstone.core.providers._protocol import IncompleteStreamError, ModelCapabilities from turnstone.core.trajectory import Role -if TYPE_CHECKING: - import pytest - # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- @@ -113,6 +112,7 @@ def _make_judge( confidence_threshold: float = 0.7, read_only_tools: bool = True, timeout: float = 60.0, + parallel_evaluations: int = 1, ) -> IntentJudge: """Create a judge with a mock provider.""" if provider is None: @@ -123,6 +123,7 @@ def _make_judge( confidence_threshold=confidence_threshold, read_only_tools=read_only_tools, timeout=timeout, + parallel_evaluations=parallel_evaluations, ) client = MagicMock() client.base_url = "https://api.openai.com/v1" @@ -164,6 +165,54 @@ def _good_verdict_json(**overrides: Any) -> str: return json.dumps(verdict) +def _llm_verdict_for(item: dict[str, Any]) -> IntentVerdict: + """Build one deterministic successful verdict for scheduler tests.""" + call_id = str(item.get("call_id", "")) + func_name = str(item.get("func_name", "")) + return IntentVerdict( + verdict_id=f"v-{call_id}", + call_id=call_id, + func_name=func_name, + func_args=json.dumps(item.get("func_args", {}), sort_keys=True), + intent_summary=f"Evaluate {call_id}", + risk_level="low", + confidence=0.99, + recommendation="approve", + reasoning="Deterministic scheduler-test verdict.", + evidence=[], + tier="llm", + judge_model="test-model", + latency_ms=1, + ) + + +class _TrackingClient: + """Tiny per-worker client whose close lifecycle is directly assertable.""" + + def __init__(self) -> None: + self.close_count = 0 + + def close(self) -> None: + self.close_count += 1 + + +def _tracking_client_factory( + judge: IntentJudge, + monkeypatch: pytest.MonkeyPatch, +) -> list[_TrackingClient]: + clients: list[_TrackingClient] = [] + lock = threading.Lock() + + def _create() -> _TrackingClient: + client = _TrackingClient() + with lock: + clients.append(client) + return client + + monkeypatch.setattr(judge, "_create_client", _create) + return clients + + # --------------------------------------------------------------------------- # JSON parsing strategies # --------------------------------------------------------------------------- @@ -574,6 +623,769 @@ class TestCancelEventSemantics: assert results[0].tier == "llm_fallback" +# --------------------------------------------------------------------------- +# Parallel batch scheduling +# --------------------------------------------------------------------------- + + +class TestParallelEvaluationScheduling: + def test_parallel_evaluations_default_and_strict_range(self) -> None: + assert JudgeConfig().parallel_evaluations == 1 + assert JudgeConfig(parallel_evaluations=1).parallel_evaluations == 1 + assert JudgeConfig(parallel_evaluations=16).parallel_evaluations == 16 + + invalid_values: list[Any] = [True, False, 0, 17, 1.0, "2", None] + for value in invalid_values: + with pytest.raises( + ValueError, + match=r"judge\.parallel_evaluations.*integer between 1 and 16", + ): + JudgeConfig(parallel_evaluations=value) + + def test_configured_width_sets_exact_peak_and_completes_every_item( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + judge = _make_judge(parallel_evaluations=3) + clients = _tracking_client_factory(judge, monkeypatch) + items = [_make_item(call_id=f"tc_{idx}") for idx in range(7)] + lock = threading.Lock() + first_wave = threading.Event() + release = threading.Event() + done = threading.Event() + started: list[str] = [] + active = 0 + peak = 0 + wait_timed_out = False + + def _evaluate( + item: dict[str, Any], + _messages: list[dict[str, Any]], + _cancel_event: threading.Event | None, + _client: Any, + *, + lane: ModelLane | None = None, + ) -> IntentVerdict: + del lane + nonlocal active, peak, wait_timed_out + with lock: + started.append(str(item["call_id"])) + active += 1 + peak = max(peak, active) + if active == 3: + first_wave.set() + released = release.wait(5.0) + with lock: + active -= 1 + wait_timed_out = wait_timed_out or not released + return _llm_verdict_for(item) + + monkeypatch.setattr(judge, "_evaluate_single", _evaluate) + results: list[IntentVerdict] = [] + judge.evaluate( + items, + [{"role": "user", "content": "test"}], + results.append, + done_callback=done.set, + ) + try: + reached_first_wave = first_wave.wait(2.0) + with lock: + first_snapshot = (list(started), active, peak) + finally: + release.set() + + assert done.wait(5.0) + assert reached_first_wave + assert set(first_snapshot[0]) == {"tc_0", "tc_1", "tc_2"} + assert first_snapshot[1:] == (3, 3) + assert wait_timed_out is False + assert len(started) == 7 + assert peak == 3 + assert active == 0 + assert len(results) == 7 + assert {verdict.call_id for verdict in results} == {f"tc_{idx}" for idx in range(7)} + assert all(verdict.tier == "llm" for verdict in results) + assert len(clients) == 3 + assert all(client.close_count == 1 for client in clients) + + def test_partial_worker_start_failure_delivers_every_fallback_once( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + judge = _make_judge(parallel_evaluations=3) + clients = _tracking_client_factory(judge, monkeypatch) + items = [_make_item(call_id=f"tc_{idx}") for idx in range(5)] + first_evaluation_started = threading.Event() + done = threading.Event() + lock = threading.Lock() + evaluated: list[str] = [] + results: list[IntentVerdict] = [] + done_count = 0 + + def _evaluate( + item: dict[str, Any], + _messages: list[dict[str, Any]], + cancel_event: threading.Event | None, + _client: Any, + *, + lane: ModelLane | None = None, + ) -> None: + del lane + with lock: + evaluated.append(str(item["call_id"])) + first_evaluation_started.set() + if cancel_event is None or not cancel_event.wait(5.0): + raise RuntimeError("worker startup failure did not abort its active sibling") + return None + + def _done() -> None: + nonlocal done_count + with lock: + done_count += 1 + done.set() + + real_thread = threading.Thread + + class _FailingStart: + def start(self) -> None: + if not first_evaluation_started.wait(5.0): + raise RuntimeError("first worker did not begin evaluation") + raise RuntimeError("second worker failed to start") + + def _thread_factory(*args: Any, **kwargs: Any) -> Any: + if kwargs.get("name") == "intent-judge-eval-2": + return _FailingStart() + return real_thread(*args, **kwargs) + + monkeypatch.setattr("turnstone.core.judge.threading.Thread", _thread_factory) + monkeypatch.setattr(judge, "_evaluate_single", _evaluate) + judge.evaluate( + items, + [{"role": "user", "content": "test"}], + results.append, + done_callback=_done, + ) + + assert done.wait(8.0) + assert first_evaluation_started.is_set() + assert evaluated == ["tc_0"] + assert len(results) == 5 + assert sorted(verdict.call_id for verdict in results) == [f"tc_{idx}" for idx in range(5)] + assert all(verdict.tier == "llm_fallback" for verdict in results) + assert all("worker initialization failed" in verdict.reasoning for verdict in results) + assert done_count == 1 + assert len(clients) == 1 + assert clients[0].close_count == 1 + + def test_first_worker_start_failure_delivers_every_fallback_once( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + judge = _make_judge(parallel_evaluations=3) + clients = _tracking_client_factory(judge, monkeypatch) + items = [_make_item(call_id=f"tc_{idx}") for idx in range(4)] + done = threading.Event() + results: list[IntentVerdict] = [] + done_count = 0 + real_thread = threading.Thread + + class _FailingStart: + def start(self) -> None: + raise RuntimeError("first worker failed to start") + + def _thread_factory(*args: Any, **kwargs: Any) -> Any: + if kwargs.get("name") == "intent-judge-eval-1": + return _FailingStart() + return real_thread(*args, **kwargs) + + def _done() -> None: + nonlocal done_count + done_count += 1 + done.set() + + monkeypatch.setattr("turnstone.core.judge.threading.Thread", _thread_factory) + judge.evaluate( + items, + [{"role": "user", "content": "test"}], + results.append, + done_callback=_done, + ) + + assert done.wait(5.0) + assert len(results) == 4 + assert sorted(verdict.call_id for verdict in results) == [f"tc_{idx}" for idx in range(4)] + assert all(verdict.tier == "llm_fallback" for verdict in results) + assert all("worker initialization failed" in verdict.reasoning for verdict in results) + assert done_count == 1 + assert clients == [] + + def test_callbacks_follow_completion_order_without_head_of_line_blocking( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + judge = _make_judge(parallel_evaluations=2) + _tracking_client_factory(judge, monkeypatch) + started = {"tc_0": threading.Event(), "tc_1": threading.Event()} + release = {"tc_0": threading.Event(), "tc_1": threading.Event()} + second_delivered = threading.Event() + done = threading.Event() + callback_order: list[str] = [] + callback_lock = threading.Lock() + + def _evaluate( + item: dict[str, Any], + _messages: list[dict[str, Any]], + _cancel_event: threading.Event | None, + _client: Any, + *, + lane: ModelLane | None = None, + ) -> IntentVerdict: + del lane + call_id = str(item["call_id"]) + started[call_id].set() + release[call_id].wait(5.0) + return _llm_verdict_for(item) + + def _callback(verdict: IntentVerdict) -> None: + with callback_lock: + callback_order.append(verdict.call_id) + if verdict.call_id == "tc_1": + second_delivered.set() + + monkeypatch.setattr(judge, "_evaluate_single", _evaluate) + judge.evaluate( + [_make_item(call_id="tc_0"), _make_item(call_id="tc_1")], + [{"role": "user", "content": "test"}], + _callback, + done_callback=done.set, + ) + try: + both_started = started["tc_0"].wait(2.0) and started["tc_1"].wait(2.0) + if both_started: + release["tc_1"].set() + delivered_before_first = second_delivered.wait(2.0) + with callback_lock: + before_first_release = list(callback_order) + finally: + release["tc_0"].set() + release["tc_1"].set() + + assert done.wait(5.0) + assert both_started + assert delivered_before_first + assert before_first_release == ["tc_1"] + assert callback_order == ["tc_1", "tc_0"] + + def test_duplicate_nonempty_call_ids_remain_distinct_indexed_work_items( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + judge = _make_judge(parallel_evaluations=2) + _tracking_client_factory(judge, monkeypatch) + items = [ + _make_item(call_id="duplicate", func_name="bash"), + _make_item(call_id="duplicate", func_name="write_file"), + ] + first_wave = threading.Event() + release = threading.Event() + done = threading.Event() + lock = threading.Lock() + started: list[str] = [] + callback_attempts: list[tuple[str, str]] = [] + + def _evaluate( + item: dict[str, Any], + _messages: list[dict[str, Any]], + _cancel_event: threading.Event | None, + _client: Any, + *, + lane: ModelLane | None = None, + ) -> IntentVerdict: + del lane + with lock: + started.append(str(item["func_name"])) + if len(started) == 2: + first_wave.set() + release.wait(5.0) + return _llm_verdict_for(item) + + monkeypatch.setattr(judge, "_evaluate_single", _evaluate) + judge.evaluate( + items, + [{"role": "user", "content": "test"}], + lambda verdict: callback_attempts.append((verdict.call_id, verdict.func_name)), + done_callback=done.set, + ) + try: + reached_first_wave = first_wave.wait(2.0) + finally: + release.set() + + assert done.wait(5.0) + assert reached_first_wave + assert set(started) == {"bash", "write_file"} + assert sorted(callback_attempts) == [ + ("duplicate", "bash"), + ("duplicate", "write_file"), + ] + + def test_cancel_stops_queued_dispatch_and_delivers_each_fallback_once( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + judge = _make_judge(parallel_evaluations=2) + clients = _tracking_client_factory(judge, monkeypatch) + items = [_make_item(call_id=f"tc_{idx}") for idx in range(5)] + cancel = threading.Event() + first_wave = threading.Event() + release = threading.Event() + done = threading.Event() + lock = threading.Lock() + started: list[str] = [] + results: list[IntentVerdict] = [] + done_count = 0 + + def _evaluate( + item: dict[str, Any], + _messages: list[dict[str, Any]], + cancel_event: threading.Event | None, + _client: Any, + *, + lane: ModelLane | None = None, + ) -> IntentVerdict | None: + del lane + with lock: + started.append(str(item["call_id"])) + if len(started) == 2: + first_wave.set() + release.wait(5.0) + if cancel_event is not None and cancel_event.is_set(): + return None + return _llm_verdict_for(item) + + def _done() -> None: + nonlocal done_count + with lock: + done_count += 1 + done.set() + + monkeypatch.setattr(judge, "_evaluate_single", _evaluate) + judge.evaluate( + items, + [{"role": "user", "content": "test"}], + results.append, + cancel_event=cancel, + done_callback=_done, + ) + try: + reached_first_wave = first_wave.wait(2.0) + cancel.set() + finally: + release.set() + + assert done.wait(5.0) + assert reached_first_wave + assert set(started) == {"tc_0", "tc_1"} + assert len(results) == 5 + assert sorted(verdict.call_id for verdict in results) == [f"tc_{idx}" for idx in range(5)] + assert all(verdict.tier == "llm_fallback" for verdict in results) + assert all("cancelled" in verdict.reasoning for verdict in results) + assert done_count == 1 + assert len(clients) == 2 + assert all(client.close_count == 1 for client in clients) + + def test_backend_auth_failure_aborts_unstarted_work_without_duplicate_verdicts( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + judge = _make_judge(parallel_evaluations=3) + clients = _tracking_client_factory(judge, monkeypatch) + items = [_make_item(call_id=f"tc_{idx}") for idx in range(6)] + all_started = threading.Event() + done = threading.Event() + lock = threading.Lock() + started: list[str] = [] + results: list[IntentVerdict] = [] + + def _evaluate( + item: dict[str, Any], + _messages: list[dict[str, Any]], + cancel_event: threading.Event | None, + _client: Any, + *, + lane: ModelLane | None = None, + ) -> IntentVerdict | None: + del lane + call_id = str(item["call_id"]) + with lock: + started.append(call_id) + if len(started) == 3: + all_started.set() + if call_id == "tc_0": + if not all_started.wait(5.0): + raise RuntimeError("scheduler did not start the configured first wave") + raise BackendAuthUnavailableError("mint unavailable") + if cancel_event is None or not cancel_event.wait(5.0): + raise RuntimeError("batch auth failure did not abort active siblings") + return None + + monkeypatch.setattr(judge, "_evaluate_single", _evaluate) + judge.evaluate( + items, + [{"role": "user", "content": "test"}], + results.append, + done_callback=done.set, + ) + + assert done.wait(8.0) + assert set(started) == {"tc_0", "tc_1", "tc_2"} + assert len(results) == 6 + assert sorted(verdict.call_id for verdict in results) == [f"tc_{idx}" for idx in range(6)] + assert all(verdict.tier == "llm_fallback" for verdict in results) + assert all("backend authentication failed" in verdict.reasoning for verdict in results) + assert len(clients) == 3 + assert all(client.close_count == 1 for client in clients) + + def test_callback_failure_does_not_abort_batch_and_closes_every_worker_client( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + judge = _make_judge(parallel_evaluations=3) + clients = _tracking_client_factory(judge, monkeypatch) + items = [_make_item(call_id=f"tc_{idx}") for idx in range(5)] + first_wave = threading.Event() + release = threading.Event() + done = threading.Event() + lock = threading.Lock() + started_count = 0 + callback_attempts: list[str] = [] + closed_at_done: list[int] = [] + + def _evaluate( + item: dict[str, Any], + _messages: list[dict[str, Any]], + _cancel_event: threading.Event | None, + _client: Any, + *, + lane: ModelLane | None = None, + ) -> IntentVerdict: + del lane + nonlocal started_count + with lock: + started_count += 1 + if started_count == 3: + first_wave.set() + release.wait(5.0) + return _llm_verdict_for(item) + + def _callback(verdict: IntentVerdict) -> None: + callback_attempts.append(verdict.call_id) + if verdict.call_id == "tc_1": + raise RuntimeError("consumer failed") + + def _done() -> None: + closed_at_done.append(sum(client.close_count for client in clients)) + done.set() + + monkeypatch.setattr(judge, "_evaluate_single", _evaluate) + judge.evaluate( + items, + [{"role": "user", "content": "test"}], + _callback, + done_callback=_done, + ) + try: + reached_first_wave = first_wave.wait(2.0) + finally: + release.set() + + assert done.wait(5.0) + assert reached_first_wave + assert sorted(callback_attempts) == [f"tc_{idx}" for idx in range(5)] + assert len(clients) == 3 + assert all(client.close_count == 1 for client in clients) + assert closed_at_done == [3] + + def test_model_alias_admission_is_a_second_concurrency_ceiling( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + provider = _make_mock_provider(_good_verdict_json()) + judge = _make_judge(provider, parallel_evaluations=4) + clients = _tracking_client_factory(judge, monkeypatch) + admission = ModelAdmission("judge", 2) + judge._lane = replace( + judge._lane, + alias="judge", + admission=admission, + ) + items = [_make_item(call_id=f"tc_{idx}") for idx in range(6)] + first_wave = threading.Event() + release = threading.Event() + cancel = threading.Event() + done = threading.Event() + lock = threading.Lock() + calls = 0 + active = 0 + peak = 0 + wait_timed_out = False + + def _stream(**_kwargs: Any) -> Any: + nonlocal calls, active, peak, wait_timed_out + with lock: + calls += 1 + active += 1 + peak = max(peak, active) + if active == 2: + first_wave.set() + released = release.wait(5.0) + with lock: + active -= 1 + wait_timed_out = wait_timed_out or not released + return as_stream(_mock_result(_good_verdict_json())) + + provider.create_streaming.side_effect = _stream + results: list[IntentVerdict] = [] + judge.evaluate( + items, + [{"role": "user", "content": "test"}], + results.append, + cancel_event=cancel, + done_callback=done.set, + ) + reached_first_wave = first_wave.wait(3.0) + active_snapshot = admission.snapshot() + release.set() + finished = done.wait(10.0) + if not finished: + cancel.set() + release.set() + done.wait(5.0) + + assert finished + assert reached_first_wave + assert active_snapshot.in_flight == 2 + assert active_snapshot.queued == 0 + assert calls == 6 + assert peak == 2 + assert active == 0 + assert wait_timed_out is False + assert len(results) == 6 + assert all(verdict.tier == "llm" for verdict in results) + assert admission.snapshot().in_flight == 0 + assert admission.snapshot().queued == 0 + assert len(clients) == 2 + assert all(client.close_count == 1 for client in clients) + + def test_hot_alias_resize_narrows_subsequent_scheduler_dispatch( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + provider = _make_mock_provider(_good_verdict_json()) + judge = _make_judge(provider, parallel_evaluations=3) + clients = _tracking_client_factory(judge, monkeypatch) + admission = ModelAdmission("judge", 3) + judge._lane = replace( + judge._lane, + alias="judge", + admission=admission, + ) + items = [_make_item(call_id=f"tc_{idx}") for idx in range(6)] + first_wave_started = threading.Event() + release_first_wave = threading.Event() + later_started = threading.Event() + release_later = threading.Event() + cancel = threading.Event() + done = threading.Event() + lock = threading.Lock() + results: list[IntentVerdict] = [] + calls = 0 + first_active = 0 + first_peak = 0 + later_active = 0 + later_peak = 0 + wait_timed_out = False + + def _stream(**_kwargs: Any) -> Any: + nonlocal calls, first_active, first_peak, later_active, later_peak, wait_timed_out + with lock: + calls += 1 + call_number = calls + if call_number <= 3: + first_active += 1 + first_peak = max(first_peak, first_active) + if first_active == 3: + first_wave_started.set() + else: + later_active += 1 + later_peak = max(later_peak, later_active) + later_started.set() + released = release_first_wave.wait(5.0) if call_number <= 3 else release_later.wait(5.0) + with lock: + wait_timed_out = wait_timed_out or not released + if call_number <= 3: + first_active -= 1 + else: + later_active -= 1 + return as_stream(_mock_result(_good_verdict_json())) + + provider.create_streaming.side_effect = _stream + judge.evaluate( + items, + [{"role": "user", "content": "test"}], + results.append, + cancel_event=cancel, + done_callback=done.set, + ) + + reached_first_wave = first_wave_started.wait(3.0) + initial_snapshot = admission.snapshot() + admission.set_limit(1) + release_first_wave.set() + reached_later = later_started.wait(5.0) + queued_snapshot = False + deadline = time.monotonic() + 3.0 + while time.monotonic() < deadline: + snapshot = admission.snapshot() + with lock: + later_call_snapshot = calls - 3 + if snapshot.in_flight == 1 and snapshot.queued == 2: + queued_snapshot = True + break + if later_call_snapshot > 1: + break + time.sleep(0.005) + with lock: + narrowed_snapshot = (calls - 3, later_active, later_peak) + release_later.set() + finished = done.wait(10.0) + if not finished: + cancel.set() + release_first_wave.set() + release_later.set() + done.wait(5.0) + + assert finished + assert reached_first_wave + assert initial_snapshot.in_flight == 3 + assert initial_snapshot.queued == 0 + assert reached_later + assert queued_snapshot + assert narrowed_snapshot == (1, 1, 1) + assert calls == 6 + assert first_peak == 3 + assert later_peak == 1 + assert first_active == 0 + assert later_active == 0 + assert wait_timed_out is False + assert len(results) == 6 + assert sorted(verdict.call_id for verdict in results) == [f"tc_{idx}" for idx in range(6)] + assert all(verdict.tier == "llm" for verdict in results) + final_snapshot = admission.snapshot() + assert final_snapshot.limit == 1 + assert final_snapshot.in_flight == 0 + assert final_snapshot.queued == 0 + assert len(clients) == 3 + assert all(client.close_count == 1 for client in clients) + + def test_concurrent_batches_share_one_alias_admission_ceiling( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + provider = _make_mock_provider(_good_verdict_json()) + admission = ModelAdmission("shared-judge", 2) + judges = [ + _make_judge(provider, parallel_evaluations=4), + _make_judge(provider, parallel_evaluations=4), + ] + client_groups = [_tracking_client_factory(judge, monkeypatch) for judge in judges] + for judge in judges: + judge._lane = replace( + judge._lane, + alias="shared-judge", + admission=admission, + ) + + first_wave = threading.Event() + release = threading.Event() + cancels = [threading.Event(), threading.Event()] + done_events = [threading.Event(), threading.Event()] + done_counts = [0, 0] + finals: list[list[IntentVerdict]] = [[], []] + lock = threading.Lock() + calls = 0 + active = 0 + peak = 0 + wait_timed_out = False + + def _stream(**_kwargs: Any) -> Any: + nonlocal calls, active, peak, wait_timed_out + with lock: + calls += 1 + active += 1 + peak = max(peak, active) + if active == 2: + first_wave.set() + released = release.wait(5.0) + with lock: + active -= 1 + wait_timed_out = wait_timed_out or not released + return as_stream(_mock_result(_good_verdict_json())) + + def _done(batch_index: int) -> None: + with lock: + done_counts[batch_index] += 1 + done_events[batch_index].set() + + provider.create_streaming.side_effect = _stream + for batch_index, judge in enumerate(judges): + items = [ + _make_item(call_id=f"batch-{batch_index}-tc-{item_index}") + for item_index in range(3) + ] + judge.evaluate( + items, + [{"role": "user", "content": f"batch {batch_index}"}], + finals[batch_index].append, + cancel_event=cancels[batch_index], + done_callback=lambda index=batch_index: _done(index), + ) + + reached_first_wave = first_wave.wait(3.0) + queued_snapshot = False + deadline = time.monotonic() + 2.0 + while time.monotonic() < deadline: + snapshot = admission.snapshot() + if snapshot.in_flight == 2 and snapshot.queued == 2: + queued_snapshot = True + break + time.sleep(0.005) + release.set() + finished = all(done.wait(10.0) for done in done_events) + if not finished: + for cancel in cancels: + cancel.set() + release.set() + for done in done_events: + done.wait(5.0) + + assert finished + assert reached_first_wave + assert queued_snapshot + assert calls == 6 + assert peak == 2 + assert active == 0 + assert wait_timed_out is False + assert [len(batch_finals) for batch_finals in finals] == [3, 3] + assert all(verdict.tier == "llm" for batch_finals in finals for verdict in batch_finals) + assert done_counts == [1, 1] + assert admission.snapshot().in_flight == 0 + assert admission.snapshot().queued == 0 + assert [len(clients) for clients in client_groups] == [2, 2] + assert all(client.close_count == 1 for clients in client_groups for client in clients) + + # --------------------------------------------------------------------------- # Multi-turn tool use # --------------------------------------------------------------------------- @@ -809,6 +1621,10 @@ class TestConfidenceArbitration: class TestPathBlocking: + def test_exact_protected_roots_are_blocked_without_reading_them(self): + for root in ("/etc", "/root", "/proc", "/sys", "/dev"): + assert IntentJudge._is_path_blocked(Path(root)) is True + def test_etc_blocked(self): assert IntentJudge._is_path_blocked(Path("/etc/passwd")) is True @@ -854,6 +1670,30 @@ class TestPathBlocking: def test_project_path_not_blocked(self): assert IntentJudge._is_path_blocked(Path("/home/user/project/main.py")) is False + def test_benign_symlink_to_protected_suffix_is_blocked(self, tmp_path): + target = tmp_path / "harmless-secret.pem" + target.write_text("test-only certificate material") + link = tmp_path / "release-notes.txt" + link.symlink_to(target) + + assert IntentJudge._is_path_blocked(link) is True + result = IntentJudge._exec_read_only_tool("read_file", {"path": str(link)}) + assert "access denied" in result + assert "test-only certificate material" not in result + + def test_benign_symlink_to_protected_component_is_blocked(self, tmp_path): + protected_dir = tmp_path / ".ssh" + protected_dir.mkdir() + target = protected_dir / "test-identity" + target.write_text("test-only private material") + link = tmp_path / "meeting-notes.txt" + link.symlink_to(target) + + assert IntentJudge._is_path_blocked(link) is True + result = IntentJudge._exec_read_only_tool("read_file", {"path": str(link)}) + assert "access denied" in result + assert "test-only private material" not in result + # --------------------------------------------------------------------------- # Read-only tool execution @@ -883,6 +1723,39 @@ class TestReadOnlyToolExecution: assert "truncated" in result assert len(result) < 50_000 + def test_read_file_bounds_the_read_request_itself( + self, + tmp_path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + test_file = tmp_path / "bounded.txt" + test_file.write_text("placeholder") + requested: list[int] = [] + + class _BoundedReader: + def __enter__(self) -> _BoundedReader: + return self + + def __exit__(self, *_exc: object) -> None: + return None + + def read(self, size: int) -> str: + requested.append(size) + return "x" * size + + def _open(_path: Path, *_args: Any, **_kwargs: Any) -> _BoundedReader: + return _BoundedReader() + + monkeypatch.setattr(Path, "open", _open) + result = IntentJudge._exec_read_only_tool( + "read_file", + {"path": str(test_file)}, + ) + + assert requested == [32_769] + assert result[:32_768] == "x" * 32_768 + assert result[32_768:].startswith("\n... (truncated") + def test_list_directory_success(self, tmp_path): (tmp_path / "file_a.txt").touch() (tmp_path / "dir_b").mkdir() @@ -890,6 +1763,49 @@ class TestReadOnlyToolExecution: assert "dir_b/" in result assert "file_a.txt" in result + def test_list_directory_stops_after_limit_probe_and_marks_omission( + self, + tmp_path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + class _Entry: + def __init__(self, index: int) -> None: + self.name = f"entry-{index:03}.txt" + + def is_dir(self) -> bool: + return False + + class _BoundedEntries: + def __init__(self) -> None: + self.next_calls = 0 + + def __iter__(self) -> _BoundedEntries: + return self + + def __next__(self) -> _Entry: + if self.next_calls >= 201: + raise AssertionError("list_directory requested a 202nd entry") + entry = _Entry(self.next_calls) + self.next_calls += 1 + return entry + + entries = _BoundedEntries() + monkeypatch.setattr(Path, "iterdir", lambda _path: entries) + + result = IntentJudge._exec_read_only_tool( + "list_directory", + {"path": str(tmp_path)}, + ) + lines = result.splitlines() + + assert entries.next_calls == 201 + assert len(lines) == 201 + assert lines[:2] == [" entry-000.txt", " entry-001.txt"] + assert lines[-2:] == [ + " entry-199.txt", + " ... (additional entries omitted)", + ] + def test_list_directory_not_found(self): result = IntentJudge._exec_read_only_tool("list_directory", {"path": "/nonexistent/dir"}) assert "Error" in result diff --git a/tests/test_judge_parallel_live.py b/tests/test_judge_parallel_live.py new file mode 100644 index 00000000..e153e1bf --- /dev/null +++ b/tests/test_judge_parallel_live.py @@ -0,0 +1,382 @@ +"""Live request-count assertions for parallel intent-judge batches. + +These tests run :class:`turnstone.core.judge.IntentJudge` through the real +OpenAI-compatible backend configured by ``TURNSTONE_TEST_BASE_URL`` (the same +contract as :mod:`tests.test_server_live`). A local threaded proxy counts the +HTTP fan-out and fully drains each genuine upstream response. It then returns +a deterministic, valid Chat Completions SSE verdict so the assertions measure +Turnstone's scheduler rather than a live model's JSON-formatting reliability. + +Run explicitly with a live backend:: + + pytest tests/test_judge_parallel_live.py -m live -v +""" + +from __future__ import annotations + +import json +import threading +from collections import Counter +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import TYPE_CHECKING, Any + +import httpx +import pytest +from openai import OpenAI + +from tests.test_model_admission_live import ( + _HOP_BY_HOP_HEADERS, + _LIVE_API_KEY, + _LIVE_BASE_URL, + _ConcurrencyCounter, +) +from turnstone.core.judge import IntentJudge, IntentVerdict, JudgeConfig +from turnstone.core.model_registry import ModelConfig, ModelRegistry +from turnstone.core.model_turn import resolve_model_binding + +if TYPE_CHECKING: + from types import TracebackType + + +_VERDICT_CONTENT = json.dumps( + { + "intent_summary": "Live scheduler probe", + "risk_level": "low", + "confidence": 0.99, + "recommendation": "approve", + "reasoning": "The deterministic proxy verdict confirms one completed judge dispatch.", + "evidence": ["The genuine upstream response was drained before this verdict."], + }, + separators=(",", ":"), +) + + +def _deterministic_sse() -> bytes: + chunks = ( + { + "id": "turnstone-live-judge", + "object": "chat.completion.chunk", + "created": 0, + "model": "turnstone-live-judge", + "choices": [ + { + "index": 0, + "delta": {"role": "assistant", "content": _VERDICT_CONTENT}, + "finish_reason": None, + } + ], + }, + { + "id": "turnstone-live-judge", + "object": "chat.completion.chunk", + "created": 0, + "model": "turnstone-live-judge", + "choices": [ + { + "index": 0, + "delta": {}, + "finish_reason": "stop", + } + ], + }, + ) + framed = "".join(f"data: {json.dumps(chunk, separators=(',', ':'))}\n\n" for chunk in chunks) + return (framed + "data: [DONE]\n\n").encode() + + +class _JudgeCountingProxyServer(ThreadingHTTPServer): + daemon_threads = True + + def __init__(self, counter: _ConcurrencyCounter) -> None: + super().__init__(("127.0.0.1", 0), _JudgeCountingProxyHandler) + self.counter = counter + self.upstream_base_url = _LIVE_BASE_URL.rstrip("/") + self.upstream_authorization = f"Bearer {_LIVE_API_KEY}" + self.alias_by_authorization: dict[str, str] = {} + + +class _JudgeCountingProxyHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + @property + def _proxy(self) -> _JudgeCountingProxyServer: + if not isinstance(self.server, _JudgeCountingProxyServer): + raise TypeError("judge counting handler requires _JudgeCountingProxyServer") + return self.server + + def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + alias = self._proxy.alias_by_authorization.get(self.headers.get("Authorization", "")) + if alias is None: + self._write_json_error(401, "unknown live-test alias credential") + return + + try: + content_length = int(self.headers.get("Content-Length", "0")) + except ValueError: + self._write_json_error(400, "invalid content length") + return + body = self.rfile.read(content_length) + # The test measures dispatch, admission, and full upstream drain. Keep + # the real probe inexpensive even if the configured model tends to + # deliberate: its content is replaced only after the response completes. + try: + upstream_body = json.loads(body) + except (json.JSONDecodeError, UnicodeDecodeError): + upstream_body = None + if isinstance(upstream_body, dict): + if "max_tokens" in upstream_body: + upstream_body["max_tokens"] = min(int(upstream_body["max_tokens"]), 16) + if "max_completion_tokens" in upstream_body: + upstream_body["max_completion_tokens"] = min( + int(upstream_body["max_completion_tokens"]), 16 + ) + body = json.dumps(upstream_body, separators=(",", ":")).encode() + + counter = self._proxy.counter + counter.enter(alias) + completed = False + try: + headers = { + key: value + for key, value in self.headers.items() + if key.lower() not in _HOP_BY_HOP_HEADERS + and key.lower() not in {"authorization", "host"} + } + headers["Authorization"] = self._proxy.upstream_authorization + headers["Accept-Encoding"] = "identity" + headers["Content-Length"] = str(len(body)) + counter.begin_forward(alias) + try: + upstream = httpx.post( + f"{self._proxy.upstream_base_url}{self.path}", + content=body, + headers=headers, + timeout=httpx.Timeout(60.0, connect=5.0), + ) + finally: + counter.end_forward(alias) + # ``httpx.post`` has consumed the complete streaming body here. + # A genuine backend rejection must fail the probe; only a successful + # response is replaced with deterministic judge output. + upstream.raise_for_status() + payload = _deterministic_sse() + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Content-Length", str(len(payload))) + self.send_header("Connection", "close") + self.end_headers() + self.wfile.write(payload) + self.wfile.flush() + completed = True + except Exception: + self._write_json_error(502, "live backend judge proxy failure") + finally: + counter.leave(alias, completed=completed) + + def log_message(self, _format: str, *args: Any) -> None: + """Keep expected local proxy traffic out of pytest output.""" + + def _write_json_error(self, status: int, detail: str) -> None: + payload = json.dumps({"error": {"message": detail}}).encode() + try: + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.send_header("Connection", "close") + self.end_headers() + self.wfile.write(payload) + self.wfile.flush() + except (BrokenPipeError, ConnectionError): + pass + + +class _JudgeCountingProxy: + def __init__(self, counter: _ConcurrencyCounter) -> None: + self._server = _JudgeCountingProxyServer(counter) + self._thread = threading.Thread( + target=self._server.serve_forever, + name="judge-parallel-live-proxy", + daemon=True, + ) + + @property + def base_url(self) -> str: + host, port = self._server.server_address[:2] + if not isinstance(host, str) or not isinstance(port, int): + raise TypeError("judge counting proxy did not bind an IPv4 TCP address") + return f"http://{host}:{port}" + + def credential_for(self, alias: str) -> str: + credential = f"turnstone-live-judge-{alias}" + self._server.alias_by_authorization[f"Bearer {credential}"] = alias + return credential + + def __enter__(self) -> _JudgeCountingProxy: + self._thread.start() + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + del exc_type, exc_value, traceback + self._server.shutdown() + self._server.server_close() + self._thread.join(timeout=5.0) + if self._thread.is_alive(): + raise AssertionError("live judge counting proxy did not stop") + + +@pytest.fixture(scope="module") +def live_model_id() -> str: + """Auto-detect the model using the established live-backend contract.""" + client = OpenAI(base_url=_LIVE_BASE_URL, api_key=_LIVE_API_KEY) + try: + models = client.models.list() + finally: + client.close() + ids = [model.id for model in models.data] + assert ids, "No models found on the live backend" + return ids[0] + + +@dataclass(frozen=True) +class _JudgeRun: + heuristics: list[IntentVerdict] + finals: list[IntentVerdict] + done_calls: int + + +def _run_live_judge_batch( + proxy: _JudgeCountingProxy, + model: str, + *, + alias_limit: int, + parallel_evaluations: int, + batch_size: int, +) -> _JudgeRun: + alias = "live-judge" + config = ModelConfig( + alias=alias, + base_url=proxy.base_url, + api_key=proxy.credential_for(alias), + model=model, + provider="openai-compatible", + max_concurrency=alias_limit, + ) + registry = ModelRegistry({alias: config}, default=alias) + done = threading.Event() + finals: list[IntentVerdict] = [] + done_calls = 0 + + def _done() -> None: + nonlocal done_calls + done_calls += 1 + done.set() + + try: + binding = resolve_model_binding(registry, alias) + judge = IntentJudge( + JudgeConfig( + enabled=True, + read_only_tools=False, + timeout=60.0, + parallel_evaluations=parallel_evaluations, + ), + binding, + ) + items = [ + { + "func_name": "bash", + "func_args": {"command": f"echo live-{index}"}, + "approval_label": "bash", + "call_id": f"live-call-{index}", + } + for index in range(batch_size) + ] + heuristics = judge.evaluate( + items, + [{"role": "user", "content": "Run the independent live scheduler probes."}], + finals.append, + done_callback=_done, + ) + assert done.wait(90.0), "live intent-judge batch did not finish" + return _JudgeRun(heuristics=heuristics, finals=finals, done_calls=done_calls) + finally: + registry.shutdown() + + +def _assert_exact_batch(run: _JudgeRun, *, batch_size: int) -> None: + expected_ids = {f"live-call-{index}" for index in range(batch_size)} + assert len(run.heuristics) == batch_size + assert len(run.finals) == batch_size + assert Counter(verdict.call_id for verdict in run.finals) == Counter( + {call_id: 1 for call_id in expected_ids} + ) + assert all(verdict.tier == "llm" for verdict in run.finals) + assert all(verdict.intent_summary == "Live scheduler probe" for verdict in run.finals) + assert run.done_calls == 1 + + +@pytest.mark.live +class TestLiveParallelJudgeCounts: + """Count genuine upstream calls made by the intent-judge scheduler.""" + + def test_parallel_width_sets_exact_batch_peak(self, live_model_id: str) -> None: + counter = _ConcurrencyCounter(rendezvous_size=3) + with _JudgeCountingProxy(counter) as proxy: + run = _run_live_judge_batch( + proxy, + live_model_id, + alias_limit=0, + parallel_evaluations=3, + batch_size=5, + ) + + _assert_exact_batch(run, batch_size=5) + counts = counter.snapshot() + assert counts.requests == 5 + assert counts.forwarded == 5 + assert counts.completed == 5 + assert counts.failed == 0 + assert counts.active == 0 + assert counts.peak == 3 + assert counts.forward_active == 0 + assert counts.forward_peak == 3 + assert counts.requests_by_alias == {"live-judge": 5} + assert counts.forwarded_by_alias == {"live-judge": 5} + assert counts.completed_by_alias == {"live-judge": 5} + assert counts.peak_by_alias == {"live-judge": 3} + assert counts.forward_peak_by_alias == {"live-judge": 3} + assert counts.rendezvous_timed_out is False + + def test_alias_cap_reduces_parallel_judge_peak(self, live_model_id: str) -> None: + counter = _ConcurrencyCounter(rendezvous_size=2) + with _JudgeCountingProxy(counter) as proxy: + run = _run_live_judge_batch( + proxy, + live_model_id, + alias_limit=2, + parallel_evaluations=4, + batch_size=4, + ) + + _assert_exact_batch(run, batch_size=4) + counts = counter.snapshot() + assert counts.requests == 4 + assert counts.forwarded == 4 + assert counts.completed == 4 + assert counts.failed == 0 + assert counts.active == 0 + assert counts.peak == 2 + assert counts.forward_active == 0 + assert counts.forward_peak == 2 + assert counts.requests_by_alias == {"live-judge": 4} + assert counts.forwarded_by_alias == {"live-judge": 4} + assert counts.completed_by_alias == {"live-judge": 4} + assert counts.peak_by_alias == {"live-judge": 2} + assert counts.forward_peak_by_alias == {"live-judge": 2} + assert counts.rendezvous_timed_out is False diff --git a/tests/test_session.py b/tests/test_session.py index 0950040b..3b4653fe 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -3130,8 +3130,13 @@ class TestLiveConfigUpdate: # Default: enabled=True assert session._judge_cfg.enabled is True + assert session._judge_cfg.parallel_evaluations == 1 - # Admin disables the judge + # Admin changes behavioral settings for the next batch. + cs.set("judge.parallel_evaluations", 6, changed_by="test") + assert session._judge_cfg.parallel_evaluations == 6 + + # Admin disables the judge. cs.set("judge.enabled", False, changed_by="test") assert session._judge_cfg.enabled is False @@ -3144,6 +3149,7 @@ class TestLiveConfigUpdate: values = {key: defn.default for key, defn in SETTINGS.items()} values["judge.smart_approvals"] = True values["judge.confidence_threshold"] = 0.4 + values["judge.parallel_evaluations"] = 5 return 7, values def get(self, _key): @@ -3160,8 +3166,16 @@ class TestLiveConfigUpdate: assert direct is not None assert stable is not None - assert (direct.smart_approvals, direct.confidence_threshold) == (True, 0.4) - assert (stable.smart_approvals, stable.confidence_threshold) == (True, 0.4) + assert ( + direct.smart_approvals, + direct.confidence_threshold, + direct.parallel_evaluations, + ) == (True, 0.4, 5) + assert ( + stable.smart_approvals, + stable.confidence_threshold, + stable.parallel_evaluations, + ) == (True, 0.4, 5) assert version == 7 def test_judge_client_config_stays_frozen(self, tmp_db): diff --git a/tests/test_settings_api.py b/tests/test_settings_api.py index c0c834b8..eccc72cb 100644 --- a/tests/test_settings_api.py +++ b/tests/test_settings_api.py @@ -175,6 +175,23 @@ class TestUpdateSetting: assert r.status_code == 400 assert "minimum" in r.json()["error"] + @pytest.mark.parametrize("invalid", [True, 1.0, "01", "+1", "1.0"]) + def test_judge_parallel_evaluations_requires_strict_integer(self, client, invalid): + r = client.put( + "/v1/api/admin/settings/judge.parallel_evaluations", + json={"value": invalid}, + ) + assert r.status_code == 400 + assert "Cannot convert" in r.json()["error"] + + def test_judge_parallel_evaluations_accepts_integer_input_string(self, client): + r = client.put( + "/v1/api/admin/settings/judge.parallel_evaluations", + json={"value": "8"}, + ) + assert r.status_code == 200 + assert r.json()["value"] == 8 + def test_update_then_list_shows_storage(self, client): client.put( "/v1/api/admin/settings/tools.timeout", diff --git a/tests/test_settings_registry.py b/tests/test_settings_registry.py index b26a0072..6b7fcc4e 100644 --- a/tests/test_settings_registry.py +++ b/tests/test_settings_registry.py @@ -63,6 +63,25 @@ class TestValidateKey: assert defn.min_value == 0.0 assert defn.max_value == 1.0 + def test_judge_parallel_evaluations_registered(self): + defn = validate_key("judge.parallel_evaluations") + assert defn.type == "int" + assert defn.default == 1 + assert defn.min_value == 1 + assert defn.max_value == 16 + assert defn.section == "judge" + assert defn.strict_int is True + + assert validate_value("judge.parallel_evaluations", "1") == 1 + assert validate_value("judge.parallel_evaluations", 16) == 16 + for invalid in (True, 1.0, "01", "+1", " 1", "1.0"): + with pytest.raises(ValueError, match="Cannot convert"): + validate_value("judge.parallel_evaluations", invalid) + with pytest.raises(ValueError, match="minimum"): + validate_value("judge.parallel_evaluations", 0) + with pytest.raises(ValueError, match="maximum"): + validate_value("judge.parallel_evaluations", 17) + def test_smart_approvals_bool_coercion(self): assert validate_value("judge.smart_approvals", "true") is True assert validate_value("judge.smart_approvals", "false") is False diff --git a/turnstone.example.toml b/turnstone.example.toml index 11cb989c..97890009 100644 --- a/turnstone.example.toml +++ b/turnstone.example.toml @@ -136,14 +136,14 @@ # NB: serving Qwen3-Reranker via vLLM REQUIRES --chat-template # (the model's chat_template.jinja) or scores are near-random. -# --- Judge (turnstone, node) --- +# --- Judge (turnstone CLI; server/console use Admin -> Judge) --- [judge] # enabled = true # Enable intent validation -# smart_approvals = false # Auto-approve high-confidence "approve" LLM verdicts (opt-in) -# confidence_threshold = 0.95 # Smart Approvals auto-approve bar (LLM recommendation=approve) -# output_guard = true # Scan tool output for security signals -# redact_secrets = true # Redact detected credentials in output +# model = "" # Registered judge alias; empty = same as session +# confidence_threshold = 0.95 # Judge verdict confidence threshold +# timeout = 120.0 # Per-turn LLM judge timeout in seconds +# parallel_evaluations = 1 # Concurrent LLM evaluations per tool-call batch (1-16) # --- Memory (turnstone, node) --- diff --git a/turnstone/cli.py b/turnstone/cli.py index 86da1fd9..3dcbd789 100644 --- a/turnstone/cli.py +++ b/turnstone/cli.py @@ -995,6 +995,16 @@ def _close_all_sessions(manager: SessionManager) -> None: print(dim(" (interrupted — background shells already signalled)")) +def _judge_parallel_evaluations_arg(value: str) -> int: + """Argparse adapter for the registry's strict bounded integer contract.""" + from turnstone.core.settings_registry import validate_value + + try: + return int(validate_value("judge.parallel_evaluations", value)) + except ValueError as exc: + raise argparse.ArgumentTypeError(str(exc)) from exc + + def main() -> None: parser = argparse.ArgumentParser( description="Interactive CLI for OpenAI-compatible models with tool calling.", @@ -1186,6 +1196,14 @@ def main() -> None: default=120.0, help="LLM judge timeout in seconds (default: 120)", ) + judge_group.add_argument( + "--judge-parallel-evaluations", + dest="judge_parallel_evaluations", + type=_judge_parallel_evaluations_arg, + default=1, + metavar="N", + help="Concurrent LLM evaluations within one judge batch, 1-16 (default: 1)", + ) judge_group.add_argument( "--judge-confidence", dest="judge_confidence", @@ -1316,6 +1334,7 @@ def main() -> None: model=args.judge_model, confidence_threshold=args.judge_confidence, timeout=args.judge_timeout, + parallel_evaluations=args.judge_parallel_evaluations, ) # ChatSession factory — captures shared config for creating workstreams diff --git a/turnstone/console/session_factory.py b/turnstone/console/session_factory.py index d83e5303..c9d5a96a 100644 --- a/turnstone/console/session_factory.py +++ b/turnstone/console/session_factory.py @@ -80,6 +80,7 @@ def build_console_session_factory( confidence_threshold=config_store.get("judge.confidence_threshold"), max_context_ratio=config_store.get("judge.max_context_ratio"), timeout=config_store.get("judge.timeout"), + parallel_evaluations=config_store.get("judge.parallel_evaluations", 1), read_only_tools=config_store.get("judge.read_only_tools"), output_guard=config_store.get("judge.output_guard"), output_guard_budget_seconds=config_store.get("judge.output_guard_budget_seconds"), diff --git a/turnstone/console/static/governance.js b/turnstone/console/static/governance.js index 4c9dc452..4a88ed92 100644 --- a/turnstone/console/static/governance.js +++ b/turnstone/console/static/governance.js @@ -3859,7 +3859,7 @@ function renderJudgeSettings() { ">" + '' + 'Enabled'; - } else if (s.type === "float") { + } else if (s.type === "float" || s.type === "int") { // currentVal/min/max are expected to be numeric, but // ``admin_list_judge_settings`` can fall back to returning the // raw stored string when deserialization fails — escape + @@ -3867,7 +3867,9 @@ function renderJudgeSettings() { // out of the value/min/max attribute boundary. inputHtml = '
' + - ' None: + if type(self.parallel_evaluations) is not int or not ( + 1 <= self.parallel_evaluations <= _MAX_PARALLEL_EVALUATIONS + ): + raise ValueError( + "judge.parallel_evaluations must be an integer between 1 and " + f"{_MAX_PARALLEL_EVALUATIONS}" + ) + + +@dataclass +class _JudgeWorkOutcome: + """One indexed worker result awaiting coordinator delivery.""" + + index: int + verdict: IntentVerdict | None + fallback_reason: str + acknowledged: threading.Event = field(default_factory=threading.Event) + + +class _JudgeBatchCancelEvent(threading.Event): + """Private batch abort composed with the caller-owned cancel event.""" + + def __init__(self, upstream: threading.Event | None) -> None: + super().__init__() + self._upstream = upstream + + def is_set(self) -> bool: + return super().is_set() or bool(self._upstream and self._upstream.is_set()) + + def wait(self, timeout: float | None = None) -> bool: + """Wait for either the private abort or the upstream event.""" + deadline = None if timeout is None else time.monotonic() + timeout + while not self.is_set(): + remaining = None if deadline is None else deadline - time.monotonic() + if remaining is not None and remaining <= 0: + return False + super().wait(0.05 if remaining is None else min(0.05, remaining)) + return True # --------------------------------------------------------------------------- @@ -1230,7 +1280,7 @@ class IntentJudge: checked_config_version=config_version_at_start, ) # The semantic lane is pinned for the judge object's lifetime. Intent - # evaluations still substitute a fresh client per daemon batch for + # evaluations still substitute a fresh client per daemon worker for # thread isolation; no provider/model/config facet is re-resolved. self._lane = binding.lane self._model = self._lane.model @@ -1245,12 +1295,13 @@ class IntentJudge: ) @staticmethod - def _fingerprint_config(config: JudgeConfig) -> tuple[str, float, float, bool]: + def _fingerprint_config(config: JudgeConfig) -> tuple[str, float, float, int, bool]: """Constructor-consumed behavior that requires a fresh judge object.""" return ( str(config.model or "").strip(), config.max_context_ratio, config.timeout, + config.parallel_evaluations, config.read_only_tools, ) @@ -1313,8 +1364,8 @@ class IntentJudge: close, and — only when ``cancel_on_approval`` is enabled — as soon as the approval gate resolves. done_callback: Invoked exactly once from the daemon's - ``finally`` when this generation finishes — normally, - cancelled, or by escape of a delivery error. ChatSession + ``finally`` when this generation finishes — normally or + cancelled. Callback errors are isolated per item. ChatSession uses it to retire the generation's cancel event from its live set (parallel task agents each spawn a generation; ``close()`` aborts whatever is still live). @@ -1374,7 +1425,7 @@ class IntentJudge: done_callback: Callable[[], None] | None = None, backend_auth_resolver: Callable[[str, ModelConfig | None], str | None] | None = None, ) -> None: - """Daemon thread: run LLM judge for each item and invoke callback. + """Daemon coordinator: run bounded LLM evaluations and invoke callback. ``cancel_event`` is an unconditional abort signal: once it fires, in-flight work stops and every remaining item is delivered as an @@ -1389,7 +1440,6 @@ class IntentJudge: no supersede, every evaluation runs to completion so all verdicts are delivered. """ - client: Any | None = None try: if cancel_event and cancel_event.is_set(): self._deliver_fallbacks( @@ -1400,6 +1450,9 @@ class IntentJudge: ) return + if not items: + return + # Preserve the caller-pinned principal on the batch lane, but let # model_turn resolve credentials only after alias admission. An # admission backlog can outlive a bearer token; carrying the @@ -1421,94 +1474,169 @@ class IntentJudge: ) return - client = self._create_client() - # One lane derivative for the whole batch: only the judge-owned - # fresh client differs from the immutable constructor binding. - # Every item and evidence turn therefore stays on one provider, - # model, capability, auth-config, and initiating-principal - # binding while credentials refresh per admitted attempt. - batch_lane = replace(batch_lane, client=client) - for idx, (item, h_verdict) in enumerate(zip(items, heuristic_verdicts, strict=True)): - if cancel_event and cancel_event.is_set(): - log.info("judge.cancelled", remaining=len(items) - idx) - self._deliver_fallbacks( - items[idx:], - heuristic_verdicts[idx:], - callback, - "judge cancelled before evaluating this call", - ) - return + width = min( + max(1, int(self._config.parallel_evaluations)), + _MAX_PARALLEL_EVALUATIONS, + len(items), + ) + alias_limit = batch_lane.admission.limit if batch_lane.admission is not None else 0 + if alias_limit > 0: + width = min(width, alias_limit) + log.info( + "judge.batch.start", + items=len(items), + parallel_evaluations=width, + configured_parallel_evaluations=self._config.parallel_evaluations, + model_alias=batch_lane.alias, + model_alias_limit=alias_limit, + ) + work: queue.Queue[int | None] = queue.Queue() + outcomes: queue.Queue[_JudgeWorkOutcome] = queue.Queue() + batch_cancel = _JudgeBatchCancelEvent(cancel_event) + abort_lock = threading.Lock() + abort_reason = "" + + for idx in range(len(items)): + work.put(idx) + for _ in range(width): + work.put(None) + + def _current_abort_reason() -> str: + if cancel_event is not None and cancel_event.is_set(): + return "judge cancelled before evaluating this call" + with abort_lock: + return abort_reason + + def _abort_batch(reason: str) -> bool: + nonlocal abort_reason + with abort_lock: + first = not abort_reason + if first: + abort_reason = reason + batch_cancel.set() + return first + + def _worker() -> None: + client: Any | None = None try: - llm_verdict = self._evaluate_single( - item, - messages, - cancel_event, - client, - lane=batch_lane, - ) - if llm_verdict: - log.info( - "judge.verdict.llm", - recommendation=llm_verdict.recommendation, - confidence=llm_verdict.confidence, - call_id=llm_verdict.call_id, - ) - callback(llm_verdict) - else: - fallback = IntentVerdict( - verdict_id=h_verdict.verdict_id, - call_id=h_verdict.call_id, - func_name=h_verdict.func_name, - func_args=h_verdict.func_args, - intent_summary=h_verdict.intent_summary, - risk_level=h_verdict.risk_level, - confidence=h_verdict.confidence, - recommendation=h_verdict.recommendation, - reasoning=h_verdict.reasoning + " (LLM judge did not return a verdict)", - evidence=h_verdict.evidence, - tier="llm_fallback", - judge_model=self._model, - latency_ms=h_verdict.latency_ms, - ) - log.info( - "judge.verdict.fallback", - recommendation=fallback.recommendation, - confidence=fallback.confidence, - call_id=fallback.call_id, - ) - callback(fallback) - # After delivering this item's verdict, check whether the - # abort signal fired while we were evaluating it. - if cancel_event and cancel_event.is_set(): - log.info("judge.cancelled.after_eval", call_id=item.get("call_id", "")) - self._deliver_fallbacks( - items[idx + 1 :], - heuristic_verdicts[idx + 1 :], - callback, - "judge cancelled before evaluating this call", - ) - return - except BackendAuthUnavailableError: - log.exception("Judge backend authentication failed") - self._deliver_fallbacks( - items[idx:], - heuristic_verdicts[idx:], - callback, - "judge backend authentication failed", - ) - return - except Exception: - log.exception( - "Judge evaluation failed for %s", - item.get("func_name", "?"), - ) - self._deliver_fallbacks([item], [h_verdict], callback, "judge evaluation error") - finally: + while True: + idx = work.get() + if idx is None: + work.task_done() + return + outcome: _JudgeWorkOutcome + try: + reason = _current_abort_reason() + if reason: + outcome = _JudgeWorkOutcome(idx, None, reason) + else: + if client is None: + try: + client = self._create_client() + except BaseException: # noqa: BLE001 - contain daemon failure + reason = "judge client initialization failed" + if _abort_batch(reason): + log.exception("Judge client initialization failed") + outcome = _JudgeWorkOutcome(idx, None, reason) + if client is not None: + worker_lane = replace(batch_lane, client=client) + try: + verdict = self._evaluate_single( + items[idx], + messages, + batch_cancel, + client, + lane=worker_lane, + ) + reason = _current_abort_reason() + outcome = _JudgeWorkOutcome( + idx, + verdict, + reason + or ( + "" + if verdict is not None + else "LLM judge did not return a verdict" + ), + ) + except BackendAuthUnavailableError: + reason = "judge backend authentication failed" + if _abort_batch(reason): + log.exception("Judge backend authentication failed") + outcome = _JudgeWorkOutcome(idx, None, reason) + except BaseException: # noqa: BLE001 - contain daemon failure + log.exception( + "Judge evaluation failed for %s", + items[idx].get("func_name", "?"), + ) + outcome = _JudgeWorkOutcome( + idx, + None, + "judge evaluation error", + ) + finally: + work.task_done() + outcomes.put(outcome) + # Callback delivery is the cancellation commit point. + # Do not refill this worker until the coordinator has + # invoked the callback and observed any resulting abort. + outcome.acknowledged.wait() + finally: + try: + if client is not None and hasattr(client, "close"): + client.close() + except Exception: + log.debug("judge.client_close_failed", exc_info=True) + + workers: list[threading.Thread] = [] try: - if client is not None and hasattr(client, "close"): - client.close() - except Exception: - log.debug("judge.client_close_failed", exc_info=True) + for idx in range(width): + worker = threading.Thread( + target=_worker, + daemon=True, + name=f"intent-judge-eval-{idx + 1}", + ) + worker.start() + workers.append(worker) + except BaseException: # noqa: BLE001 - preserve exact-once fallbacks + reason = "judge worker initialization failed" + _abort_batch(reason) + log.exception("Judge worker initialization failed") + if not workers: + self._deliver_fallbacks(items, heuristic_verdicts, callback, abort_reason) + return + + delivered: set[int] = set() + for _ in range(len(items)): + outcome = outcomes.get() + try: + if outcome.index in delivered: + log.error("judge.verdict.duplicate", index=outcome.index) + continue + delivered.add(outcome.index) + h_verdict = heuristic_verdicts[outcome.index] + verdict = outcome.verdict or self._fallback_verdict( + h_verdict, + outcome.fallback_reason, + ) + log.info( + "judge.verdict.llm" + if outcome.verdict is not None + else "judge.verdict.fallback", + recommendation=verdict.recommendation, + confidence=verdict.confidence, + call_id=verdict.call_id, + ) + try: + callback(verdict) + except Exception: + log.debug("judge.verdict_delivery_failed", exc_info=True) + finally: + outcome.acknowledged.set() + + for worker in workers: + worker.join() + finally: if done_callback is not None: try: done_callback() @@ -1524,22 +1652,28 @@ class IntentJudge: ) -> None: """Deliver ``llm_fallback`` verdicts (heuristic content) for items the judge didn't complete.""" for _item, h_verdict in zip(remaining_items, remaining_verdicts, strict=True): - fallback = IntentVerdict( - verdict_id=h_verdict.verdict_id, - call_id=h_verdict.call_id, - func_name=h_verdict.func_name, - func_args=h_verdict.func_args, - intent_summary=h_verdict.intent_summary, - risk_level=h_verdict.risk_level, - confidence=h_verdict.confidence, - recommendation=h_verdict.recommendation, - reasoning=h_verdict.reasoning + f" ({reason})", - evidence=h_verdict.evidence, - tier="llm_fallback", - judge_model=self._model, - latency_ms=h_verdict.latency_ms, - ) - callback(fallback) + try: + callback(self._fallback_verdict(h_verdict, reason)) + except Exception: + log.debug("judge.verdict_delivery_failed", exc_info=True) + + def _fallback_verdict(self, h_verdict: IntentVerdict, reason: str) -> IntentVerdict: + """Relabel one heuristic verdict as an LLM fallback.""" + return IntentVerdict( + verdict_id=h_verdict.verdict_id, + call_id=h_verdict.call_id, + func_name=h_verdict.func_name, + func_args=h_verdict.func_args, + intent_summary=h_verdict.intent_summary, + risk_level=h_verdict.risk_level, + confidence=h_verdict.confidence, + recommendation=h_verdict.recommendation, + reasoning=h_verdict.reasoning + f" ({reason})", + evidence=h_verdict.evidence, + tier="llm_fallback", + judge_model=self._model, + latency_ms=h_verdict.latency_ms, + ) def _evaluate_single( self, @@ -1590,8 +1724,8 @@ class IntentJudge: if self._config.read_only_tools: tools = list(_JUDGE_TOOL_SCHEMAS) - # ``lane`` is the constructor-pinned binding with only the fresh - # batch client substituted. No registry/config facet is re-resolved + # ``lane`` is the constructor-pinned binding with only the worker's + # fresh client substituted. No registry/config facet is re-resolved # inside an evaluation; window sizing and wire capabilities therefore # cannot disagree. @@ -1915,12 +2049,15 @@ class IntentJudge: ] # Paths the judge is never allowed to read (security hardening). - _BLOCKED_PREFIXES: tuple[str, ...] = ( - "/etc/", - "/root/", - "/proc/", - "/sys/", - "/dev/", + _BLOCKED_ROOTS: tuple[Path, ...] = tuple( + Path(root).resolve() + for root in ( + "/etc", + "/root", + "/proc", + "/sys", + "/dev", + ) ) _BLOCKED_PARTS: frozenset[str] = frozenset( { @@ -1933,15 +2070,19 @@ class IntentJudge: _BLOCKED_SUFFIXES: tuple[str, ...] = (".pem", ".key", ".p12", ".pfx") @staticmethod - def _is_path_blocked(path: Path) -> bool: - """Return True if *path* should not be readable by the judge.""" - resolved = str(path.resolve()) - if any(resolved.startswith(p) for p in IntentJudge._BLOCKED_PREFIXES): + def _is_resolved_path_blocked(path: Path) -> bool: + """Return whether an already-resolved path is protected.""" + if any(path == root or root in path.parents for root in IntentJudge._BLOCKED_ROOTS): return True if IntentJudge._BLOCKED_PARTS & set(path.parts): return True return path.suffix.lower() in IntentJudge._BLOCKED_SUFFIXES + @staticmethod + def _is_path_blocked(path: Path) -> bool: + """Return True if *path* resolves to a protected location.""" + return IntentJudge._is_resolved_path_blocked(path.resolve()) + @staticmethod def _exec_read_only_tool(name: str, args: dict[str, Any]) -> str: """Execute a read-only tool directly (no session pipeline). @@ -1951,27 +2092,46 @@ class IntentJudge: try: if name == "read_file": path = Path(str(args.get("path", ""))) - if IntentJudge._is_path_blocked(path): + resolved_path = path.resolve() + if IntentJudge._is_resolved_path_blocked(resolved_path): return f"Error: access denied: {path}" - if not path.is_file(): + if not resolved_path.is_file(): return f"Error: file not found: {path}" - content = path.read_text(encoding="utf-8", errors="replace") - # Cap at 32KB to avoid blowing context - if len(content) > 32768: - return content[:32768] + f"\n... (truncated, {len(content)} bytes total)" + # Bound acquisition itself, not merely the returned string: + # a parallel batch must not materialize one unbounded file per + # worker before applying the judge context cap. + with resolved_path.open("r", encoding="utf-8", errors="replace") as handle: + content = handle.read(_JUDGE_READ_LIMIT + 1) + if len(content) > _JUDGE_READ_LIMIT: + try: + total_bytes = resolved_path.stat().st_size + size_note = f", {total_bytes} bytes total" + except OSError: + size_note = "" + return content[:_JUDGE_READ_LIMIT] + f"\n... (truncated{size_note})" return content if name == "list_directory": path = Path(str(args.get("path", ""))) - if IntentJudge._is_path_blocked(path): + resolved_path = path.resolve() + if IntentJudge._is_resolved_path_blocked(resolved_path): return f"Error: access denied: {path}" - if not path.is_dir(): + if not resolved_path.is_dir(): return f"Error: directory not found: {path}" - entries = sorted(path.iterdir())[:200] # cap at 200 entries + # Bound collection before sorting so concurrent evidence + # workers retain at most N+1 directory entries each. + entries = sorted( + itertools.islice(resolved_path.iterdir(), _JUDGE_DIRECTORY_LIMIT + 1), + key=lambda entry: entry.name, + ) + truncated = len(entries) > _JUDGE_DIRECTORY_LIMIT + entries = entries[:_JUDGE_DIRECTORY_LIMIT] lines: list[str] = [] for entry in entries: suffix = "/" if entry.is_dir() else "" lines.append(f" {entry.name}{suffix}") + if truncated: + lines.append(" ... (additional entries omitted)") return "\n".join(lines) or "(empty directory)" return f"Error: unknown tool: {name}" diff --git a/turnstone/core/session.py b/turnstone/core/session.py index c952df54..69d3c387 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -2930,6 +2930,12 @@ class ChatSession: raise RuntimeError("judge config composition requires a base config") from turnstone.core.judge import JudgeConfig + parallel_evaluations = setting("judge.parallel_evaluations") + if parallel_evaluations is None: + # Backward compatibility for duck-typed stores that predate this + # registry key. Real ConfigStore snapshots always carry defaults. + parallel_evaluations = 1 + return JudgeConfig( enabled=setting("judge.enabled"), model=jc.model, @@ -2937,6 +2943,7 @@ class ChatSession: confidence_threshold=setting("judge.confidence_threshold"), max_context_ratio=setting("judge.max_context_ratio"), timeout=setting("judge.timeout"), + parallel_evaluations=parallel_evaluations, read_only_tools=setting("judge.read_only_tools"), output_guard=setting("judge.output_guard"), output_guard_budget_seconds=setting("judge.output_guard_budget_seconds"), diff --git a/turnstone/core/settings_registry.py b/turnstone/core/settings_registry.py index 7524e0b7..6ba9585b 100644 --- a/turnstone/core/settings_registry.py +++ b/turnstone/core/settings_registry.py @@ -29,6 +29,7 @@ class SettingDef: restart_required: bool = False help: str = "" # plain-English explanation for non-experts reference_url: str = "" # link to arXiv, docs, or provider reference + strict_int: bool = False # reject bool/float/non-canonical strings before int coercion # Default auto-compaction trigger as a fraction of the context window. Shared @@ -647,6 +648,20 @@ def _build_registry() -> dict[str, SettingDef]: "judge", min_value=5.0, ), + SettingDef( + "judge.parallel_evaluations", + "int", + 1, + "Maximum concurrent LLM evaluations per judge batch", + "judge", + min_value=1, + max_value=16, + strict_int=True, + help="How many independent tool calls from one approval batch the intent judge " + "evaluates at once. Keep this at 1 for serial evaluation, or raise it to reduce " + "latency for wide batches. The judge model alias's Max concurrent generations " + "setting remains the process-wide ceiling and may reduce the actual overlap.", + ), SettingDef( "judge.read_only_tools", "bool", @@ -950,7 +965,21 @@ def validate_value(key: str, raw_value: Any) -> Any: # Type coercion try: if defn.type == "int": - typed: Any = int(raw_value) + if defn.strict_int: + if type(raw_value) is int: + typed: Any = raw_value + elif isinstance(raw_value, str): + typed = int(raw_value) + if str(typed) != raw_value: + raise ValueError( + f"Expected a canonical integer string for {key}, got {raw_value!r}" + ) + else: + raise TypeError( + f"Expected an integer for {key}, got {type(raw_value).__name__}" + ) + else: + typed = int(raw_value) elif defn.type == "float": typed = float(raw_value) elif defn.type == "bool": diff --git a/turnstone/server.py b/turnstone/server.py index b3a60e77..1c73a982 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -6005,6 +6005,7 @@ def main() -> None: confidence_threshold=config_store.get("judge.confidence_threshold"), max_context_ratio=config_store.get("judge.max_context_ratio"), timeout=config_store.get("judge.timeout"), + parallel_evaluations=config_store.get("judge.parallel_evaluations", 1), read_only_tools=config_store.get("judge.read_only_tools"), output_guard=config_store.get("judge.output_guard"), output_guard_budget_seconds=config_store.get("judge.output_guard_budget_seconds"),