From 3690373c3c6a182a2647bd3b894a648bb6fc66bd Mon Sep 17 00:00:00 2001 From: Joshua Jeon Date: Mon, 5 Oct 2026 16:53:00 -0700 Subject: [PATCH 1/3] feat: capture per-control execution duration [SAO-18023] --- engine/src/agent_control_engine/core.py | 9 + engine/tests/test_execution_duration.py | 254 ++++++++++++++++++ models/README.md | 14 + models/src/agent_control_models/controls.py | 5 + models/tests/test_control_match_duration.py | 58 ++++ .../src/agent_control/control_decorators.py | 5 +- .../src/agent_control/evaluation_events.py | 1 + sdks/python/tests/test_execution_duration.py | 197 ++++++++++++++ .../src/generated/models/control-match.ts | 6 + sdks/typescript/tests/client-api.test.ts | 41 +++ .../tests/test_execution_duration_response.py | 131 +++++++++ 11 files changed, 720 insertions(+), 1 deletion(-) create mode 100644 engine/tests/test_execution_duration.py create mode 100644 models/tests/test_control_match_duration.py create mode 100644 sdks/python/tests/test_execution_duration.py create mode 100644 server/tests/test_execution_duration_response.py diff --git a/engine/src/agent_control_engine/core.py b/engine/src/agent_control_engine/core.py index ab170e822..58e270eb1 100644 --- a/engine/src/agent_control_engine/core.py +++ b/engine/src/agent_control_engine/core.py @@ -7,6 +7,7 @@ import functools import logging import os +import time from collections.abc import Sequence from dataclasses import dataclass from typing import Any, Literal, Protocol @@ -201,6 +202,7 @@ class _EvalTask: item: ControlWithIdentity task: asyncio.Task[None] | None = None result: EvaluatorResult | None = None + execution_duration_ms: float | None = None @dataclass @@ -673,6 +675,7 @@ async def process( async def evaluate_control(eval_task: _EvalTask) -> None: """Evaluate a single control, respecting cancellation and timeout.""" + started_at = time.perf_counter() try: evaluation = await self._evaluate_condition( eval_task.item, @@ -701,6 +704,9 @@ async def evaluate_control(eval_task: _EvalTask) -> None: error_msg, message_prefix="Condition evaluation failed", ) + finally: + if eval_task.result is not None: + eval_task.execution_duration_ms = (time.perf_counter() - started_at) * 1000 # Create and start all tasks for eval_task in eval_tasks: @@ -753,6 +759,7 @@ async def wait_for_deny() -> None: control_name=eval_task.item.name, action=eval_task.item.control.action.decision, result=eval_task.result, + execution_duration_ms=eval_task.execution_duration_ms, steering_context=eval_task.item.control.action.steering_context, ) ) @@ -776,6 +783,7 @@ async def wait_for_deny() -> None: control_name=eval_task.item.name, action=eval_task.item.control.action.decision, result=eval_task.result, + execution_duration_ms=eval_task.execution_duration_ms, steering_context=steer_ctx, ) ) @@ -792,6 +800,7 @@ async def wait_for_deny() -> None: control_name=eval_task.item.name, action=eval_task.item.control.action.decision, result=eval_task.result, + execution_duration_ms=eval_task.execution_duration_ms, steering_context=eval_task.item.control.action.steering_context, ) ) diff --git a/engine/tests/test_execution_duration.py b/engine/tests/test_execution_duration.py new file mode 100644 index 000000000..b52c154ba --- /dev/null +++ b/engine/tests/test_execution_duration.py @@ -0,0 +1,254 @@ +"""Public engine contract tests for per-control execution durations.""" + +import asyncio +from dataclasses import dataclass +from typing import Any, Literal +from unittest.mock import AsyncMock, Mock + +import pytest +from agent_control_engine.core import ControlEngine +from agent_control_evaluators import Evaluator, EvaluatorMetadata, register_evaluator +from agent_control_models import ControlDefinition, EvaluationRequest, EvaluatorResult, Step +from pydantic import BaseModel + + +class DurationConfig(BaseModel): + matched: bool = False + raises_error: bool = False + blocks: bool = False + + +class DurationEvaluator(Evaluator[DurationConfig]): + metadata = EvaluatorMetadata( + name="test-duration", + version="1.0.0", + description="Returns a configured outcome or waits for cancellation", + ) + config_model = DurationConfig + + async def evaluate(self, data: Any) -> EvaluatorResult: + # Yield once so concurrent controls all start before one completes. + await asyncio.sleep(0) + if self.config.blocks: + await asyncio.Event().wait() + if self.config.raises_error: + raise RuntimeError("evaluation failed") + return EvaluatorResult(matched=self.config.matched, confidence=1.0) + + +@dataclass +class DurationControl: + id: int + name: str + control: ControlDefinition + + +@pytest.fixture(autouse=True) +def register_duration_evaluator() -> None: + register_evaluator(DurationEvaluator) + + +@pytest.fixture +def evaluation_request() -> EvaluationRequest: + return EvaluationRequest( + agent_name="duration-test-agent", + step=Step(type="llm", name="test-step", input="test", output=None), + stage="pre", + ) + + +def make_control( + control_id: int = 1, + *, + context: Literal["server", "sdk"] = "server", + matched: bool = False, + raises_error: bool = False, + blocks: bool = False, + decision: Literal["observe", "deny"] = "observe", +) -> DurationControl: + return DurationControl( + id=control_id, + name=f"duration-control-{control_id}", + control=ControlDefinition( + description="Duration test control", + enabled=True, + execution=context, + scope={"stages": ["pre"], "step_types": ["llm"]}, + condition={ + "selector": {"path": "input"}, + "evaluator": { + "name": "test-duration", + "config": { + "matched": matched, + "raises_error": raises_error, + "blocks": blocks, + }, + }, + }, + action={"decision": decision}, + ), + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("context", ["server", "sdk"]) +@pytest.mark.parametrize("category", ["matches", "non_matches", "errors"]) +@pytest.mark.parametrize("elapsed_seconds", [0.0, 0.0125]) +async def test_completed_controls_report_elapsed_milliseconds( + monkeypatch: pytest.MonkeyPatch, + evaluation_request: EvaluationRequest, + context: Literal["server", "sdk"], + category: str, + elapsed_seconds: float, +) -> None: + # Given: a completed control and a controlled clock instead of wall-clock assertions + control = make_control( + context=context, + matched=category == "matches", + raises_error=category == "errors", + ) + clock = Mock(side_effect=[10.0, 10.0 + elapsed_seconds]) + monkeypatch.setattr("agent_control_engine.core.time.perf_counter", clock) + + # When: evaluating via the public engine entrypoint + response = await ControlEngine([control], context=context).process(evaluation_request) + + # Then: all completed outcome categories retain their own duration, including zero + results = getattr(response, category) + assert results is not None + assert len(results) == 1 + assert results[0].execution_duration_ms == pytest.approx(elapsed_seconds * 1000) + assert results[0].result.matched is (category == "matches") + assert bool(results[0].result.error) is (category == "errors") + assert response.is_safe is True + + +@pytest.mark.asyncio +async def test_parallel_controls_keep_individual_durations( + monkeypatch: pytest.MonkeyPatch, evaluation_request: EvaluationRequest +) -> None: + # Given: parallel controls start together and complete at different clock readings + clock = Mock(side_effect=[10.0, 10.0, 10.0125, 10.025]) + monkeypatch.setattr("agent_control_engine.core.time.perf_counter", clock) + controls = [make_control(1), make_control(2)] + + # When: evaluating both controls in the same engine request + response = await ControlEngine(controls).process(evaluation_request) + + # Then: each control reports its own elapsed time rather than the batch duration + assert response.non_matches is not None + durations = {result.control_id: result.execution_duration_ms for result in response.non_matches} + assert durations == pytest.approx({1: 12.5, 2: 25.0}) + + +@pytest.mark.asyncio +async def test_condition_traversal_error_retains_elapsed_duration( + monkeypatch: pytest.MonkeyPatch, evaluation_request: EvaluationRequest +) -> None: + # Given: an internal traversal failure forced to exercise the outer error handler + engine = ControlEngine([make_control(decision="deny")]) + monkeypatch.setattr( + engine, "_evaluate_condition", AsyncMock(side_effect=ValueError("condition failed")) + ) + clock = Mock(side_effect=[10.0, 10.007]) + monkeypatch.setattr("agent_control_engine.core.time.perf_counter", clock) + + # When: evaluating via the public engine entrypoint + response = await engine.process(evaluation_request) + + # Then: the completed error is timed and still fails closed + assert response.is_safe is False + assert response.errors is not None + assert len(response.errors) == 1 + assert response.errors[0].execution_duration_ms == pytest.approx(7.0) + assert response.errors[0].result.error == "ValueError: condition failed" + + +@pytest.mark.asyncio +async def test_timed_out_evaluation_retains_elapsed_duration( + monkeypatch: pytest.MonkeyPatch, evaluation_request: EvaluationRequest +) -> None: + # Given: a blocking evaluator with a bounded timeout and a controlled elapsed clock + monkeypatch.setattr(DurationEvaluator, "get_timeout_seconds", lambda self: 0.001) + clock = Mock(side_effect=[10.0, 10.001]) + monkeypatch.setattr("agent_control_engine.core.time.perf_counter", clock) + + # When: the evaluator times out without producing a normal result + response = await ControlEngine([make_control(blocks=True, decision="deny")]).process( + evaluation_request + ) + + # Then: the timeout error is a completed evaluation with an elapsed duration + assert response.is_safe is False + assert response.errors is not None + assert len(response.errors) == 1 + assert response.errors[0].execution_duration_ms == pytest.approx(1.0) + assert response.errors[0].result.error is not None + assert response.errors[0].result.error.startswith("TimeoutError:") + + +@pytest.mark.asyncio +async def test_cancelled_controls_do_not_report_partial_durations( + monkeypatch: pytest.MonkeyPatch, evaluation_request: EvaluationRequest +) -> None: + # Given: a blocking control that is cancelled by a completed deny control + clock = Mock(side_effect=[10.0, 10.0, 10.020]) + monkeypatch.setattr("agent_control_engine.core.time.perf_counter", clock) + controls = [make_control(1, blocks=True), make_control(2, matched=True, decision="deny")] + + # When: a deny match cancels the unfinished evaluation + response = await ControlEngine(controls).process(evaluation_request) + + # Then: only the completed deny is returned, with no fabricated cancelled result + assert response.is_safe is False + assert response.matches is not None + assert len(response.matches) == 1 + assert response.matches[0].control_id == 2 + assert response.matches[0].execution_duration_ms == pytest.approx(20.0) + assert response.non_matches is None + assert response.errors is None + assert clock.call_count == 3 + + +@pytest.mark.asyncio +async def test_skipped_control_does_not_start_duration_measurement( + monkeypatch: pytest.MonkeyPatch, evaluation_request: EvaluationRequest +) -> None: + # Given: a disabled control that must not evaluate + control = make_control() + control.control.enabled = False + clock = Mock(side_effect=AssertionError("skipped control must not be timed")) + monkeypatch.setattr("agent_control_engine.core.time.perf_counter", clock) + + # When: processing a request without applicable controls + response = await ControlEngine([control]).process(evaluation_request) + + # Then: there are no results or clock reads for the skipped control + assert response.matches is None + assert response.non_matches is None + assert response.errors is None + clock.assert_not_called() + + +@pytest.mark.asyncio +async def test_precheck_error_has_no_execution_duration( + monkeypatch: pytest.MonkeyPatch, evaluation_request: EvaluationRequest +) -> None: + # Given: validation is bypassed to exercise the defensive invalid-regex precheck + control = make_control() + payload = control.control.model_dump() + payload["scope"]["step_name_regex"] = "[invalid(regex" + control.control = ControlDefinition.model_validate( + payload, context={"allow_invalid_step_name_regex": True} + ) + clock = Mock(side_effect=AssertionError("precheck failure must not be timed")) + monkeypatch.setattr("agent_control_engine.core.time.perf_counter", clock) + + # When: processing a control that fails before evaluation + response = await ControlEngine([control]).process(evaluation_request) + + # Then: the error retains an unknown duration rather than zero + assert response.errors is not None + assert len(response.errors) == 1 + assert response.errors[0].execution_duration_ms is None + clock.assert_not_called() diff --git a/models/README.md b/models/README.md index db98868ae..fb09f055f 100644 --- a/models/README.md +++ b/models/README.md @@ -17,6 +17,20 @@ agent = Agent(agent_name="support-bot", agent_description="Support agent") step = Step(type="llm", name="chat", input="hello") ``` +## Control execution duration + +Evaluation responses include an optional top-level `execution_duration_ms` on each +`ControlMatch` in `matches`, `non_matches`, and `errors`. It measures that control's +condition evaluation in wall-clock milliseconds, including the visited condition +leaves, evaluator work, and time waiting for an evaluator concurrency slot. +Control loading and request/response overhead are outside this measurement. + +Completed evaluations include a nonnegative value, including evaluator failures. +Precheck failures have no duration, and canceled evaluations produce no result. +Older responses without the field remain valid. The Python SDK copies the value +to `ControlExecutionEvent.execution_duration_ms`; the TypeScript SDK exposes it as +`executionDurationMs`. + ## Tests ```bash diff --git a/models/src/agent_control_models/controls.py b/models/src/agent_control_models/controls.py index 1e2bb9e9c..46e8fd8ec 100644 --- a/models/src/agent_control_models/controls.py +++ b/models/src/agent_control_models/controls.py @@ -957,6 +957,11 @@ class ControlMatch(BaseModel): result: EvaluatorResult = Field( ..., description="Evaluator result (confidence, message, metadata)" ) + execution_duration_ms: float | None = Field( + default=None, + ge=0, + description="Control evaluation wall-clock duration in milliseconds, if executed", + ) steering_context: SteeringContext | None = Field( None, description="Steering context for steer actions if configured" diff --git a/models/tests/test_control_match_duration.py b/models/tests/test_control_match_duration.py new file mode 100644 index 000000000..6b3e13f73 --- /dev/null +++ b/models/tests/test_control_match_duration.py @@ -0,0 +1,58 @@ +"""Control execution duration response compatibility.""" + +import pytest +from agent_control_models.controls import ControlMatch +from pydantic import ValidationError + + +@pytest.mark.parametrize("duration_ms", [None, 0.0, 12.5]) +def test_control_match_preserves_optional_execution_duration(duration_ms: float | None) -> None: + # Given: a response with a recorded duration or an unevaluated control. + payload = { + "control_id": 1, + "control_name": "Example control", + "action": "deny", + "result": {"matched": False, "confidence": 0.0}, + "execution_duration_ms": duration_ms, + } + + # When: parsing and serializing the response model. + match = ControlMatch.model_validate(payload) + reconstructed = ControlMatch.model_validate_json(match.model_dump_json()) + + # Then: the duration stays top-level and zero is preserved. + assert reconstructed.execution_duration_ms == duration_ms + assert match.model_dump()["execution_duration_ms"] == duration_ms + assert "execution_duration_ms" not in match.result.model_dump() + + +def test_control_match_accepts_legacy_response_without_duration() -> None: + # Given: an older server response with no duration field. + payload = { + "control_id": 1, + "control_name": "Example control", + "action": "deny", + "result": {"matched": False, "confidence": 0.0}, + } + + # When: parsing the response model. + match = ControlMatch.model_validate(payload) + + # Then: no duration is fabricated for the older response. + assert match.execution_duration_ms is None + + +def test_control_match_rejects_negative_execution_duration() -> None: + # Given: a response containing an invalid elapsed time. + payload = { + "control_id": 1, + "control_name": "Example control", + "action": "deny", + "result": {"matched": False, "confidence": 0.0}, + "execution_duration_ms": -1, + } + + # When: parsing the response model. + with pytest.raises(ValidationError, match="greater than or equal to 0"): + ControlMatch.model_validate(payload) + # Then: the response follows the same nonnegative duration contract as events. diff --git a/sdks/python/src/agent_control/control_decorators.py b/sdks/python/src/agent_control/control_decorators.py index fee78333d..38249e98f 100644 --- a/sdks/python/src/agent_control/control_decorators.py +++ b/sdks/python/src/agent_control/control_decorators.py @@ -339,6 +339,7 @@ async def _evaluate( "metadata": m.result.metadata, }, "control_execution_id": m.control_execution_id, + "execution_duration_ms": m.execution_duration_ms, "steering_context": ( {"message": m.steering_context.message} if m.steering_context else None @@ -360,6 +361,7 @@ async def _evaluate( "metadata": e.result.metadata, }, "control_execution_id": e.control_execution_id, + "execution_duration_ms": e.execution_duration_ms, "steering_context": ( {"message": e.steering_context.message} if e.steering_context else None @@ -381,6 +383,7 @@ async def _evaluate( "metadata": nm.result.metadata, }, "control_execution_id": nm.control_execution_id, + "execution_duration_ms": nm.execution_duration_ms, "steering_context": ( {"message": nm.steering_context.message} if nm.steering_context else None @@ -707,7 +710,7 @@ def _log_single_control( matched=matched, action=normalize_action(str(control_data.get("action", "observe"))), confidence=control_data.get("result", {}).get("confidence", 0.0), - duration_ms=control_data.get("result", {}).get("execution_duration_ms"), + duration_ms=control_data.get("execution_duration_ms"), control_execution_id=control_data.get("control_execution_id"), check_stage=check_stage, ) diff --git a/sdks/python/src/agent_control/evaluation_events.py b/sdks/python/src/agent_control/evaluation_events.py index 46c9ac153..ac3f48b28 100644 --- a/sdks/python/src/agent_control/evaluation_events.py +++ b/sdks/python/src/agent_control/evaluation_events.py @@ -141,6 +141,7 @@ def _build_events_for_matches( action=match.action, matched=matched, confidence=match.result.confidence, + execution_duration_ms=match.execution_duration_ms, timestamp=now, evaluator_name=evaluator_name, selector_path=selector_path, diff --git a/sdks/python/tests/test_execution_duration.py b/sdks/python/tests/test_execution_duration.py new file mode 100644 index 000000000..54426d767 --- /dev/null +++ b/sdks/python/tests/test_execution_duration.py @@ -0,0 +1,197 @@ +"""Per-control duration survives SDK response parsing and event reconstruction.""" + +from collections.abc import Iterator, Sequence +from functools import partial +from typing import Any, Literal +from unittest.mock import MagicMock, patch + +import httpx +import pytest +from agent_control_models import ControlExecutionEvent, Step +from agent_control_telemetry import REGISTERED_CONTROL_EVENT_SINK_NAME +from agent_control_telemetry.sinks import BaseControlEventSink, SinkResult + +from agent_control import ( + AgentControlClient, + control, + register_control_event_sink, + unregister_control_event_sink, +) +from agent_control._state import state +from agent_control.evaluation import check_evaluation, check_evaluation_with_local +from agent_control.otel_sink import control_event_to_otel_span +from agent_control.settings import configure_settings, get_settings + +ResultCategory = Literal["matches", "errors", "non_matches"] + + +class RecordingSink(BaseControlEventSink): + """Capture the events delivered through the public sink API.""" + + def __init__(self) -> None: + self.events: list[ControlExecutionEvent] = [] + + def write_events(self, events: Sequence[ControlExecutionEvent]) -> SinkResult: + self.events.extend(events) + return SinkResult(accepted=len(events), dropped=0) + + +@pytest.fixture +def recording_sink(monkeypatch: pytest.MonkeyPatch) -> Iterator[RecordingSink]: + # Given: an isolated SDK session exporting to a registered event sink. + previous_settings = get_settings().model_dump() + for name in ("current_agent", "target_type", "target_id", "server_controls"): + monkeypatch.setattr(state, name, None) + sink = RecordingSink() + register_control_event_sink(sink) + configure_settings( + observability_enabled=True, + observability_sink_name=REGISTERED_CONTROL_EVENT_SINK_NAME, + ) + try: + yield sink + finally: + unregister_control_event_sink(sink) + configure_settings(**previous_settings) + + +def _server_response(category: ResultCategory, duration: float | None) -> dict[str, Any]: + match: dict[str, Any] = { + "control_id": 2, + "control_name": "server-control", + "control_execution_id": "server-execution", + "action": "observe", + "result": { + "matched": category == "matches", + "confidence": 1.0, + "error": "Evaluator failed" if category == "errors" else None, + }, + } + if duration is not None: + match["execution_duration_ms"] = duration + return {"is_safe": category != "errors", "confidence": 1.0, category: [match]} + + +def _control_payload(control_id: int, execution: str) -> dict[str, Any]: + return { + "id": control_id, + "name": f"{execution}-control", + "control": { + "execution": execution, + "scope": {"stages": ["pre"]}, + "condition": { + "selector": {"path": "input"}, + "evaluator": {"name": "regex", "config": {"pattern": "hello"}}, + }, + "action": {"decision": "observe"}, + }, + } + + +@pytest.mark.parametrize("category", ["matches", "errors", "non_matches"]) +@pytest.mark.parametrize("duration", [12.5, 0.0, None]) +async def test_server_duration_reaches_events_and_otel( + recording_sink: RecordingSink, + category: ResultCategory, + duration: float | None, +) -> None: + # Given: a real SDK client receiving a timed or legacy server response. + transport = httpx.MockTransport( + lambda request: httpx.Response(200, json=_server_response(category, duration)) + ) + + # When: evaluating through the public server-only SDK helper. + async with AgentControlClient(base_url="https://server.test", transport=transport) as client: + result = await check_evaluation( + client, + agent_name="agent-000000000001", + step=Step(type="llm", name="chat", input="hello"), + stage="pre", + ) + + # Then: the typed result and exported event retain the exact per-control value. + matches = getattr(result, category) + assert matches is not None + assert matches[0].execution_duration_ms == duration + assert len(recording_sink.events) == 1 + event = recording_sink.events[0] + assert event.execution_duration_ms == duration + assert "execution_duration_ms" not in event.metadata + span = control_event_to_otel_span(event) + if duration is None: + assert "agent_control.execution_duration_ms" not in span.attributes + else: + assert span.attributes["agent_control.execution_duration_ms"] == duration + assert span.end_time_unix_nano - span.start_time_unix_nano == int(duration * 1_000_000) + + +async def test_mixed_local_and_server_events_keep_their_own_durations( + recording_sink: RecordingSink, +) -> None: + # Given: a real local regex control and a server control with a distinct duration. + transport = httpx.MockTransport( + lambda request: httpx.Response(200, json=_server_response("non_matches", 27.5)) + ) + controls = [_control_payload(1, "sdk"), _control_payload(2, "server")] + + # When: evaluating both controls through the public local-first SDK helper. + async with AgentControlClient(base_url="https://server.test", transport=transport) as client: + result = await check_evaluation_with_local( + client, + agent_name="agent-000000000001", + step=Step(type="llm", name="chat", input="hello"), + stage="pre", + controls=controls, + trace_id="a" * 32, + span_id="b" * 16, + ) + + # Then: local engine timing and server timing each reach the corresponding event. + assert result.matches is not None + assert result.non_matches is not None + local_duration = result.matches[0].execution_duration_ms + assert local_duration is not None + assert local_duration >= 0 + assert len(recording_sink.events) == 2 + durations = {event.control_id: event.execution_duration_ms for event in recording_sink.events} + assert durations == {1: local_duration, 2: 27.5} + + +@pytest.mark.parametrize("category", ["matches", "errors", "non_matches"]) +@pytest.mark.parametrize("duration", [12.5, 0.0, None]) +async def test_decorator_preserves_duration_in_events_and_debug_logs( + recording_sink: RecordingSink, + category: ResultCategory, + duration: float | None, +) -> None: + # Given: a decorated control site using cached controls and a mock HTTP server. + transport = httpx.MockTransport( + lambda request: httpx.Response(200, json=_server_response(category, duration)) + ) + agent = MagicMock(agent_name="agent-000000000001") + + @control() + async def chat(message: str) -> str: + return message + + # When: calling the public decorator through its real evaluation and reconstruction paths. + with ( + patch("agent_control.current_agent", return_value=agent), + patch("agent_control.get_server_controls", return_value=[_control_payload(2, "server")]), + patch( + "agent_control.control_decorators.AgentControlClient", + partial(AgentControlClient, transport=transport), + ), + patch("agent_control.control_decorators.log_control_evaluation") as log_evaluation, + ): + if category == "errors": + with pytest.raises(RuntimeError, match="Control evaluation failed on server"): + await chat("hello") + else: + assert await chat("hello") == "hello" + + # Then: reconstruction retains duration for every category, including zero and missing values. + assert len(recording_sink.events) == 1 + assert recording_sink.events[0].execution_duration_ms == duration + log_evaluation.assert_called_once() + assert log_evaluation.call_args.kwargs["duration_ms"] == duration diff --git a/sdks/typescript/src/generated/models/control-match.ts b/sdks/typescript/src/generated/models/control-match.ts index 6a7caa1f5..edf416abf 100644 --- a/sdks/typescript/src/generated/models/control-match.ts +++ b/sdks/typescript/src/generated/models/control-match.ts @@ -38,6 +38,10 @@ export type ControlMatch = { * Name of the control */ controlName: string; + /** + * Control evaluation wall-clock duration in milliseconds, if executed + */ + executionDurationMs?: number | null | undefined; /** * Result from a control evaluator. * @@ -69,6 +73,7 @@ export const ControlMatch$inboundSchema: z.ZodMiniType = control_execution_id: types.optional(types.string()), control_id: types.number(), control_name: types.string(), + execution_duration_ms: z.optional(z.nullable(types.number())), result: EvaluatorResult$inboundSchema, steering_context: z.optional(z.nullable(SteeringContext$inboundSchema)), }), @@ -77,6 +82,7 @@ export const ControlMatch$inboundSchema: z.ZodMiniType = "control_execution_id": "controlExecutionId", "control_id": "controlId", "control_name": "controlName", + "execution_duration_ms": "executionDurationMs", "steering_context": "steeringContext", }); }), diff --git a/sdks/typescript/tests/client-api.test.ts b/sdks/typescript/tests/client-api.test.ts index b0fa91cb6..5d1dc591c 100644 --- a/sdks/typescript/tests/client-api.test.ts +++ b/sdks/typescript/tests/client-api.test.ts @@ -21,6 +21,47 @@ describe("AgentControlClient API wiring", () => { vi.restoreAllMocks(); }); + it.each([12.5, 0, null, undefined])( + "preserves per-control execution durations through evaluation responses (%s)", + async (executionDurationMs) => { + const match = { + control_id: 1, + control_name: "Example control", + action: "observe", + result: { matched: true, confidence: 1.0 }, + execution_duration_ms: executionDurationMs, + }; + vi.mocked(globalThis.fetch).mockResolvedValueOnce( + jsonResponse({ + is_safe: true, + confidence: 1.0, + matches: [match], + non_matches: [ + { ...match, result: { matched: false, confidence: 1.0 } }, + ], + errors: [ + { + ...match, + result: { matched: false, confidence: 0.0, error: "Evaluation failed" }, + }, + ], + }), + ); + const client = new AgentControlClient(); + client.init({ agentName: "test-agent", serverUrl: "https://api.example.com" }); + + const response = await client.evaluation.evaluate({ + agentName: "test-agent", + stage: "pre", + step: { type: "llm", name: "answer", input: "hello" }, + }); + + expect(response.matches?.[0]?.executionDurationMs).toBe(executionDurationMs); + expect(response.nonMatches?.[0]?.executionDurationMs).toBe(executionDurationMs); + expect(response.errors?.[0]?.executionDurationMs).toBe(executionDurationMs); + }, + ); + it("throws when endpoint groups are accessed before init", () => { const client = new AgentControlClient(); diff --git a/server/tests/test_execution_duration_response.py b/server/tests/test_execution_duration_response.py new file mode 100644 index 000000000..dec932d69 --- /dev/null +++ b/server/tests/test_execution_duration_response.py @@ -0,0 +1,131 @@ +"""HTTP contract tests for per-control execution durations without database I/O.""" + +from unittest.mock import AsyncMock + +import pytest +from agent_control_models import ControlMatch, EvaluationResponse, EvaluatorResult +from fastapi.testclient import TestClient + +from agent_control_server.endpoints import evaluation +from agent_control_server.endpoints.evaluation import SAFE_EVALUATOR_ERROR + + +@pytest.fixture(scope="session", autouse=True) +def db_schema() -> None: + """Override database setup: control loading is mocked for this contract test.""" + + +@pytest.fixture(autouse=True) +def clean_db() -> None: + """Override database cleanup: these tests neither read nor write database rows.""" + + +@pytest.fixture +def engine_response(monkeypatch: pytest.MonkeyPatch) -> AsyncMock: + """Stub evaluation I/O while retaining the real HTTP response handling.""" + monkeypatch.setattr(evaluation, "_load_engine_controls", AsyncMock(return_value=[])) + process = AsyncMock() + monkeypatch.setattr(evaluation.ControlEngine, "process", process) + return process + + +def test_evaluation_response_preserves_duration_for_each_result_category( + client: TestClient, + engine_response: AsyncMock, +) -> None: + # Given: engine results with distinct per-control timings, including zero. + # Control loading and engine processing are stubbed to avoid database/evaluator I/O. + engine_response.return_value = EvaluationResponse( + is_safe=False, + confidence=0.9, + matches=[ + ControlMatch( + control_id=1, + control_name="matched-control", + action="deny", + execution_duration_ms=12.5, + result=EvaluatorResult(matched=True, confidence=0.9), + ) + ], + errors=[ + ControlMatch( + control_id=2, + control_name="failed-control", + action="observe", + execution_duration_ms=3.25, + result=EvaluatorResult( + matched=False, + confidence=0.0, + error="RuntimeError: private evaluator detail", + ), + ) + ], + non_matches=[ + ControlMatch( + control_id=3, + control_name="non-matched-control", + action="observe", + execution_duration_ms=0.0, + result=EvaluatorResult(matched=False, confidence=0.9), + ) + ], + ) + + # When: a caller evaluates a step through the public HTTP endpoint. + response = client.post( + "/api/v1/evaluation", + json={ + "agent_name": "duration-contract-agent", + "stage": "pre", + "step": {"type": "llm", "name": "answer", "input": "question"}, + }, + ) + + # Then: each category retains its own top-level duration through serialization. + assert response.status_code == 200 + payload = response.json() + assert payload["matches"][0]["execution_duration_ms"] == 12.5 + assert payload["errors"][0]["execution_duration_ms"] == 3.25 + assert payload["non_matches"][0]["execution_duration_ms"] == 0.0 + assert payload["errors"][0]["result"]["error"] == SAFE_EVALUATOR_ERROR + assert payload["errors"][0]["result"]["message"] == SAFE_EVALUATOR_ERROR + assert "private evaluator detail" not in response.text + for category in ("matches", "errors", "non_matches"): + assert "execution_duration_ms" not in payload[category][0]["result"] + + +@pytest.mark.parametrize("has_duration_field", [False, True], ids=["omitted", "null"]) +def test_evaluation_response_preserves_unknown_duration_as_null( + client: TestClient, + engine_response: AsyncMock, + has_duration_field: bool, +) -> None: + # Given: an old result without timing, or an explicitly unknown timing. + # Control loading and engine processing are stubbed to avoid database/evaluator I/O. + match_payload: dict[str, object] = { + "control_id": 1, + "control_name": "untimed-control", + "action": "observe", + "result": {"matched": False, "confidence": 0.0}, + } + if has_duration_field: + match_payload["execution_duration_ms"] = None + engine_response.return_value = EvaluationResponse( + is_safe=True, + confidence=1.0, + non_matches=[ControlMatch.model_validate(match_payload)], + ) + + # When: a caller evaluates a step through the public HTTP endpoint. + response = client.post( + "/api/v1/evaluation", + json={ + "agent_name": "duration-contract-agent", + "stage": "pre", + "step": {"type": "llm", "name": "answer", "input": "question"}, + }, + ) + + # Then: unknown timing remains null rather than becoming a measured zero. + assert response.status_code == 200 + assert response.json()["non_matches"][0]["execution_duration_ms"] is None From a418507fa79851f6b431f00c59553dcd2ca9cf01 Mon Sep 17 00:00:00 2001 From: Joshua Jeon Date: Thu, 8 Oct 2026 11:06:30 -0700 Subject: [PATCH 2/3] feat: add SDK-local control timing example --- examples/README.md | 1 + examples/sdk_local_timing/README.md | 31 ++++ examples/sdk_local_timing/demo.py | 208 +++++++++++++++++++++++++ examples/sdk_local_timing/test_demo.py | 40 +++++ 4 files changed, 280 insertions(+) create mode 100644 examples/sdk_local_timing/README.md create mode 100644 examples/sdk_local_timing/demo.py create mode 100644 examples/sdk_local_timing/test_demo.py diff --git a/examples/README.md b/examples/README.md index f780be53c..535ce1b18 100644 --- a/examples/README.md +++ b/examples/README.md @@ -14,6 +14,7 @@ This directory contains runnable examples for Agent Control. Each example has it | Customer Support Agent | Enterprise scenario with PII protection, prompt-injection defense, and multiple tools. | https://docs.agentcontrol.dev/examples/customer-support | | DeepEval | Build a custom evaluator using DeepEval GEval metrics. | https://docs.agentcontrol.dev/examples/deepeval | | Galileo Luna Direct | Direct Luna scorer invoke evaluation (`scorer_id` required; `scorer_label`/`scorer_version_id` optional) with a composite Agent Control condition. | `examples/galileo_luna/` | +| SDK-local timing | Compare per-control duration with whole SDK call time using synthetic local controls and no network. | `examples/sdk_local_timing/` | | LangChain SQL Agent | Protect a SQL agent from dangerous queries with server-side controls. | https://docs.agentcontrol.dev/examples/langchain-sql | | Steer Action Demo | Banking transfer agent showcasing observe, deny, and steer actions. | https://docs.agentcontrol.dev/examples/steer-action-demo | | Target Context | Bind controls to opaque external targets (e.g. `env=prod`) and let the SDK pin one target per session. | https://docs.agentcontrol.dev/examples/target-context | diff --git a/examples/sdk_local_timing/README.md b/examples/sdk_local_timing/README.md new file mode 100644 index 000000000..c75f3d451 --- /dev/null +++ b/examples/sdk_local_timing/README.md @@ -0,0 +1,31 @@ +# SDK-local control timing example + +This example runs five synthetic controls through the Python SDK's real +`check_evaluation_with_local` path. It covers list allow/block, regex steer, +SQL block, and JSON validation. Each case uses one control, so its result and +duration have a clear owner. + +PR #283 measures the full control condition evaluation. It includes evaluator +work and any wait for an evaluator slot. It excludes control loading and SDK +call overhead. The separate `sdk_call` distribution includes those costs. +These values are not isolated evaluator-call timings. + +From the repository root: + +```bash +make sync +uv run --package agent-control-sdk python examples/sdk_local_timing/demo.py --iterations 100 --warmup 5 +``` + +The default output is a small per-control p50/p95 table. Pass `--json` for +counts, min, p50, p95 and max values in a machine-readable report. The +`control` number comes from `ControlMatch.execution_duration_ms`; `sdk_call` +is measured around the SDK call by this example. + +The example disables SDK observability for its run and restores the previous +settings afterward. An HTTP mock rejects all outbound requests. No account, +API key, server, Luna scorer, or GPU is needed. The warmup runs are excluded +from the reported nearest-rank p50 and p95 values. + +This is a sequential local demonstration. It does not establish concurrent +SDK capacity or server/Luna3 throughput. Those require a separate load test. diff --git a/examples/sdk_local_timing/demo.py b/examples/sdk_local_timing/demo.py new file mode 100644 index 000000000..482406731 --- /dev/null +++ b/examples/sdk_local_timing/demo.py @@ -0,0 +1,208 @@ +"""Compare SDK-local per-control time with the full SDK call, without network I/O.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import math +import time +from dataclasses import dataclass +from typing import Any, Literal + +import httpx +from agent_control import AgentControlClient +from agent_control.evaluation import check_evaluation_with_local +from agent_control.settings import configure_settings, get_settings +from agent_control_models import Step + + +@dataclass(frozen=True) +class Scenario: + name: str + evaluator: str + config: dict[str, Any] + selector: str + decision: Literal["deny", "steer", "observe"] + step_input: dict[str, Any] + expected_bucket: Literal["matches", "non_matches"] + + +LIST_CONFIG = { + "values": ["delete_customer"], + "logic": "any", + "match_on": "match", + "match_mode": "exact", + "case_sensitive": False, +} + +SCENARIOS = ( + Scenario( + "list-allow", + "list", + LIST_CONFIG, + "input.action", + "deny", + {"action": "read_customer"}, + "non_matches", + ), + Scenario( + "list-block", + "list", + LIST_CONFIG, + "input.action", + "deny", + {"action": "delete_customer"}, + "matches", + ), + Scenario( + "regex-steer", + "regex", + {"pattern": "transfer"}, + "input.action", + "steer", + {"action": "transfer"}, + "matches", + ), + Scenario( + "sql-block", + "sql", + {"blocked_operations": ["DROP"]}, + "input.sql", + "deny", + {"sql": "DROP TABLE customers"}, + "matches", + ), + Scenario( + "json-valid", + "json", + {"required_fields": ["customer_id"]}, + "input.payload", + "observe", + {"payload": '{"customer_id":"synthetic-1"}'}, + "non_matches", + ), +) + + +def control_payload(scenario: Scenario, control_id: int) -> dict[str, Any]: + return { + "id": control_id, + "name": scenario.name, + "control": { + "enabled": True, + "execution": "sdk", + "scope": {"step_types": ["tool"], "stages": ["pre"], "step_names": ["timing_probe"]}, + "condition": { + "selector": {"path": scenario.selector}, + "evaluator": {"name": scenario.evaluator, "config": scenario.config}, + }, + "action": {"decision": scenario.decision}, + }, + } + + +def distribution(values: list[float]) -> dict[str, float | int]: + """Summarize a nonempty sample using nearest-rank percentiles.""" + ordered = sorted(values) + + def rank(fraction: float) -> float: + return ordered[max(0, math.ceil(fraction * len(ordered)) - 1)] + + return { + "count": len(ordered), + "min_ms": ordered[0], + "p50_ms": rank(0.50), + "p95_ms": rank(0.95), + "max_ms": ordered[-1], + } + + +def reject_network(request: httpx.Request) -> httpx.Response: + raise AssertionError(f"SDK-local demo attempted HTTP to {request.url.host}") + + +def format_table(report: dict[str, Any]) -> str: + """Render each local control's timing beside the enclosing SDK call.""" + rows = [ + f"iterations={report['iterations']} warmup={report['warmup']}", + "scenario outcome control p50/p95 (ms) SDK call p50/p95 (ms)", + ] + for name, result in report["scenarios"].items(): + control = result["control"] + sdk_call = result["sdk_call"] + rows.append( + f"{name:<13} {result['outcome']:<12} " + f"{control['p50_ms']:>8.3f}/{control['p95_ms']:>8.3f} " + f"{sdk_call['p50_ms']:>8.3f}/{sdk_call['p95_ms']:>8.3f}" + ) + return "\n".join(rows) + + +async def run(*, iterations: int, warmup: int) -> dict[str, Any]: + """Run each synthetic control locally and validate its measured result.""" + if not 1 <= iterations <= 100_000 or not 0 <= warmup <= 10_000: + raise ValueError("iterations must be 1..100000 and warmup must be 0..10000") + + previous_settings = get_settings().model_dump() + configure_settings(observability_enabled=False) + try: + report: dict[str, Any] = {"iterations": iterations, "warmup": warmup, "scenarios": {}} + async with AgentControlClient( + base_url="https://unused.invalid", transport=httpx.MockTransport(reject_network) + ) as client: + for control_id, scenario in enumerate(SCENARIOS, start=1): + control = control_payload(scenario, control_id) + step = Step(type="tool", name="timing_probe", input=scenario.step_input) + control_times: list[float] = [] + sdk_times: list[float] = [] + for index in range(warmup + iterations): + started_at = time.perf_counter() + result = await check_evaluation_with_local( + client, "sdk-local-timing-demo", step, "pre", [control] + ) + sdk_ms = (time.perf_counter() - started_at) * 1000 + bucket = getattr(result, scenario.expected_bucket) or [] + other_buckets = ( + ["non_matches"] if scenario.expected_bucket == "matches" else ["matches"] + ) + ["errors"] + if len(bucket) != 1 or any(getattr(result, name) for name in other_buckets): + raise RuntimeError(f"{scenario.name}: unexpected control outcome") + if bucket[0].control_id != control_id: + raise RuntimeError(f"{scenario.name}: wrong control result") + expected_match = scenario.expected_bucket == "matches" + if ( + bucket[0].action != scenario.decision + or bucket[0].result.matched != expected_match + or result.is_safe != (not expected_match) + ): + raise RuntimeError(f"{scenario.name}: unexpected action or safety result") + duration_ms = bucket[0].execution_duration_ms + if duration_ms is None or not math.isfinite(duration_ms) or duration_ms < 0: + raise RuntimeError(f"{scenario.name}: missing or invalid control duration") + if index >= warmup: + control_times.append(duration_ms) + sdk_times.append(sdk_ms) + report["scenarios"][scenario.name] = { + "evaluator": scenario.evaluator, + "outcome": scenario.expected_bucket, + "control": distribution(control_times), + "sdk_call": distribution(sdk_times), + } + return report + finally: + configure_settings(**previous_settings) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--iterations", type=int, default=100) + parser.add_argument("--warmup", type=int, default=5) + parser.add_argument("--json", action="store_true", help="emit machine-readable samples summary") + args = parser.parse_args() + report = asyncio.run(run(iterations=args.iterations, warmup=args.warmup)) + print(json.dumps(report, indent=2) if args.json else format_table(report)) + + +if __name__ == "__main__": + main() diff --git a/examples/sdk_local_timing/test_demo.py b/examples/sdk_local_timing/test_demo.py new file mode 100644 index 000000000..b882423b6 --- /dev/null +++ b/examples/sdk_local_timing/test_demo.py @@ -0,0 +1,40 @@ +"""Behavior checks for the SDK-local timing example.""" + +import pytest + +from examples.sdk_local_timing.demo import distribution, format_table, run + + +def test_distribution_uses_nearest_rank() -> None: + # Given: a sample with a known upper-tail observation. + values = list(range(1, 21)) + + # When: the collector summarizes the sample. + summary = distribution(values) + + # Then: p95 is the nineteenth ordered value. + assert summary == {"count": 20, "min_ms": 1, "p50_ms": 10, "p95_ms": 19, "max_ms": 20} + + +@pytest.mark.asyncio +async def test_cases_run_locally_and_have_control_time() -> None: + # Given: an HTTP transport that rejects any outbound request. + + # When: each synthetic case runs through the SDK local path after one warmup. + report = await run(iterations=2, warmup=1) + + # Then: each case returns exactly one measured control result. + assert set(report["scenarios"]) == { + "list-allow", + "list-block", + "regex-steer", + "sql-block", + "json-valid", + } + for case in report["scenarios"].values(): + assert case["control"]["count"] == 2 + assert case["control"]["min_ms"] >= 0 + assert case["sdk_call"]["min_ms"] >= case["control"]["min_ms"] + table = format_table(report) + assert "control p50/p95 (ms)" in table + assert "regex-steer" in table From 5fc7444669d727c42878c5b236e043f137a76838 Mon Sep 17 00:00:00 2001 From: Joshua Jeon Date: Thu, 8 Oct 2026 11:09:56 -0700 Subject: [PATCH 3/3] feat: add bounded concurrent SDK-local smoke --- examples/sdk_local_timing/README.md | 10 ++++- examples/sdk_local_timing/demo.py | 60 ++++++++++++++++++++------ examples/sdk_local_timing/test_demo.py | 3 +- 3 files changed, 56 insertions(+), 17 deletions(-) diff --git a/examples/sdk_local_timing/README.md b/examples/sdk_local_timing/README.md index c75f3d451..a91b6589b 100644 --- a/examples/sdk_local_timing/README.md +++ b/examples/sdk_local_timing/README.md @@ -22,10 +22,16 @@ counts, min, p50, p95 and max values in a machine-readable report. The `control` number comes from `ControlMatch.execution_duration_ms`; `sdk_call` is measured around the SDK call by this example. +For a bounded SDK-local throughput smoke, add `--concurrency 8 --iterations 1000`. +The table then includes observed SDK calls per second for each independent +scenario. This uses one Python process and one event loop without a target +request rate; it is not a production capacity number. + The example disables SDK observability for its run and restores the previous settings afterward. An HTTP mock rejects all outbound requests. No account, API key, server, Luna scorer, or GPU is needed. The warmup runs are excluded from the reported nearest-rank p50 and p95 values. -This is a sequential local demonstration. It does not establish concurrent -SDK capacity or server/Luna3 throughput. Those require a separate load test. +This example does not establish SDK capacity or server/Luna3 throughput. Those +require a configured workload, stack, acceptance thresholds and a separate +load test. diff --git a/examples/sdk_local_timing/demo.py b/examples/sdk_local_timing/demo.py index 482406731..4cce02140 100644 --- a/examples/sdk_local_timing/demo.py +++ b/examples/sdk_local_timing/demo.py @@ -125,8 +125,9 @@ def reject_network(request: httpx.Request) -> httpx.Response: def format_table(report: dict[str, Any]) -> str: """Render each local control's timing beside the enclosing SDK call.""" rows = [ - f"iterations={report['iterations']} warmup={report['warmup']}", - "scenario outcome control p50/p95 (ms) SDK call p50/p95 (ms)", + f"iterations={report['iterations']} warmup={report['warmup']} " + f"concurrency={report['concurrency']}", + "scenario outcome control p50/p95 (ms) SDK call p50/p95 (ms) calls/s", ] for name, result in report["scenarios"].items(): control = result["control"] @@ -134,29 +135,34 @@ def format_table(report: dict[str, Any]) -> str: rows.append( f"{name:<13} {result['outcome']:<12} " f"{control['p50_ms']:>8.3f}/{control['p95_ms']:>8.3f} " - f"{sdk_call['p50_ms']:>8.3f}/{sdk_call['p95_ms']:>8.3f}" + f"{sdk_call['p50_ms']:>8.3f}/{sdk_call['p95_ms']:>8.3f} " + f"{result['observed_calls_per_sec']:>8.1f}" ) return "\n".join(rows) -async def run(*, iterations: int, warmup: int) -> dict[str, Any]: +async def run(*, iterations: int, warmup: int, concurrency: int = 1) -> dict[str, Any]: """Run each synthetic control locally and validate its measured result.""" - if not 1 <= iterations <= 100_000 or not 0 <= warmup <= 10_000: - raise ValueError("iterations must be 1..100000 and warmup must be 0..10000") + if not (1 <= iterations <= 100_000 and 0 <= warmup <= 10_000 and 1 <= concurrency <= 64): + raise ValueError("iterations must be 1..100000, warmup 0..10000, and concurrency 1..64") previous_settings = get_settings().model_dump() configure_settings(observability_enabled=False) try: - report: dict[str, Any] = {"iterations": iterations, "warmup": warmup, "scenarios": {}} + report: dict[str, Any] = { + "iterations": iterations, + "warmup": warmup, + "concurrency": concurrency, + "scenarios": {}, + } async with AgentControlClient( base_url="https://unused.invalid", transport=httpx.MockTransport(reject_network) ) as client: for control_id, scenario in enumerate(SCENARIOS, start=1): control = control_payload(scenario, control_id) step = Step(type="tool", name="timing_probe", input=scenario.step_input) - control_times: list[float] = [] - sdk_times: list[float] = [] - for index in range(warmup + iterations): + + async def evaluate_once() -> tuple[float, float]: started_at = time.perf_counter() result = await check_evaluation_with_local( client, "sdk-local-timing-demo", step, "pre", [control] @@ -180,14 +186,37 @@ async def run(*, iterations: int, warmup: int) -> dict[str, Any]: duration_ms = bucket[0].execution_duration_ms if duration_ms is None or not math.isfinite(duration_ms) or duration_ms < 0: raise RuntimeError(f"{scenario.name}: missing or invalid control duration") - if index >= warmup: - control_times.append(duration_ms) - sdk_times.append(sdk_ms) + return duration_ms, sdk_ms + + for _ in range(warmup): + await evaluate_once() + + worker_count = min(concurrency, iterations) + + async def worker(count: int) -> tuple[list[float], list[float]]: + control_samples: list[float] = [] + sdk_samples: list[float] = [] + for _ in range(count): + duration_ms, sdk_ms = await evaluate_once() + control_samples.append(duration_ms) + sdk_samples.append(sdk_ms) + return control_samples, sdk_samples + + counts = [ + iterations // worker_count + (worker_index < iterations % worker_count) + for worker_index in range(worker_count) + ] + run_started_at = time.perf_counter() + worker_results = await asyncio.gather(*(worker(count) for count in counts)) + elapsed_seconds = time.perf_counter() - run_started_at + control_times = [sample for controls, _ in worker_results for sample in controls] + sdk_times = [sample for _, calls in worker_results for sample in calls] report["scenarios"][scenario.name] = { "evaluator": scenario.evaluator, "outcome": scenario.expected_bucket, "control": distribution(control_times), "sdk_call": distribution(sdk_times), + "observed_calls_per_sec": iterations / elapsed_seconds, } return report finally: @@ -198,9 +227,12 @@ def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--iterations", type=int, default=100) parser.add_argument("--warmup", type=int, default=5) + parser.add_argument("--concurrency", type=int, default=1) parser.add_argument("--json", action="store_true", help="emit machine-readable samples summary") args = parser.parse_args() - report = asyncio.run(run(iterations=args.iterations, warmup=args.warmup)) + report = asyncio.run( + run(iterations=args.iterations, warmup=args.warmup, concurrency=args.concurrency) + ) print(json.dumps(report, indent=2) if args.json else format_table(report)) diff --git a/examples/sdk_local_timing/test_demo.py b/examples/sdk_local_timing/test_demo.py index b882423b6..b74631bea 100644 --- a/examples/sdk_local_timing/test_demo.py +++ b/examples/sdk_local_timing/test_demo.py @@ -21,7 +21,7 @@ async def test_cases_run_locally_and_have_control_time() -> None: # Given: an HTTP transport that rejects any outbound request. # When: each synthetic case runs through the SDK local path after one warmup. - report = await run(iterations=2, warmup=1) + report = await run(iterations=2, warmup=1, concurrency=2) # Then: each case returns exactly one measured control result. assert set(report["scenarios"]) == { @@ -35,6 +35,7 @@ async def test_cases_run_locally_and_have_control_time() -> None: assert case["control"]["count"] == 2 assert case["control"]["min_ms"] >= 0 assert case["sdk_call"]["min_ms"] >= case["control"]["min_ms"] + assert case["observed_calls_per_sec"] > 0 table = format_table(report) assert "control p50/p95 (ms)" in table assert "regex-steer" in table