mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
feat(judge): parallelize batch evaluations (#991)
This commit is contained in:
Patrick Buckley
@@ -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
|
||||
{
|
||||
|
||||
+15
-10
@@ -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).
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
version https://git-lfs.github.com/spec/v1
|
||||
oid sha256:dde4f417956534a70b0e37b24cfe9de383897780669c9da81833c8084d572dfe
|
||||
size 269928
|
||||
oid sha256:636a6b2fc1075e4863421e68b99efe7f6f6f62cedbcff36ef0934c055f39fd46
|
||||
size 281161
|
||||
|
||||
+60
-47
@@ -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)
|
||||
|
||||
|
||||
+21
-8
@@ -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.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -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"}
|
||||
|
||||
+30
-2
@@ -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):
|
||||
|
||||
@@ -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
|
||||
|
||||
+920
-4
@@ -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
|
||||
|
||||
@@ -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
|
||||
+17
-3
@@ -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):
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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) ---
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"),
|
||||
|
||||
@@ -3859,7 +3859,7 @@ function renderJudgeSettings() {
|
||||
">" +
|
||||
'<span class="toggle-track" aria-hidden="true"></span>' +
|
||||
'<span class="toggle-label">Enabled</span></label>';
|
||||
} 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 =
|
||||
'<div style="display:flex;gap:8px;align-items:center">' +
|
||||
'<input type="number" step="0.01" data-judge-key="' +
|
||||
'<input type="number" step="' +
|
||||
(s.type === "int" ? "1" : "0.01") +
|
||||
'" data-judge-key="' +
|
||||
eKey +
|
||||
'" value="' +
|
||||
escapeHtml(String(currentVal != null ? currentVal : "")) +
|
||||
|
||||
@@ -173,6 +173,7 @@ _CONFIG_MAP: dict[str, dict[str, str]] = {
|
||||
"confidence_threshold": "judge_confidence",
|
||||
"max_context_ratio": "judge_context_ratio",
|
||||
"timeout": "judge_timeout",
|
||||
"parallel_evaluations": "judge_parallel_evaluations",
|
||||
"read_only_tools": "judge_read_only_tools",
|
||||
},
|
||||
"memory": {
|
||||
|
||||
+289
-129
@@ -8,9 +8,11 @@ is a fast, pure-function rule engine with zero external dependencies.
|
||||
from __future__ import annotations
|
||||
|
||||
import fnmatch
|
||||
import itertools
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import queue
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
@@ -47,6 +49,10 @@ if TYPE_CHECKING:
|
||||
|
||||
log = get_logger(__name__)
|
||||
|
||||
_MAX_PARALLEL_EVALUATIONS = 16
|
||||
_JUDGE_READ_LIMIT = 32_768
|
||||
_JUDGE_DIRECTORY_LIMIT = 200
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Data structures
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -118,6 +124,50 @@ class JudgeConfig:
|
||||
# False (default) = the daemon runs every item to completion; only a
|
||||
# generation supersede (next batch) or session close aborts it.
|
||||
cancel_on_approval: bool = False
|
||||
# Maximum tool-call evaluations this judge may run concurrently within
|
||||
# one approval batch. The selected model alias's admission limit remains
|
||||
# the process-wide ceiling across judge and non-judge traffic.
|
||||
parallel_evaluations: int = 1
|
||||
|
||||
def __post_init__(self) -> 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}"
|
||||
|
||||
@@ -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"),
|
||||
|
||||
@@ -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":
|
||||
|
||||
@@ -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"),
|
||||
|
||||
Reference in New Issue
Block a user